Skip to main content

max / makenotwork

synckit-server: group-scoped changelog, routes, and its own table (Groups phase 2, p2-changelog) Group push/pull plus the group management REST surface, gated on membership (reads) and admin (member management). JWT-authed on the same dual-rate-limit tier as personal sync, dual /api/sync + /api/v1/sync paths. The group changelog lives in its own sync_group_log table (migration 172), superseding the sync_log.group_id column from 171. A shared column would have forced every personal-scope query (pull, key rotation, status, cleanup) to remember AND group_id IS NULL or silently leak GCK-encrypted group rows into a user's personal sync and let personal rotation re-encrypt them under the wrong key. A separate table makes that isolation structural, not a per-query discipline -- personal sync_log stays untouched. Same principle as the crypto layer's AeadContext. - db::synckit::log: push_group_changes / pull_group_changes_filtered on sync_group_log; DbSyncGroupLogEntry model (no key_id -- group entries key off the group's gck_version). - routes/synckit/groups: create_group, list_groups, add_member/remove_member/ list_members (admin/member gated, email -> verified user), get_grant (serves the caller's own sealed grant), group_push / group_pull. - validation::validate_sync_group_name; group request/response types + router. - migration 171 amended to IF NOT EXISTS (re-run-safety hygiene). - tests/workflows/db_synckit_groups: 9 DB-layer tests incl. the isolation invariant that group entries never appear in personal pull/status (and vice versa). All green against local postgres. Deferred (marked in-code): paid-write gate + SSE notify for group push (p2-billing); member-pubkey storage + GCK rotation routes and group changelog retention (p3). Design: wiki synckit-groups-design. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-23 18:39 UTC
Commit: 730ec79f2c7c27153b6e3ce3fce3946e7b054044
Parent: d760011
9 files changed, +1245 insertions, -7 deletions
M server/migrations/171_synckit_groups.sql +5 -5
@@ -7,7 +7,7 @@
7 7
8 8 -- A group: a shared changelog owned by one admin, whose members each hold a
9 9 -- grant of the same GCK sealed to their identity public key.
10 - CREATE TABLE sync_groups (
10 + CREATE TABLE IF NOT EXISTS sync_groups (
11 11 id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
12 12 app_id UUID NOT NULL REFERENCES sync_apps(id) ON DELETE CASCADE,
13 13 -- The admin who mints the GCK and manages membership. The group's changelog
@@ -21,10 +21,10 @@ CREATE TABLE sync_groups (
21 21 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
22 22 );
23 23
24 - CREATE INDEX idx_sync_groups_app_admin ON sync_groups (app_id, admin_user_id);
24 + CREATE INDEX IF NOT EXISTS idx_sync_groups_app_admin ON sync_groups (app_id, admin_user_id);
25 25
26 26 -- Membership + the per-member sealed GCK grant. One row per (group, user).
27 - CREATE TABLE sync_group_members (
27 + CREATE TABLE IF NOT EXISTS sync_group_members (
28 28 group_id UUID NOT NULL REFERENCES sync_groups(id) ON DELETE CASCADE,
29 29 user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
30 30 -- Reserved for the later per-key permission system. MVP is admin | member;
@@ -41,7 +41,7 @@ CREATE TABLE sync_group_members (
41 41 );
42 42
43 43 -- "Which groups am I in" is a per-user lookup on pull/list.
44 - CREATE INDEX idx_sync_group_members_user ON sync_group_members (user_id);
44 + CREATE INDEX IF NOT EXISTS idx_sync_group_members_user ON sync_group_members (user_id);
45 45
46 46 -- Group scope on the changelog. NULL = personal: the existing per-user semantics
47 47 -- are untouched, so no backfill is needed. A group entry is scoped by
@@ -51,4 +51,4 @@ ALTER TABLE sync_log ADD COLUMN group_id UUID REFERENCES sync_groups(id) ON DELE
51 51
52 52 -- Partial index for the group pull path; personal pulls keep using the existing
53 53 -- (app_id, user_id, seq) access pattern and never touch this index.
54 - CREATE INDEX idx_sync_log_group ON sync_log (app_id, group_id, seq) WHERE group_id IS NOT NULL;
54 + CREATE INDEX IF NOT EXISTS idx_sync_log_group ON sync_log (app_id, group_id, seq) WHERE group_id IS NOT NULL;
A server/migrations/172_synckit_group_log.sql +39
@@ -0,0 +1,39 @@
1 + -- SyncKit Groups: give the group changelog its own table.
2 + --
3 + -- Supersedes the sync_log.group_id column added in 171. Keeping group entries in
4 + -- sync_log meant every personal-scope query (pull, key rotation, status, cleanup)
5 + -- had to remember `AND group_id IS NULL` or silently leak GCK-encrypted group rows
6 + -- into a user's personal sync -- and let personal key rotation re-encrypt them with
7 + -- the wrong key. A dedicated table makes that isolation structural: sync_log stays
8 + -- purely personal and unchanged, and the group changelog evolves on its own.
9 + -- Design: wiki synckit-groups-design.
10 +
11 + DROP INDEX IF EXISTS idx_sync_log_group;
12 + ALTER TABLE sync_log DROP COLUMN IF EXISTS group_id;
13 +
14 + CREATE TABLE IF NOT EXISTS sync_group_log (
15 + seq BIGSERIAL PRIMARY KEY,
16 + app_id UUID NOT NULL REFERENCES sync_apps(id) ON DELETE CASCADE,
17 + group_id UUID NOT NULL REFERENCES sync_groups(id) ON DELETE CASCADE,
18 + -- The member who pushed this change (provenance/audit). The group, not the
19 + -- user, owns the entry, so pulls are scoped by group_id, not user_id.
20 + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
21 + device_id UUID NOT NULL REFERENCES sync_devices(id) ON DELETE CASCADE,
22 + -- Client-generated batch id for idempotent push (same role as sync_log).
23 + batch_id UUID,
24 + table_name VARCHAR(100) NOT NULL,
25 + operation VARCHAR(10) NOT NULL
26 + CHECK (operation IN ('INSERT', 'UPDATE', 'DELETE')),
27 + row_id VARCHAR(255) NOT NULL,
28 + client_timestamp TIMESTAMPTZ NOT NULL,
29 + -- Ciphertext sealed under the group's GCK (generation = sync_groups.gck_version).
30 + -- No key_id column: group entries key off the GCK, not the per-user
31 + -- sync_keys.key_id that personal rotation tracks.
32 + data JSONB,
33 + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
34 + );
35 +
36 + -- Group pull: (app_id, group_id, seq > cursor) ordered by seq.
37 + CREATE INDEX IF NOT EXISTS idx_sync_group_log_pull ON sync_group_log (app_id, group_id, seq);
38 + -- Group push idempotency: MAX(seq) for a (group, batch).
39 + CREATE INDEX IF NOT EXISTS idx_sync_group_log_batch ON sync_group_log (group_id, batch_id);
M server/src/db/models/synckit.rs +32
@@ -131,6 +131,38 @@ pub struct DbSyncGroupMember {
131 131 pub added_at: DateTime<Utc>,
132 132 }
133 133
134 + /// An entry in a group's append-only shared change log (`sync_group_log`).
135 + ///
136 + /// The group changelog is a separate table from the personal [`DbSyncLogEntry`]
137 + /// so personal-scope queries can never see group rows (which are sealed under the
138 + /// group's GCK, not the per-user key). There is deliberately no `key_id`: a group
139 + /// entry's key generation is the group's `gck_version`, not `sync_keys.key_id`.
140 + #[derive(Debug, Clone, FromRow, Serialize)]
141 + pub struct DbSyncGroupLogEntry {
142 + /// Server-assigned monotonic sequence number; the group pull cursor.
143 + pub seq: i64,
144 + /// Sync app this entry belongs to.
145 + pub app_id: SyncAppId,
146 + /// The group whose shared log this entry is in.
147 + pub group_id: SyncGroupId,
148 + /// The member who pushed this change (provenance).
149 + pub user_id: UserId,
150 + /// Device that originated this change.
151 + pub device_id: SyncDeviceId,
152 + /// Opaque table name from the client.
153 + pub table_name: String,
154 + /// Type of change (insert, update, or delete).
155 + pub operation: super::super::SyncOperation,
156 + /// Client-side row identifier (opaque to the server).
157 + pub row_id: String,
158 + /// Timestamp assigned by the client when the change was made.
159 + pub client_timestamp: DateTime<Utc>,
160 + /// Encrypted row data (sealed under the group GCK). Null for deletes.
161 + pub data: Option<serde_json::Value>,
162 + /// When the server received and recorded this entry.
163 + pub created_at: DateTime<Utc>,
164 + }
165 +
134 166 /// A blob uploaded to S3 via SyncKit, tracked for dedup and cleanup.
135 167 #[derive(Debug, Clone, FromRow, Serialize)]
136 168 pub struct DbSyncBlob {
M server/src/db/synckit/log.rs +135 -1
@@ -6,7 +6,7 @@ use serde_json::Value as JsonValue;
6 6 use sqlx::PgPool;
7 7
8 8 use crate::db::models::*;
9 - use crate::db::{SyncAppId, SyncDeviceId, UserId};
9 + use crate::db::{SyncAppId, SyncDeviceId, SyncGroupId, UserId};
10 10 use crate::error::Result;
11 11
12 12 // ── Sync Log ──
@@ -129,6 +129,140 @@ pub async fn push_sync_changes(
129 129 Ok(max_seq)
130 130 }
131 131
132 + /// Push a batch of changes to a **group's** shared changelog (`sync_group_log`).
133 + /// Returns the highest seq assigned. The group scope (not the pushing user) owns
134 + /// the entries, so every member's pull sees them.
135 + ///
136 + /// Mirrors [`push_sync_changes`] but writes the separate `sync_group_log` table
137 + /// and scopes idempotency by `(app_id, group_id, batch_id)`. Group entries carry
138 + /// no `key_id`: their key generation is the group's `gck_version`, not the
139 + /// per-user `sync_keys.key_id`. The dedicated table is what keeps group rows out
140 + /// of every personal-scope query with no `group_id IS NULL` guard to remember.
141 + /// Membership is enforced by the caller.
142 + #[allow(clippy::type_complexity)]
143 + #[tracing::instrument(skip_all)]
144 + pub async fn push_group_changes(
145 + pool: &PgPool,
146 + app_id: SyncAppId,
147 + group_id: SyncGroupId,
148 + user_id: UserId,
149 + device_id: SyncDeviceId,
150 + batch_id: uuid::Uuid,
151 + changes: &[(
152 + String,
153 + String,
154 + String,
155 + chrono::DateTime<chrono::Utc>,
156 + Option<JsonValue>,
157 + )],
158 + ) -> Result<i64> {
159 + if changes.is_empty() {
160 + return Ok(0);
161 + }
162 +
163 + let mut table_names: Vec<String> = Vec::with_capacity(changes.len());
164 + let mut operations: Vec<String> = Vec::with_capacity(changes.len());
165 + let mut row_ids: Vec<String> = Vec::with_capacity(changes.len());
166 + let mut client_timestamps: Vec<chrono::DateTime<chrono::Utc>> =
167 + Vec::with_capacity(changes.len());
168 + let mut data_values: Vec<JsonValue> = Vec::with_capacity(changes.len());
169 +
170 + for (table_name, operation, row_id, client_timestamp, data) in changes {
171 + table_names.push(table_name.clone());
172 + operations.push(operation.clone());
173 + row_ids.push(row_id.clone());
174 + client_timestamps.push(*client_timestamp);
175 + data_values.push(data.clone().unwrap_or(JsonValue::Null));
176 + }
177 +
178 + let mut tx = pool.begin().await?;
179 +
180 + // Serialize concurrent redeliveries of the same batch within this group, so
181 + // the MAX(seq) idempotency check below is race-free (same pattern as the
182 + // per-user push lock, keyed by group instead of user).
183 + sqlx::query(
184 + "SELECT pg_advisory_xact_lock(hashtextextended('synckit_group_push:' || $1::text || ':' || $2::text, 0))",
185 + )
186 + .bind(group_id)
187 + .bind(batch_id)
188 + .execute(&mut *tx)
189 + .await?;
190 +
191 + let existing: (Option<i64>,) = sqlx::query_as(
192 + "SELECT MAX(seq) FROM sync_group_log WHERE app_id = $1 AND group_id = $2 AND batch_id = $3",
193 + )
194 + .bind(app_id)
195 + .bind(group_id)
196 + .bind(batch_id)
197 + .fetch_one(&mut *tx)
198 + .await?;
199 +
200 + if let Some(max_seq) = existing.0 {
201 + tx.rollback().await.ok();
202 + return Ok(max_seq);
203 + }
204 +
205 + let seqs: Vec<i64> = sqlx::query_scalar(
206 + r#"
207 + INSERT INTO sync_group_log (app_id, user_id, device_id, group_id, batch_id, table_name, operation, row_id, client_timestamp, data)
208 + SELECT $1, $2, $3, $4, $5, t.*
209 + FROM UNNEST($6::text[], $7::text[], $8::text[], $9::timestamptz[], $10::jsonb[]) AS t
210 + RETURNING seq
211 + "#,
212 + )
213 + .bind(app_id)
214 + .bind(user_id)
215 + .bind(device_id)
216 + .bind(group_id)
217 + .bind(batch_id)
218 + .bind(&table_names)
219 + .bind(&operations)
220 + .bind(&row_ids)
221 + .bind(&client_timestamps)
222 + .bind(&data_values)
223 + .fetch_all(&mut *tx)
224 + .await?;
225 +
226 + tx.commit().await?;
227 +
228 + Ok(seqs.iter().copied().max().unwrap_or(0))
229 + }
230 +
231 + /// Pull a group's changes since a cursor from `sync_group_log`, with the same
232 + /// optional table/timestamp filters as [`pull_sync_changes_filtered`], scoped by
233 + /// `(app_id, group_id)`. Membership is enforced by the caller.
234 + #[tracing::instrument(skip_all)]
235 + pub async fn pull_group_changes_filtered(
236 + pool: &PgPool,
237 + app_id: SyncAppId,
238 + group_id: SyncGroupId,
239 + cursor: i64,
240 + limit: i64,
241 + tables: Option<&[String]>,
242 + since: Option<chrono::DateTime<chrono::Utc>>,
243 + ) -> Result<Vec<DbSyncGroupLogEntry>> {
244 + let entries = sqlx::query_as::<_, DbSyncGroupLogEntry>(
245 + r#"
246 + SELECT * FROM sync_group_log
247 + WHERE app_id = $1 AND group_id = $2 AND seq > $3
248 + AND ($5::text[] IS NULL OR table_name = ANY($5))
249 + AND ($6::timestamptz IS NULL OR client_timestamp >= $6)
250 + ORDER BY seq ASC
251 + LIMIT $4
252 + "#,
253 + )
254 + .bind(app_id)
255 + .bind(group_id)
256 + .bind(cursor)
257 + .bind(limit)
258 + .bind(tables)
259 + .bind(since)
260 + .fetch_all(pool)
261 + .await?;
262 +
263 + Ok(entries)
264 + }
265 +
132 266 /// Pull changes since a cursor for a user within an app.
133 267 ///
134 268 /// Prefer `pull_sync_changes_filtered` for new code; it supports optional
A server/src/routes/synckit/groups.rs +412
@@ -0,0 +1,412 @@
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,
18 + };
19 + use sqlx::PgPool;
20 +
21 + use crate::{
22 + constants,
23 + db::{self, DbSyncGroup, SyncGroupId, UserId},
24 + error::{AppError, Result},
25 + synckit_auth::SyncUser,
26 + validation,
27 + };
28 +
29 + use super::{
30 + AddMemberRequest, CreateGroupRequest, GroupGrantResponse, GroupMemberResponse, GroupResponse,
31 + PullChangeEntry, PullRequest, PullResponse, PushRequest, PushResponse,
32 + };
33 +
34 + /// Fetch a group scoped to the caller's app, or 404. Guards every group handler
35 + /// against cross-app id guessing before any membership check.
36 + async fn require_group(
37 + db: &PgPool,
38 + app_id: db::SyncAppId,
39 + group_id: SyncGroupId,
40 + ) -> Result<DbSyncGroup> {
41 + db::synckit::get_group(db, app_id, group_id)
42 + .await?
43 + .ok_or(AppError::NotFound)
44 + }
45 +
46 + /// Reject a caller who is not a member of the group (403). The gate for
47 + /// group-scoped reads and writes.
48 + async fn require_member(db: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<()> {
49 + if db::synckit::is_group_member(db, group_id, user_id).await? {
50 + Ok(())
51 + } else {
52 + Err(AppError::Forbidden)
53 + }
54 + }
55 +
56 + /// Reject a caller who is not the group admin (403). The gate for membership
57 + /// management.
58 + async fn require_admin(db: &PgPool, group_id: SyncGroupId, user_id: UserId) -> Result<()> {
59 + if db::synckit::is_group_admin(db, group_id, user_id).await? {
60 + Ok(())
61 + } else {
62 + Err(AppError::Forbidden)
63 + }
64 + }
65 +
66 + /// Create a group. The caller becomes its admin and first member, carrying the
67 + /// GCK they sealed to their own identity key.
68 + #[utoipa::path(post, path = "/api/v1/sync/groups", tag = "SyncKit",
69 + request_body = CreateGroupRequest,
70 + responses((status = 200, description = "Created group", body = GroupResponse)),
71 + security(("bearer" = [])),
72 + )]
73 + #[tracing::instrument(skip_all, name = "synckit::create_group")]
74 + pub(super) async fn create_group(
75 + State(db): State<PgPool>,
76 + sync_user: SyncUser,
77 + Json(req): Json<CreateGroupRequest>,
78 + ) -> Result<impl IntoResponse> {
79 + validation::validate_sync_group_name(&req.name)?;
80 + if req.admin_sealed_gck.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES {
81 + return Err(AppError::BadRequest(
82 + "Sealed key exceeds size limit".to_string(),
83 + ));
84 + }
85 +
86 + let group = db::synckit::create_group(
87 + &db,
88 + sync_user.app_id,
89 + sync_user.user_id,
90 + &req.name,
91 + &req.admin_sealed_gck,
92 + )
93 + .await?;
94 +
95 + Ok(Json(GroupResponse::from(group)))
96 + }
97 +
98 + /// List the groups the caller belongs to within this app.
99 + #[utoipa::path(get, path = "/api/v1/sync/groups", tag = "SyncKit",
100 + responses((status = 200, description = "Groups the user belongs to", body = Vec<GroupResponse>)),
101 + security(("bearer" = [])),
102 + )]
103 + #[tracing::instrument(skip_all, name = "synckit::list_groups")]
104 + pub(super) async fn list_groups(
105 + State(db): State<PgPool>,
106 + sync_user: SyncUser,
107 + ) -> Result<impl IntoResponse> {
108 + let groups =
109 + db::synckit::list_groups_for_user(&db, sync_user.app_id, sync_user.user_id).await?;
110 + let response: Vec<GroupResponse> = groups.into_iter().map(GroupResponse::from).collect();
111 + Ok(Json(response))
112 + }
113 +
114 + /// Add a member to a group (or replace their grant). Admin only.
115 + ///
116 + /// The admin resolves the member out of band, seals the current GCK to that
117 + /// member's public key, and posts `{member_email, sealed_gck}`. The server maps
118 + /// the email to a verified account and stores the opaque grant at the group's
119 + /// current GCK generation.
120 + #[utoipa::path(post, path = "/api/v1/sync/groups/{id}/members", tag = "SyncKit",
121 + params(("id" = String, Path, description = "Group ID")),
122 + request_body = AddMemberRequest,
123 + responses((status = 204, description = "Member added"), (status = 403, description = "Not the group admin")),
124 + security(("bearer" = [])),
125 + )]
126 + #[tracing::instrument(skip_all, name = "synckit::add_group_member")]
127 + pub(super) async fn add_member(
128 + State(db): State<PgPool>,
129 + sync_user: SyncUser,
130 + Path(group_id): Path<SyncGroupId>,
131 + Json(req): Json<AddMemberRequest>,
132 + ) -> Result<impl IntoResponse> {
133 + let group = require_group(&db, sync_user.app_id, group_id).await?;
134 + require_admin(&db, group_id, sync_user.user_id).await?;
135 +
136 + if req.sealed_gck.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES {
137 + return Err(AppError::BadRequest(
138 + "Sealed key exceeds size limit".to_string(),
139 + ));
140 + }
141 + let role = req.role.as_deref().unwrap_or("member");
142 + if role != "member" && role != "admin" {
143 + return Err(AppError::BadRequest(
144 + "role must be 'member' or 'admin'".to_string(),
145 + ));
146 + }
147 +
148 + let email = db::Email::new(&req.member_email)
149 + .map_err(|_| AppError::BadRequest("Invalid email address".to_string()))?;
150 + let member_id = db::users::get_verified_user_id_by_email(&db, &email)
151 + .await?
152 + .ok_or_else(|| AppError::BadRequest("No verified account for that email".to_string()))?;
153 +
154 + // The grant the admin sends is sealed under the group's current GCK, so it is
155 + // stored at that generation.
156 + db::synckit::add_or_update_member(
157 + &db,
158 + group_id,
159 + member_id,
160 + role,
161 + &req.sealed_gck,
162 + group.gck_version,
163 + )
164 + .await?;
165 +
166 + Ok(StatusCode::NO_CONTENT)
167 + }
168 +
169 + /// Remove a member from a group. Admin only.
170 + ///
171 + /// This drops the member from future group writes. Forward secrecy for writes
172 + /// after removal comes from the admin then rotating the GCK (p3); data the member
173 + /// already pulled is already in their hands.
174 + #[utoipa::path(delete, path = "/api/v1/sync/groups/{id}/members/{user_id}", tag = "SyncKit",
175 + params(
176 + ("id" = String, Path, description = "Group ID"),
177 + ("user_id" = String, Path, description = "Member user ID"),
178 + ),
179 + responses((status = 204, description = "Member removed"), (status = 404, description = "Not a member")),
180 + security(("bearer" = [])),
181 + )]
182 + #[tracing::instrument(skip_all, name = "synckit::remove_group_member")]
183 + pub(super) async fn remove_member(
184 + State(db): State<PgPool>,
185 + sync_user: SyncUser,
186 + Path((group_id, member_id)): Path<(SyncGroupId, UserId)>,
187 + ) -> Result<impl IntoResponse> {
188 + let group = require_group(&db, sync_user.app_id, group_id).await?;
189 + require_admin(&db, group_id, sync_user.user_id).await?;
190 +
191 + // The admin cannot remove themselves; that would orphan the group. Deleting a
192 + // group is a separate action (not yet exposed).
193 + if member_id == group.admin_user_id {
194 + return Err(AppError::BadRequest(
195 + "The group admin cannot be removed".to_string(),
196 + ));
197 + }
198 +
199 + if !db::synckit::remove_member(&db, group_id, member_id).await? {
200 + return Err(AppError::NotFound);
201 + }
202 +
203 + Ok(StatusCode::NO_CONTENT)
204 + }
205 +
206 + /// List a group's members (id, role, joined-at). Members only. Grants are not
207 + /// included — each member fetches only their own via `/grant`.
208 + #[utoipa::path(get, path = "/api/v1/sync/groups/{id}/members", tag = "SyncKit",
209 + params(("id" = String, Path, description = "Group ID")),
210 + responses((status = 200, description = "Group members", body = Vec<GroupMemberResponse>)),
211 + security(("bearer" = [])),
212 + )]
213 + #[tracing::instrument(skip_all, name = "synckit::list_group_members")]
214 + pub(super) async fn list_members(
215 + State(db): State<PgPool>,
216 + sync_user: SyncUser,
217 + Path(group_id): Path<SyncGroupId>,
218 + ) -> Result<impl IntoResponse> {
219 + require_group(&db, sync_user.app_id, group_id).await?;
220 + require_member(&db, group_id, sync_user.user_id).await?;
221 +
222 + let members = db::synckit::list_members(&db, group_id).await?;
223 + let response: Vec<GroupMemberResponse> = members
224 + .into_iter()
225 + .map(|m| GroupMemberResponse {
226 + user_id: m.user_id,
227 + role: m.role,
228 + added_at: m.added_at,
229 + })
230 + .collect();
231 + Ok(Json(response))
232 + }
233 +
234 + /// Fetch the caller's own sealed GCK grant for a group, so their device can open
235 + /// the GCK and read the group changelog. Members only.
236 + #[utoipa::path(get, path = "/api/v1/sync/groups/{id}/grant", tag = "SyncKit",
237 + params(("id" = String, Path, description = "Group ID")),
238 + responses(
239 + (status = 200, description = "The caller's sealed grant", body = GroupGrantResponse),
240 + (status = 403, description = "Not a member"),
241 + ),
242 + security(("bearer" = [])),
243 + )]
244 + #[tracing::instrument(skip_all, name = "synckit::get_group_grant")]
245 + pub(super) async fn get_grant(
246 + State(db): State<PgPool>,
247 + sync_user: SyncUser,
248 + Path(group_id): Path<SyncGroupId>,
249 + ) -> Result<impl IntoResponse> {
250 + require_group(&db, sync_user.app_id, group_id).await?;
251 +
252 + let (sealed_gck, gck_version) = db::synckit::get_member_grant(&db, group_id, sync_user.user_id)
253 + .await?
254 + .ok_or(AppError::Forbidden)?;
255 +
256 + Ok(Json(GroupGrantResponse {
257 + sealed_gck,
258 + gck_version,
259 + }))
260 + }
261 +
262 + /// Push encrypted changes to a group's shared changelog. Members only.
263 + #[utoipa::path(post, path = "/api/v1/sync/groups/{id}/push", tag = "SyncKit",
264 + params(("id" = String, Path, description = "Group ID")),
265 + request_body = PushRequest,
266 + responses((status = 200, description = "New cursor position", body = PushResponse)),
267 + security(("bearer" = [])),
268 + )]
269 + #[tracing::instrument(skip_all, name = "synckit::group_push", fields(group_id))]
270 + pub(super) async fn group_push(
271 + State(db): State<PgPool>,
272 + sync_user: SyncUser,
273 + Path(group_id): Path<SyncGroupId>,
274 + Json(req): Json<PushRequest>,
275 + ) -> Result<impl IntoResponse> {
276 + require_group(&db, sync_user.app_id, group_id).await?;
277 + require_member(&db, group_id, sync_user.user_id).await?;
278 + // NOTE: the paid-write gate and SSE push notification (present on personal
279 + // push) are attached in p2-billing; group writes are open in this slice.
280 +
281 + if req.changes.is_empty() {
282 + return Err(AppError::BadRequest("No changes provided".to_string()));
283 + }
284 + if req.changes.len() > constants::SYNCKIT_PUSH_MAX_CHANGES {
285 + return Err(AppError::BadRequest(format!(
286 + "Maximum {} changes per push",
287 + constants::SYNCKIT_PUSH_MAX_CHANGES
288 + )));
289 + }
290 + for change in &req.changes {
291 + validation::validate_sync_table_name(&change.table)?;
292 + validation::validate_sync_row_id(&change.row_id)?;
293 + if change.op == db::SyncOperation::Delete && change.data.is_some() {
294 + return Err(AppError::BadRequest(
295 + "DELETE operations should not include data".to_string(),
296 + ));
297 + }
298 + }
299 +
300 + // The pushing device must belong to the pushing user (membership is a
301 + // separate, group-level check above).
302 + if !db::synckit::sync_device_belongs(&db, req.device_id, sync_user.app_id, sync_user.user_id)
303 + .await?
304 + {
305 + return Err(AppError::BadRequest("Unknown device".to_string()));
306 + }
307 + db::synckit::touch_sync_device(&db, req.device_id).await?;
308 +
309 + let changes: Vec<_> = req
310 + .changes
311 + .iter()
312 + .map(|c| {
313 + (
314 + c.table.clone(),
315 + c.op.to_string(),
316 + c.row_id.clone(),
317 + c.timestamp,
318 + c.data.clone(),
319 + )
320 + })
321 + .collect();
322 +
323 + let cursor = db::synckit::push_group_changes(
324 + &db,
325 + sync_user.app_id,
326 + group_id,
327 + sync_user.user_id,
328 + req.device_id,
329 + req.batch_id,
330 + &changes,
331 + )
332 + .await?;
333 +
334 + Ok(Json(PushResponse { cursor }))
335 + }
336 +
337 + /// Pull a group's changes after a cursor. Members only.
338 + #[utoipa::path(post, path = "/api/v1/sync/groups/{id}/pull", tag = "SyncKit",
339 + params(("id" = String, Path, description = "Group ID")),
340 + request_body = PullRequest,
341 + responses((status = 200, description = "Changes since cursor", body = PullResponse)),
342 + security(("bearer" = [])),
343 + )]
344 + #[tracing::instrument(skip_all, name = "synckit::group_pull", fields(group_id))]
345 + pub(super) async fn group_pull(
346 + State(db): State<PgPool>,
347 + sync_user: SyncUser,
348 + Path(group_id): Path<SyncGroupId>,
349 + Json(req): Json<PullRequest>,
350 + ) -> Result<impl IntoResponse> {
351 + require_group(&db, sync_user.app_id, group_id).await?;
352 + require_member(&db, group_id, sync_user.user_id).await?;
353 +
354 + if !db::synckit::sync_device_belongs(&db, req.device_id, sync_user.app_id, sync_user.user_id)
355 + .await?
356 + {
357 + return Err(AppError::BadRequest("Unknown device".to_string()));
358 + }
359 +
360 + if let Some(ref tables) = req.tables {
361 + if tables.len() > 50 {
362 + return Err(AppError::BadRequest(
363 + "Maximum 50 table names per filter".to_string(),
364 + ));
365 + }
366 + for table in tables {
367 + validation::validate_sync_table_name(table)?;
368 + }
369 + }
370 +
371 + let page_size = constants::SYNCKIT_PULL_PAGE_SIZE;
372 + let entries = db::synckit::pull_group_changes_filtered(
373 + &db,
374 + sync_user.app_id,
375 + group_id,
376 + req.cursor,
377 + page_size,
378 + req.tables.as_deref(),
379 + req.since,
380 + )
381 + .await?;
382 +
383 + let has_more = entries.len() as i64 == page_size;
384 + let new_cursor = entries.last().map(|e| e.seq).unwrap_or(req.cursor);
385 +
386 + // Touch the device for activity, but do NOT advance the per-device personal
387 + // compaction cursor here — that cursor governs personal-changelog retention
388 + // and must not be moved by a group pull. Group changelog retention is a
389 + // separate concern (future work).
390 + db::synckit::touch_sync_device(&db, req.device_id).await?;
391 +
392 + let changes: Vec<PullChangeEntry> = entries
393 + .into_iter()
394 + .map(|e| PullChangeEntry {
395 + seq: e.seq,
396 + device_id: e.device_id,
397 + table: e.table_name,
398 + op: e.operation.to_string(),
399 + row_id: e.row_id,
400 + timestamp: e.client_timestamp,
401 + data: e.data,
402 + // Group entries key off the group's GCK, not a per-user key_id.
403 + key_id: None,
404 + })
405 + .collect();
406 +
407 + Ok(Json(PullResponse {
408 + changes,
409 + cursor: new_cursor,
410 + has_more,
411 + }))
412 + }
M server/src/routes/synckit/mod.rs +121 -1
@@ -18,6 +18,7 @@ pub(crate) mod apps;
18 18 pub(crate) mod auth;
19 19 pub(crate) mod billing;
20 20 pub(crate) mod blobs;
21 + pub(crate) mod groups;
21 22 pub(crate) mod keys;
22 23 mod subscribe;
23 24 pub(crate) mod sync;
@@ -33,7 +34,7 @@ use crate::{
33 34 CsrfRouter, delete_csrf, delete_csrf_skip, patch_csrf, post_csrf, post_csrf_skip, put_csrf,
34 35 put_csrf_skip,
35 36 },
36 - db::{self, SyncAppId, SyncDeviceId, SyncOperation, SyncPlatform, UserId},
37 + db::{self, SyncAppId, SyncDeviceId, SyncGroupId, SyncOperation, SyncPlatform, UserId},
37 38 };
38 39
39 40 /// Reason strings for synckit CSRF Skip routes. The auth_routes and
@@ -191,6 +192,73 @@ pub(crate) struct SyncAccountResponse {
191 192 pub username: String,
192 193 }
193 194
195 + // ── Group types ──
196 +
197 + #[derive(Deserialize, utoipa::ToSchema)]
198 + pub(crate) struct CreateGroupRequest {
199 + pub name: String,
200 + /// The Group Content Key sealed to the creating admin's own identity public
201 + /// key (base64), opaque to the server.
202 + pub admin_sealed_gck: String,
203 + }
204 +
205 + #[derive(Serialize, utoipa::ToSchema)]
206 + pub(crate) struct GroupResponse {
207 + #[schema(value_type = String)]
208 + id: SyncGroupId,
209 + #[schema(value_type = String)]
210 + app_id: SyncAppId,
211 + #[schema(value_type = String)]
212 + admin_user_id: UserId,
213 + name: String,
214 + gck_version: i32,
215 + #[schema(value_type = String)]
216 + created_at: DateTime<Utc>,
217 + }
218 +
219 + impl From<db::DbSyncGroup> for GroupResponse {
220 + fn from(g: db::DbSyncGroup) -> Self {
221 + Self {
222 + id: g.id,
223 + app_id: g.app_id,
224 + admin_user_id: g.admin_user_id,
225 + name: g.name,
226 + gck_version: g.gck_version,
227 + created_at: g.created_at,
228 + }
229 + }
230 + }
231 +
232 + #[derive(Deserialize, utoipa::ToSchema)]
233 + pub(crate) struct AddMemberRequest {
234 + /// The member's account email, resolved to a verified user server-side.
235 + pub member_email: String,
236 + /// The GCK sealed to the member's identity public key (base64), produced by
237 + /// the admin with the group's current GCK. Opaque to the server.
238 + pub sealed_gck: String,
239 + /// Optional role: "member" (default) or "admin".
240 + #[serde(default)]
241 + pub role: Option<String>,
242 + }
243 +
244 + #[derive(Serialize, utoipa::ToSchema)]
245 + pub(crate) struct GroupMemberResponse {
246 + #[schema(value_type = String)]
247 + user_id: UserId,
248 + role: String,
249 + #[schema(value_type = String)]
250 + added_at: DateTime<Utc>,
251 + }
252 +
253 + #[derive(Serialize, utoipa::ToSchema)]
254 + pub(crate) struct GroupGrantResponse {
255 + /// The caller's sealed GCK grant (base64); opened client-side with the
256 + /// member's identity private key.
257 + sealed_gck: String,
258 + /// The GCK generation this grant was sealed under.
259 + gck_version: i32,
260 + }
261 +
194 262 /// Status of the authenticated user's subscription to this app's cloud sync.
195 263 /// Shape matches `synckit_client::SubscriptionStatus`.
196 264 #[derive(Serialize, utoipa::ToSchema)]
@@ -697,6 +765,58 @@ pub fn synckit_routes(synckit_jwt_secret: Option<std::sync::Arc<String>>) -> Csr
697 765 "/api/v1/sync/pull",
698 766 post_csrf_skip(SYNCKIT_JWT_SKIP, sync::sync_pull),
699 767 )
768 + // Group sync: shared changelogs. Membership/admin gating lives inside the
769 + // handlers (SyncUser identifies the caller); same JWT auth + dual rate
770 + // limit as personal sync. GET+POST on one path merge, as with devices.
771 + .route(
772 + "/api/sync/groups",
773 + post_csrf_skip(SYNCKIT_JWT_SKIP, groups::create_group),
774 + )
775 + .route(
776 + "/api/v1/sync/groups",
777 + post_csrf_skip(SYNCKIT_JWT_SKIP, groups::create_group),
778 + )
779 + .route_get("/api/sync/groups", get(groups::list_groups))
780 + .route_get("/api/v1/sync/groups", get(groups::list_groups))
781 + .route(
782 + "/api/sync/groups/{id}/members",
783 + post_csrf_skip(SYNCKIT_JWT_SKIP, groups::add_member),
784 + )
785 + .route(
786 + "/api/v1/sync/groups/{id}/members",
787 + post_csrf_skip(SYNCKIT_JWT_SKIP, groups::add_member),
788 + )
789 + .route_get("/api/sync/groups/{id}/members", get(groups::list_members))
790 + .route_get(
791 + "/api/v1/sync/groups/{id}/members",
792 + get(groups::list_members),
793 + )
794 + .route(
795 + "/api/sync/groups/{id}/members/{user_id}",
796 + delete_csrf_skip(SYNCKIT_JWT_SKIP, groups::remove_member),
797 + )
798 + .route(
799 + "/api/v1/sync/groups/{id}/members/{user_id}",
800 + delete_csrf_skip(SYNCKIT_JWT_SKIP, groups::remove_member),
801 + )
802 + .route_get("/api/sync/groups/{id}/grant", get(groups::get_grant))
803 + .route_get("/api/v1/sync/groups/{id}/grant", get(groups::get_grant))
804 + .route(
805 + "/api/sync/groups/{id}/push",
806 + post_csrf_skip(SYNCKIT_JWT_SKIP, groups::group_push),
807 + )
808 + .route(
809 + "/api/v1/sync/groups/{id}/push",
810 + post_csrf_skip(SYNCKIT_JWT_SKIP, groups::group_push),
811 + )
812 + .route(
813 + "/api/sync/groups/{id}/pull",
814 + post_csrf_skip(SYNCKIT_JWT_SKIP, groups::group_pull),
815 + )
816 + .route(
817 + "/api/v1/sync/groups/{id}/pull",
818 + post_csrf_skip(SYNCKIT_JWT_SKIP, groups::group_pull),
819 + )
700 820 .route_get("/api/sync/subscribe", get(subscribe::sync_subscribe))
701 821 .route_get("/api/v1/sync/subscribe", get(subscribe::sync_subscribe))
702 822 .route_get("/api/sync/status", get(sync::sync_status))
M server/src/validation/mod.rs +20
@@ -195,6 +195,26 @@ pub fn validate_sync_device_name(name: &str) -> Result<(), AppError> {
195 195 Ok(())
196 196 }
197 197
198 + /// Validate a sync group name. Same shape as an app/device name: non-empty,
199 + /// bounded by the `VARCHAR(100)` column, no control characters.
200 + pub fn validate_sync_group_name(name: &str) -> Result<(), AppError> {
201 + if name.is_empty() {
202 + return Err(AppError::validation("Group name is required".to_string()));
203 + }
204 + if name.chars().count() > limits::SYNC_APP_NAME_MAX {
205 + return Err(AppError::validation(format!(
206 + "Group name must be {} characters or less",
207 + limits::SYNC_APP_NAME_MAX
208 + )));
209 + }
210 + if name.chars().any(|c| c.is_control()) {
211 + return Err(AppError::validation(
212 + "Group name must not contain control characters".to_string(),
213 + ));
214 + }
215 + Ok(())
216 + }
217 +
198 218 /// Validate a sync table name
199 219 pub fn validate_sync_table_name(name: &str) -> Result<(), AppError> {
200 220 if name.is_empty() {
A server/tests/workflows/db_synckit_groups.rs +480
@@ -0,0 +1,480 @@
1 + //! DB-layer contract tests for `db::synckit::groups` and the group-scoped
2 + //! changelog (`push_group_changes` / `pull_group_changes_filtered`).
3 + //!
4 + //! The load-bearing test here is `group_entries_never_appear_in_personal_sync`:
5 + //! group changes live in their own `sync_group_log` table, so personal-scope
6 + //! queries (personal pull, status) can never see GCK-encrypted group rows. That
7 + //! isolation is structural, not a `group_id IS NULL` guard sprinkled across every
8 + //! personal query. Design: wiki synckit-groups-design.
9 +
10 + use crate::harness::db::TestDb;
11 + use makenotwork::db::synckit;
12 + use makenotwork::db::{SyncAppId, SyncDeviceId, UserId};
13 + use uuid::Uuid;
14 +
15 + /// Seed a verified user.
16 + async fn seed_user(pool: &sqlx::PgPool, username: &str) -> UserId {
17 + let hash = makenotwork::auth::hash_password("password123").expect("hash");
18 + sqlx::query_scalar::<_, UserId>(
19 + "INSERT INTO users (username, email, password_hash, email_verified)
20 + VALUES ($1, $2, $3, true) RETURNING id",
21 + )
22 + .bind(username)
23 + .bind(format!("{username}@test.com"))
24 + .bind(&hash)
25 + .fetch_one(pool)
26 + .await
27 + .expect("seed user")
28 + }
29 +
30 + /// Seed a sync app owned by `user`.
31 + async fn seed_app(pool: &sqlx::PgPool, user: UserId, name: &str) -> SyncAppId {
32 + sqlx::query_scalar::<_, SyncAppId>(
33 + "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix)
34 + VALUES ($1, $2, $3, $4) RETURNING id",
35 + )
36 + .bind(user)
37 + .bind(name)
38 + .bind(format!("hash_{name}"))
39 + .bind(&name[..name.len().min(8)])
40 + .fetch_one(pool)
41 + .await
42 + .expect("seed sync app")
43 + }
44 +
45 + /// Seed a device row for a user within an app.
46 + async fn seed_device(
47 + pool: &sqlx::PgPool,
48 + app: SyncAppId,
49 + user: UserId,
50 + name: &str,
51 + ) -> SyncDeviceId {
52 + sqlx::query_scalar::<_, SyncDeviceId>(
53 + "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
54 + VALUES ($1, $2, $3, 'macos') RETURNING id",
55 + )
56 + .bind(app)
57 + .bind(user)
58 + .bind(name)
59 + .fetch_one(pool)
60 + .await
61 + .expect("seed device")
62 + }
63 +
64 + /// One INSERT change tuple in the shape `push_*_changes` expects.
65 + fn change(
66 + table: &str,
67 + row: &str,
68 + v: i32,
69 + ) -> (
70 + String,
71 + String,
72 + String,
73 + chrono::DateTime<chrono::Utc>,
74 + Option<serde_json::Value>,
75 + ) {
76 + (
77 + table.to_string(),
78 + "INSERT".to_string(),
79 + row.to_string(),
80 + chrono::Utc::now(),
81 + Some(serde_json::json!({ "v": v })),
82 + )
83 + }
84 +
85 + // ── Group + membership + grants ──────────────────────────────────────────────
86 +
87 + #[tokio::test]
88 + async fn create_group_enrolls_admin_as_member_and_admin() {
89 + let db = TestDb::new().await;
90 + let admin = seed_user(&db.pool, "grp_admin").await;
91 + let app = seed_app(&db.pool, admin, "grpcreate").await;
92 +
93 + let group = synckit::create_group(&db.pool, app, admin, "Team", "sealed_admin")
94 + .await
95 + .unwrap();
96 +
97 + assert_eq!(group.gck_version, 1);
98 + assert!(
99 + synckit::is_group_member(&db.pool, group.id, admin)
100 + .await
101 + .unwrap()
102 + );
103 + assert!(
104 + synckit::is_group_admin(&db.pool, group.id, admin)
105 + .await
106 + .unwrap()
107 + );
108 + assert_eq!(
109 + synckit::get_member_grant(&db.pool, group.id, admin)
110 + .await
111 + .unwrap(),
112 + Some(("sealed_admin".to_string(), 1)),
113 + );
114 + }
115 +
116 + #[tokio::test]
117 + async fn add_member_stores_grant_and_gates_admin() {
118 + let db = TestDb::new().await;
119 + let admin = seed_user(&db.pool, "gm_admin").await;
120 + let app = seed_app(&db.pool, admin, "gmadd").await;
121 + let group = synckit::create_group(&db.pool, app, admin, "Team", "sealed_admin")
122 + .await
123 + .unwrap();
124 +
125 + let bob = seed_user(&db.pool, "gm_bob").await;
126 + synckit::add_or_update_member(
127 + &db.pool,
128 + group.id,
129 + bob,
130 + "member",
131 + "sealed_bob",
132 + group.gck_version,
133 + )
134 + .await
135 + .unwrap();
136 +
137 + assert!(
138 + synckit::is_group_member(&db.pool, group.id, bob)
139 + .await
140 + .unwrap()
141 + );
142 + assert!(
143 + !synckit::is_group_admin(&db.pool, group.id, bob)
144 + .await
145 + .unwrap()
146 + );
147 + assert_eq!(
148 + synckit::get_member_grant(&db.pool, group.id, bob)
149 + .await
150 + .unwrap(),
151 + Some(("sealed_bob".to_string(), 1)),
152 + );
153 +
154 + // A non-member has no grant and is not a member.
155 + let carol = seed_user(&db.pool, "gm_carol").await;
156 + assert!(
157 + !synckit::is_group_member(&db.pool, group.id, carol)
158 + .await
159 + .unwrap()
160 + );
161 + assert_eq!(
162 + synckit::get_member_grant(&db.pool, group.id, carol)
163 + .await
164 + .unwrap(),
165 + None
166 + );
167 + }
168 +
169 + #[tokio::test]
170 + async fn add_or_update_member_is_idempotent_upsert() {
171 + let db = TestDb::new().await;
172 + let admin = seed_user(&db.pool, "up_admin").await;
173 + let app = seed_app(&db.pool, admin, "upsert").await;
174 + let group = synckit::create_group(&db.pool, app, admin, "Team", "sealed_admin")
175 + .await
176 + .unwrap();
177 + let bob = seed_user(&db.pool, "up_bob").await;
178 +
179 + synckit::add_or_update_member(&db.pool, group.id, bob, "member", "grant_v1", 1)
180 + .await
181 + .unwrap();
182 + // Re-adding replaces the grant in place, not a second row.
183 + synckit::add_or_update_member(&db.pool, group.id, bob, "member", "grant_v2", 1)
184 + .await
185 + .unwrap();
186 +
187 + assert_eq!(
188 + synckit::get_member_grant(&db.pool, group.id, bob)
189 + .await
190 + .unwrap(),
191 + Some(("grant_v2".to_string(), 1)),
192 + );
193 + // admin + bob, no duplicate.
194 + assert_eq!(
195 + synckit::list_members(&db.pool, group.id)
196 + .await
197 + .unwrap()
198 + .len(),
199 + 2
200 + );
201 + }
202 +
203 + #[tokio::test]
204 + async fn list_groups_for_user_is_membership_scoped() {
205 + let db = TestDb::new().await;
206 + let admin = seed_user(&db.pool, "lg_admin").await;
207 + let app = seed_app(&db.pool, admin, "listgrp").await;
208 + let g1 = synckit::create_group(&db.pool, app, admin, "One", "s")
209 + .await
210 + .unwrap();
211 + let _g2 = synckit::create_group(&db.pool, app, admin, "Two", "s")
212 + .await
213 + .unwrap();
214 +
215 + let bob = seed_user(&db.pool, "lg_bob").await;
216 + synckit::add_or_update_member(&db.pool, g1.id, bob, "member", "sb", 1)
217 + .await
218 + .unwrap();
219 + let carol = seed_user(&db.pool, "lg_carol").await;
220 +
221 + assert_eq!(
222 + synckit::list_groups_for_user(&db.pool, app, admin)
223 + .await
224 + .unwrap()
225 + .len(),
226 + 2
227 + );
228 + assert_eq!(
229 + synckit::list_groups_for_user(&db.pool, app, bob)
230 + .await
231 + .unwrap()
232 + .len(),
233 + 1
234 + );
235 + assert_eq!(
236 + synckit::list_groups_for_user(&db.pool, app, carol)
237 + .await
238 + .unwrap()
239 + .len(),
240 + 0
241 + );
242 + }
243 +
244 + #[tokio::test]
245 + async fn remove_member_revokes_membership() {
246 + let db = TestDb::new().await;
247 + let admin = seed_user(&db.pool, "rm_admin").await;
248 + let app = seed_app(&db.pool, admin, "removem").await;
249 + let group = synckit::create_group(&db.pool, app, admin, "Team", "s")
250 + .await
251 + .unwrap();
252 + let bob = seed_user(&db.pool, "rm_bob").await;
253 + synckit::add_or_update_member(&db.pool, group.id, bob, "member", "sb", 1)
254 + .await
255 + .unwrap();
256 +
257 + assert!(
258 + synckit::remove_member(&db.pool, group.id, bob)
259 + .await
260 + .unwrap()
261 + );
262 + assert!(
263 + !synckit::is_group_member(&db.pool, group.id, bob)
264 + .await
265 + .unwrap()
266 + );
267 + // Removing again is a no-op.
268 + assert!(
269 + !synckit::remove_member(&db.pool, group.id, bob)
270 + .await
271 + .unwrap()
272 + );
273 + }
274 +
275 + // ── Group changelog push/pull ────────────────────────────────────────────────
276 +
277 + #[tokio::test]
278 + async fn group_push_pull_roundtrip_and_idempotent() {
279 + let db = TestDb::new().await;
280 + let admin = seed_user(&db.pool, "pp_admin").await;
281 + let app = seed_app(&db.pool, admin, "pushpull").await;
282 + let group = synckit::create_group(&db.pool, app, admin, "Team", "s")
283 + .await
284 + .unwrap();
285 + let device = seed_device(&db.pool, app, admin, "dev").await;
286 +
287 + let batch = Uuid::new_v4();
288 + let changes = vec![change("tasks", "r1", 1), change("tasks", "r2", 2)];
289 + let cursor =
290 + synckit::push_group_changes(&db.pool, app, group.id, admin, device, batch, &changes)
291 + .await
292 + .unwrap();
293 + assert!(cursor > 0);
294 +
295 + let pulled = synckit::pull_group_changes_filtered(&db.pool, app, group.id, 0, 100, None, None)
296 + .await
297 + .unwrap();
298 + assert_eq!(pulled.len(), 2);
299 + assert!(pulled.iter().all(|e| e.group_id == group.id));
300 +
301 + // Re-pushing the same batch returns the same cursor and inserts nothing new.
302 + let cursor2 =
303 + synckit::push_group_changes(&db.pool, app, group.id, admin, device, batch, &changes)
304 + .await
305 + .unwrap();
306 + assert_eq!(cursor2, cursor);
307 + let pulled2 = synckit::pull_group_changes_filtered(&db.pool, app, group.id, 0, 100, None, None)
308 + .await
309 + .unwrap();
310 + assert_eq!(pulled2.len(), 2);
311 + }
312 +
313 + #[tokio::test]
314 + async fn group_pull_is_scoped_to_its_group() {
315 + let db = TestDb::new().await;
316 + let admin = seed_user(&db.pool, "sc_admin").await;
317 + let app = seed_app(&db.pool, admin, "scoped").await;
318 + let g1 = synckit::create_group(&db.pool, app, admin, "One", "s")
319 + .await
320 + .unwrap();
321 + let g2 = synckit::create_group(&db.pool, app, admin, "Two", "s")
322 + .await
323 + .unwrap();
324 + let device = seed_device(&db.pool, app, admin, "dev").await;
325 +
326 + synckit::push_group_changes(
327 + &db.pool,
328 + app,
329 + g1.id,
330 + admin,
331 + device,
332 + Uuid::new_v4(),
333 + &[change("t", "a", 1)],
334 + )
335 + .await
336 + .unwrap();
337 + synckit::push_group_changes(
338 + &db.pool,
339 + app,
340 + g2.id,
341 + admin,
342 + device,
343 + Uuid::new_v4(),
344 + &[change("t", "b", 2)],
345 + )
346 + .await
347 + .unwrap();
348 +
349 + // Each group sees only its own entries.
350 + assert_eq!(
351 + synckit::pull_group_changes_filtered(&db.pool, app, g1.id, 0, 100, None, None)
352 + .await
353 + .unwrap()
354 + .len(),
355 + 1
356 + );
357 + assert_eq!(
358 + synckit::pull_group_changes_filtered(&db.pool, app, g2.id, 0, 100, None, None)
359 + .await
360 + .unwrap()
361 + .len(),
362 + 1
363 + );
364 + }
365 +
366 + /// The payoff of the dedicated `sync_group_log` table: group changes are
367 + /// invisible to personal-scope queries, and vice versa, with no per-query guard.
368 + #[tokio::test]
369 + async fn group_entries_never_appear_in_personal_sync() {
370 + let db = TestDb::new().await;
371 + let user = seed_user(&db.pool, "iso_user").await;
372 + let app = seed_app(&db.pool, user, "isolate").await;
373 + let group = synckit::create_group(&db.pool, app, user, "Team", "s")
374 + .await
375 + .unwrap();
376 + let device = seed_device(&db.pool, app, user, "dev").await;
377 +
378 + // Push two GROUP changes as this user's device.
379 + synckit::push_group_changes(
380 + &db.pool,
381 + app,
382 + group.id,
383 + user,
384 + device,
385 + Uuid::new_v4(),
386 + &[change("tasks", "g1", 1), change("tasks", "g2", 2)],
387 + )
388 + .await
389 + .unwrap();
390 +
391 + // The user's PERSONAL pull sees nothing — group rows are in a different table.
392 + let personal = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, None)
393 + .await
394 + .unwrap();
395 + assert!(
396 + personal.is_empty(),
397 + "group entries must not leak into personal pull"
398 + );
399 +
400 + // Personal status counts zero personal changes.
401 + let (total, latest) = synckit::get_sync_status(&db.pool, app, user).await.unwrap();
402 + assert_eq!(total, 0);
403 + assert_eq!(latest, None);
404 +
405 + // A PERSONAL push does not appear in the group pull either.
406 + synckit::push_sync_changes(
407 + &db.pool,
408 + app,
409 + user,
410 + device,
411 + Uuid::new_v4(),
412 + &[change("notes", "p1", 9)],
413 + )
414 + .await
415 + .unwrap();
416 + let group_pull =
417 + synckit::pull_group_changes_filtered(&db.pool, app, group.id, 0, 100, None, None)
418 + .await
419 + .unwrap();
420 + assert_eq!(
421 + group_pull.len(),
422 + 2,
423 + "personal entries must not leak into group pull"
424 + );
425 +
426 + // And the personal pull now sees exactly the one personal entry.
427 + let personal2 = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, None)
428 + .await
429 + .unwrap();
430 + assert_eq!(personal2.len(), 1);
431 + }
432 +
433 + // ── Rotation ─────────────────────────────────────────────────────────────────
434 +
435 + #[tokio::test]
436 + async fn rotate_group_gck_bumps_version_regrants_and_drops_removed() {
437 + let db = TestDb::new().await;
438 + let admin = seed_user(&db.pool, "rot_admin").await;
439 + let app = seed_app(&db.pool, admin, "grprot").await;
440 + let group = synckit::create_group(&db.pool, app, admin, "Team", "admin_v1")
441 + .await
442 + .unwrap();
443 + let bob = seed_user(&db.pool, "rot_bob").await;
444 + let carol = seed_user(&db.pool, "rot_carol").await;
445 + synckit::add_or_update_member(&db.pool, group.id, bob, "member", "bob_v1", 1)
446 + .await
447 + .unwrap();
448 + synckit::add_or_update_member(&db.pool, group.id, carol, "member", "carol_v1", 1)
449 + .await
450 + .unwrap();
451 +
452 + // Rotate to generation 2, re-granting admin + bob only (carol is removed).
453 + let grants = vec![(admin, "admin_v2".to_string()), (bob, "bob_v2".to_string())];
454 + synckit::rotate_group_gck(&db.pool, group.id, 2, &grants)
455 + .await
456 + .unwrap();
457 +
458 + let g = synckit::get_group(&db.pool, app, group.id)
459 + .await
460 + .unwrap()
461 + .unwrap();
462 + assert_eq!(g.gck_version, 2);
463 + assert_eq!(
464 + synckit::get_member_grant(&db.pool, group.id, admin)
465 + .await
466 + .unwrap(),
467 + Some(("admin_v2".to_string(), 2))
468 + );
469 + assert_eq!(
470 + synckit::get_member_grant(&db.pool, group.id, bob)
471 + .await
472 + .unwrap(),
473 + Some(("bob_v2".to_string(), 2))
474 + );
475 + assert!(
476 + !synckit::is_group_member(&db.pool, group.id, carol)
477 + .await
478 + .unwrap()
479 + );
480 + }
M server/tests/workflows/mod.rs +1
@@ -40,6 +40,7 @@ mod db_scan_jobs_layer;
40 40 mod db_scanning_layer;
41 41 mod db_ssh_keys_layer;
42 42 mod db_synckit_billing_layer;
43 + mod db_synckit_groups;
43 44 mod db_synckit_rotation;
44 45 mod db_transactions_layer;
45 46 mod db_users_layer;