Skip to main content

max / makenotwork

16.4 KB · 460 lines History Blame Raw
1 //! SyncKit group management and group-scoped push/pull.
2 //!
3 //! A group is a shared, end-to-end-encrypted changelog. The admin mints a Group
4 //! Content Key (GCK) client-side and seals it to each member's identity public
5 //! key; the server stores membership and the opaque sealed grants and never sees
6 //! the GCK or any plaintext. Group push/pull are gated on membership; management
7 //! actions (add/remove member) are gated on being the group admin.
8 //!
9 //! Deferred to later slices (noted where they'd hook in): the paid-write gate and
10 //! SSE push notifications for groups (p2-billing), member public-key storage and
11 //! GCK rotation (p3). Design: wiki synckit-groups-design.
12
13 use axum::{
14 Json,
15 extract::{Path, State},
16 http::StatusCode,
17 response::{IntoResponse, Response},
18 };
19 use serde_json::json;
20 use sqlx::PgPool;
21
22 use crate::{
23 constants,
24 db::{self, DbSyncGroup, SyncGroupId, UserId},
25 error::{AppError, Result},
26 synckit_auth::SyncUser,
27 validation,
28 };
29
30 use super::{
31 AddMemberRequest, CreateGroupRequest, GroupGrantResponse, GroupMemberPubkey,
32 GroupMemberResponse, GroupResponse, PullChangeEntry, PullRequest, PullResponse, PushRequest,
33 PushResponse,
34 };
35
36 /// Fetch a group scoped to the caller's app, or 404. Guards every group handler
37 /// against cross-app id guessing before any membership check.
38 async fn require_group(
39 db: &PgPool,
40 app_id: db::SyncAppId,
41 group_id: SyncGroupId,
42 ) -> Result<DbSyncGroup> {
43 db::synckit::get_group(db, app_id, group_id)
44 .await?
45 .ok_or(AppError::NotFound)
46 }
47
48 /// Reject a caller who is not a member of the group (403). The gate for
49 /// group-scoped reads and writes.
50 async fn require_member(db: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<()> {
51 if db::synckit::is_group_member(db, group_id, user_id).await? {
52 Ok(())
53 } else {
54 Err(AppError::Forbidden)
55 }
56 }
57
58 /// Reject a caller who is not the group admin (403). The gate for membership
59 /// management.
60 async fn require_admin(db: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<()> {
61 if db::synckit::is_group_admin(db, group_id, user_id).await? {
62 Ok(())
63 } else {
64 Err(AppError::Forbidden)
65 }
66 }
67
68 /// Create a group. The caller becomes its admin and first member, carrying the
69 /// GCK they sealed to their own identity key.
70 #[utoipa::path(post, path = "/api/v1/sync/groups", tag = "SyncKit",
71 request_body = CreateGroupRequest,
72 responses((status = 200, description = "Created group", body = GroupResponse)),
73 security(("bearer" = [])),
74 )]
75 #[tracing::instrument(skip_all, name = "synckit::create_group")]
76 pub(super) async fn create_group(
77 State(db): State<PgPool>,
78 sync_user: SyncUser,
79 Json(req): Json<CreateGroupRequest>,
80 ) -> Result<impl IntoResponse> {
81 validation::validate_sync_group_name(&req.name)?;
82 if req.admin_sealed_gck.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES
83 || req.admin_pubkey.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES
84 {
85 return Err(AppError::BadRequest(
86 "Sealed key exceeds size limit".to_string(),
87 ));
88 }
89
90 let group = db::synckit::create_group(
91 &db,
92 req.id,
93 sync_user.app_id,
94 sync_user.user_id,
95 &req.name,
96 &req.admin_sealed_gck,
97 &req.admin_pubkey,
98 )
99 .await?;
100
101 Ok(Json(GroupResponse::from(group)))
102 }
103
104 /// List the groups the caller belongs to within this app.
105 #[utoipa::path(get, path = "/api/v1/sync/groups", tag = "SyncKit",
106 responses((status = 200, description = "Groups the user belongs to", body = Vec<GroupResponse>)),
107 security(("bearer" = [])),
108 )]
109 #[tracing::instrument(skip_all, name = "synckit::list_groups")]
110 pub(super) async fn list_groups(
111 State(db): State<PgPool>,
112 sync_user: SyncUser,
113 ) -> Result<impl IntoResponse> {
114 let groups =
115 db::synckit::list_groups_for_user(&db, sync_user.app_id, sync_user.user_id).await?;
116 let response: Vec<GroupResponse> = groups.into_iter().map(GroupResponse::from).collect();
117 Ok(Json(response))
118 }
119
120 /// Add a member to a group (or replace their grant). Admin only.
121 ///
122 /// The admin resolves the member out of band, seals the current GCK to that
123 /// member's public key, and posts `{member_email, sealed_gck}`. The server maps
124 /// the email to a verified account and stores the opaque grant at the group's
125 /// current GCK generation.
126 #[utoipa::path(post, path = "/api/v1/sync/groups/{id}/members", tag = "SyncKit",
127 params(("id" = String, Path, description = "Group ID")),
128 request_body = AddMemberRequest,
129 responses((status = 204, description = "Member added"), (status = 403, description = "Not the group admin")),
130 security(("bearer" = [])),
131 )]
132 #[tracing::instrument(skip_all, name = "synckit::add_group_member")]
133 pub(super) async fn add_member(
134 State(db): State<PgPool>,
135 sync_user: SyncUser,
136 Path(group_id): Path<SyncGroupId>,
137 Json(req): Json<AddMemberRequest>,
138 ) -> Result<impl IntoResponse> {
139 let group = require_group(&db, sync_user.app_id, group_id).await?;
140 require_admin(&db, group_id, sync_user.user_id).await?;
141
142 if req.sealed_gck.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES
143 || req.member_pubkey.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES
144 {
145 return Err(AppError::BadRequest(
146 "Sealed key exceeds size limit".to_string(),
147 ));
148 }
149 let role = req.role.as_deref().unwrap_or("member");
150 if role != "member" && role != "admin" {
151 return Err(AppError::BadRequest(
152 "role must be 'member' or 'admin'".to_string(),
153 ));
154 }
155
156 let email = db::Email::new(&req.member_email)
157 .map_err(|_| AppError::BadRequest("Invalid email address".to_string()))?;
158 let member_id = db::users::get_verified_user_id_by_email(&db, &email)
159 .await?
160 .ok_or_else(|| AppError::BadRequest("No verified account for that email".to_string()))?;
161
162 // The grant the admin sends is sealed under the group's current GCK, so it is
163 // stored at that generation, along with the member's public key (for re-seal
164 // on a later rotation).
165 db::synckit::add_or_update_member(
166 &db,
167 group_id,
168 member_id,
169 role,
170 &req.sealed_gck,
171 group.gck_version,
172 &req.member_pubkey,
173 )
174 .await?;
175
176 Ok(StatusCode::NO_CONTENT)
177 }
178
179 /// List every member's identity public key. Admin only: the admin re-seals a
180 /// rotated GCK to each of these on member removal.
181 #[utoipa::path(get, path = "/api/v1/sync/groups/{id}/pubkeys", tag = "SyncKit",
182 params(("id" = String, Path, description = "Group ID")),
183 responses((status = 200, description = "Member public keys", body = Vec<GroupMemberPubkey>)),
184 security(("bearer" = [])),
185 )]
186 #[tracing::instrument(skip_all, name = "synckit::list_group_pubkeys")]
187 pub(super) async fn list_pubkeys(
188 State(db): State<PgPool>,
189 sync_user: SyncUser,
190 Path(group_id): Path<SyncGroupId>,
191 ) -> Result<impl IntoResponse> {
192 require_group(&db, sync_user.app_id, group_id).await?;
193 require_admin(&db, group_id, sync_user.user_id).await?;
194
195 let pubkeys = db::synckit::list_member_pubkeys(&db, group_id).await?;
196 let response: Vec<GroupMemberPubkey> = pubkeys
197 .into_iter()
198 .map(|(user_id, pubkey)| GroupMemberPubkey { user_id, pubkey })
199 .collect();
200 Ok(Json(response))
201 }
202
203 /// Remove a member from a group. Admin only.
204 ///
205 /// This drops the member from future group writes. Forward secrecy for writes
206 /// after removal comes from the admin then rotating the GCK (p3); data the member
207 /// already pulled is already in their hands.
208 #[utoipa::path(delete, path = "/api/v1/sync/groups/{id}/members/{user_id}", tag = "SyncKit",
209 params(
210 ("id" = String, Path, description = "Group ID"),
211 ("user_id" = String, Path, description = "Member user ID"),
212 ),
213 responses((status = 204, description = "Member removed"), (status = 404, description = "Not a member")),
214 security(("bearer" = [])),
215 )]
216 #[tracing::instrument(skip_all, name = "synckit::remove_group_member")]
217 pub(super) async fn remove_member(
218 State(db): State<PgPool>,
219 sync_user: SyncUser,
220 Path((group_id, member_id)): Path<(SyncGroupId, UserId)>,
221 ) -> Result<impl IntoResponse> {
222 let group = require_group(&db, sync_user.app_id, group_id).await?;
223 require_admin(&db, group_id, sync_user.user_id).await?;
224
225 // The admin cannot remove themselves; that would orphan the group. Deleting a
226 // group is a separate action (not yet exposed).
227 if member_id == group.admin_user_id {
228 return Err(AppError::BadRequest(
229 "The group admin cannot be removed".to_string(),
230 ));
231 }
232
233 if !db::synckit::remove_member(&db, group_id, member_id).await? {
234 return Err(AppError::NotFound);
235 }
236
237 Ok(StatusCode::NO_CONTENT)
238 }
239
240 /// List a group's members (id, role, joined-at). Members only. Grants are not
241 /// included; each member fetches only their own via `/grant`.
242 #[utoipa::path(get, path = "/api/v1/sync/groups/{id}/members", tag = "SyncKit",
243 params(("id" = String, Path, description = "Group ID")),
244 responses((status = 200, description = "Group members", body = Vec<GroupMemberResponse>)),
245 security(("bearer" = [])),
246 )]
247 #[tracing::instrument(skip_all, name = "synckit::list_group_members")]
248 pub(super) async fn list_members(
249 State(db): State<PgPool>,
250 sync_user: SyncUser,
251 Path(group_id): Path<SyncGroupId>,
252 ) -> Result<impl IntoResponse> {
253 require_group(&db, sync_user.app_id, group_id).await?;
254 require_member(&db, group_id, sync_user.user_id).await?;
255
256 let members = db::synckit::list_members(&db, group_id).await?;
257 let response: Vec<GroupMemberResponse> = members
258 .into_iter()
259 .map(|m| GroupMemberResponse {
260 user_id: m.user_id,
261 email: m.email,
262 role: m.role,
263 added_at: m.added_at,
264 })
265 .collect();
266 Ok(Json(response))
267 }
268
269 /// Fetch the caller's own sealed GCK grant for a group, so their device can open
270 /// the GCK and read the group changelog. Members only.
271 #[utoipa::path(get, path = "/api/v1/sync/groups/{id}/grant", tag = "SyncKit",
272 params(("id" = String, Path, description = "Group ID")),
273 responses(
274 (status = 200, description = "The caller's sealed grant", body = GroupGrantResponse),
275 (status = 403, description = "Not a member"),
276 ),
277 security(("bearer" = [])),
278 )]
279 #[tracing::instrument(skip_all, name = "synckit::get_group_grant")]
280 pub(super) async fn get_grant(
281 State(db): State<PgPool>,
282 sync_user: SyncUser,
283 Path(group_id): Path<SyncGroupId>,
284 ) -> Result<impl IntoResponse> {
285 require_group(&db, sync_user.app_id, group_id).await?;
286
287 let (sealed_gck, gck_version) = db::synckit::get_member_grant(&db, group_id, sync_user.user_id)
288 .await?
289 .ok_or(AppError::Forbidden)?;
290
291 Ok(Json(GroupGrantResponse {
292 sealed_gck,
293 gck_version,
294 }))
295 }
296
297 /// Push encrypted changes to a group's shared changelog. Members only.
298 #[utoipa::path(post, path = "/api/v1/sync/groups/{id}/push", tag = "SyncKit",
299 params(("id" = String, Path, description = "Group ID")),
300 request_body = PushRequest,
301 responses((status = 200, description = "New cursor position", body = PushResponse)),
302 security(("bearer" = [])),
303 )]
304 #[tracing::instrument(skip_all, name = "synckit::group_push", fields(group_id))]
305 pub(super) async fn group_push(
306 State(db): State<PgPool>,
307 sync_user: SyncUser,
308 Path(group_id): Path<SyncGroupId>,
309 Json(req): Json<PushRequest>,
310 ) -> Result<Response> {
311 let group = require_group(&db, sync_user.app_id, group_id).await?;
312 require_member(&db, group_id, sync_user.user_id).await?;
313
314 // Group writes bill to the admin's slot (Groups billing decision): the paid
315 // gate is checked against the *admin's* entitlement, not the pushing member's,
316 // so a member with no subscription of their own can still contribute to a
317 // group whose admin pays. Reads (group_pull) stay open, as personal pull does.
318 if !db::synckit::internal_write_allowed(&db, sync_user.app_id, group.admin_user_id).await? {
319 return Ok((
320 StatusCode::PAYMENT_REQUIRED,
321 Json(json!({ "reason": "no_subscription" })),
322 )
323 .into_response());
324 }
325 // NOTE: SSE push notification to group members is deferred to p3; members'
326 // devices pick up group changes on their next scheduler tick meanwhile.
327
328 if req.changes.is_empty() {
329 return Err(AppError::BadRequest("No changes provided".to_string()));
330 }
331 if req.changes.len() > constants::SYNCKIT_PUSH_MAX_CHANGES {
332 return Err(AppError::BadRequest(format!(
333 "Maximum {} changes per push",
334 constants::SYNCKIT_PUSH_MAX_CHANGES
335 )));
336 }
337 for change in &req.changes {
338 validation::validate_sync_table_name(&change.table)?;
339 validation::validate_sync_row_id(&change.row_id)?;
340 if change.op == db::SyncOperation::Delete && change.data.is_some() {
341 return Err(AppError::BadRequest(
342 "DELETE operations should not include data".to_string(),
343 ));
344 }
345 }
346
347 // The pushing device must belong to the pushing user (membership is a
348 // separate, group-level check above).
349 if !db::synckit::sync_device_belongs(&db, req.device_id, sync_user.app_id, sync_user.user_id)
350 .await?
351 {
352 return Err(AppError::BadRequest("Unknown device".to_string()));
353 }
354 db::synckit::touch_sync_device(&db, req.device_id).await?;
355
356 let changes: Vec<_> = req
357 .changes
358 .iter()
359 .map(|c| {
360 (
361 c.table.clone(),
362 c.op.to_string(),
363 c.row_id.clone(),
364 c.timestamp,
365 c.data.clone(),
366 )
367 })
368 .collect();
369
370 let cursor = db::synckit::push_group_changes(
371 &db,
372 sync_user.app_id,
373 group_id,
374 sync_user.user_id,
375 req.device_id,
376 req.batch_id,
377 &changes,
378 )
379 .await?;
380
381 Ok(Json(PushResponse { cursor }).into_response())
382 }
383
384 /// Pull a group's changes after a cursor. Members only.
385 #[utoipa::path(post, path = "/api/v1/sync/groups/{id}/pull", tag = "SyncKit",
386 params(("id" = String, Path, description = "Group ID")),
387 request_body = PullRequest,
388 responses((status = 200, description = "Changes since cursor", body = PullResponse)),
389 security(("bearer" = [])),
390 )]
391 #[tracing::instrument(skip_all, name = "synckit::group_pull", fields(group_id))]
392 pub(super) async fn group_pull(
393 State(db): State<PgPool>,
394 sync_user: SyncUser,
395 Path(group_id): Path<SyncGroupId>,
396 Json(req): Json<PullRequest>,
397 ) -> Result<impl IntoResponse> {
398 require_group(&db, sync_user.app_id, group_id).await?;
399 require_member(&db, group_id, sync_user.user_id).await?;
400
401 if !db::synckit::sync_device_belongs(&db, req.device_id, sync_user.app_id, sync_user.user_id)
402 .await?
403 {
404 return Err(AppError::BadRequest("Unknown device".to_string()));
405 }
406
407 if let Some(ref tables) = req.tables {
408 if tables.len() > 50 {
409 return Err(AppError::BadRequest(
410 "Maximum 50 table names per filter".to_string(),
411 ));
412 }
413 for table in tables {
414 validation::validate_sync_table_name(table)?;
415 }
416 }
417
418 let page_size = constants::SYNCKIT_PULL_PAGE_SIZE;
419 let entries = db::synckit::pull_group_changes_filtered(
420 &db,
421 sync_user.app_id,
422 group_id,
423 req.cursor,
424 page_size,
425 req.tables.as_deref(),
426 req.since,
427 )
428 .await?;
429
430 let has_more = entries.len() as i64 == page_size;
431 let new_cursor = entries.last().map_or(req.cursor, |e| e.seq);
432
433 // Touch the device for activity, but do NOT advance the per-device personal
434 // compaction cursor here; that cursor governs personal-changelog retention
435 // and must not be moved by a group pull. Group changelog retention is a
436 // separate concern (future work).
437 db::synckit::touch_sync_device(&db, req.device_id).await?;
438
439 let changes: Vec<PullChangeEntry> = entries
440 .into_iter()
441 .map(|e| PullChangeEntry {
442 seq: e.seq,
443 device_id: e.device_id,
444 table: e.table_name,
445 op: e.operation.to_string(),
446 row_id: e.row_id,
447 timestamp: e.client_timestamp,
448 data: e.data,
449 // Group entries key off the group's GCK, not a per-user key_id.
450 key_id: None,
451 })
452 .collect();
453
454 Ok(Json(PullResponse {
455 changes,
456 cursor: new_cursor,
457 has_more,
458 }))
459 }
460