From 8f98c00e0a3e6bd7be99bc883e5cf5a63ec077f5 Mon Sep 17 00:00:00 2001 From: noevidence1017 Date: Wed, 29 Jul 2026 03:03:57 +0100 Subject: [PATCH] feat(conversations): sequence group epoch changes and emit system events Membership changes and epoch transitions now go into an ordered, gap-free per-conversation log, so a client that missed commits can fetch exactly what it missed and converge on the same epoch as everyone else. - Chat messages are ordered by (createdAt, id), which is fine for a timeline but wrong for group control: applying a join and a leave in the wrong order derives a different key schedule, and a timestamp cursor can silently skip an event written slightly out of clock order. group_control_events therefore carries its own strictly monotonic sequence, and conversations.epoch records the resulting epoch. - Sequencing is serialized on the conversation row: the epoch is bumped with UPDATE ... RETURNING in the same transaction that assigns the sequence, so a concurrent join and leave are forced into a real order. The unique index on (conversationId, sequence) is the backstop, not the mechanism. - The membership row and its control event commit together. A member written without the epoch bump announcing them would leave every other client unaware of someone who can now decrypt, which is the divergence this log exists to prevent. The live broadcast follows the commit, so a client reacting to the event always finds the membership already in place. - Every event persists a content_type='system' message in one stable shape and links to it, so the timeline and the control log cannot disagree. Live fan-out emits group_system_event, epoch_changed and new_message, the last so existing timeline rendering picks it up unchanged. - GET /conversations/:id/epoch answers "am I behind?" as an integer compare. GET /conversations/:id/group-control returns missed events in replay order with an exclusive cursor, so replaying the same cursor never re-applies an event. POST /conversations/:id/group-control sequences a client MLS commit; the payload is opaque and capped, since the server orders group control rather than interpreting it. - The last member leaving emits nothing: the conversation and its log are deleted, so there is nobody left to reconcile. docs/group-epoch-sync.md documents the log, the endpoints and the catch-up protocol. --- .../drizzle/0001_group_control_events.sql | 20 + apps/backend/drizzle/meta/0001_snapshot.json | 1669 +++++++++++++++++ apps/backend/drizzle/meta/_journal.json | 7 + .../__tests__/conversations.routes.test.ts | 209 +++ .../src/__tests__/groupControl.test.ts | 462 +++++ apps/backend/src/db/schema.ts | 74 + apps/backend/src/lib/eventEnvelope.ts | 3 + apps/backend/src/routes/conversations.ts | 238 ++- apps/backend/src/services/groupControl.ts | 282 +++ docs/group-epoch-sync.md | 155 ++ 10 files changed, 3110 insertions(+), 9 deletions(-) create mode 100644 apps/backend/drizzle/0001_group_control_events.sql create mode 100644 apps/backend/drizzle/meta/0001_snapshot.json create mode 100644 apps/backend/src/__tests__/groupControl.test.ts create mode 100644 apps/backend/src/services/groupControl.ts create mode 100644 docs/group-epoch-sync.md diff --git a/apps/backend/drizzle/0001_group_control_events.sql b/apps/backend/drizzle/0001_group_control_events.sql new file mode 100644 index 00000000..fefc847b --- /dev/null +++ b/apps/backend/drizzle/0001_group_control_events.sql @@ -0,0 +1,20 @@ +CREATE TYPE "public"."group_control_event_type" AS ENUM('member_added', 'member_removed', 'member_left', 'commit');--> statement-breakpoint +CREATE TABLE "group_control_events" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "conversation_id" uuid NOT NULL, + "sequence" integer NOT NULL, + "epoch" integer NOT NULL, + "event_type" "group_control_event_type" NOT NULL, + "actor_user_id" uuid, + "target_user_id" uuid, + "message_id" uuid, + "payload" text, + "created_at" timestamp DEFAULT now() NOT NULL +); +--> statement-breakpoint +ALTER TABLE "conversations" ADD COLUMN "epoch" integer DEFAULT 0 NOT NULL;--> statement-breakpoint +ALTER TABLE "group_control_events" ADD CONSTRAINT "group_control_events_conversation_id_conversations_id_fk" FOREIGN KEY ("conversation_id") REFERENCES "public"."conversations"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "group_control_events" ADD CONSTRAINT "group_control_events_actor_user_id_users_id_fk" FOREIGN KEY ("actor_user_id") REFERENCES "public"."users"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "group_control_events" ADD CONSTRAINT "group_control_events_target_user_id_users_id_fk" FOREIGN KEY ("target_user_id") REFERENCES "public"."users"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint +ALTER TABLE "group_control_events" ADD CONSTRAINT "group_control_events_message_id_messages_id_fk" FOREIGN KEY ("message_id") REFERENCES "public"."messages"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint +CREATE UNIQUE INDEX "group_control_conversation_sequence_idx" ON "group_control_events" USING btree ("conversation_id","sequence"); \ No newline at end of file diff --git a/apps/backend/drizzle/meta/0001_snapshot.json b/apps/backend/drizzle/meta/0001_snapshot.json new file mode 100644 index 00000000..5ec7534f --- /dev/null +++ b/apps/backend/drizzle/meta/0001_snapshot.json @@ -0,0 +1,1669 @@ +{ + "id": "2e17fd56-4cca-4b30-b479-2b9e98191e61", + "prevId": "d5682005-cccf-4e2e-992d-d66a5d6d3f4c", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.conversation_members": { + "name": "conversation_members", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "conversation_id": { + "name": "conversation_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "last_read_message_id": { + "name": "last_read_message_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "is_muted": { + "name": "is_muted", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "is_archived": { + "name": "is_archived", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "joined_at": { + "name": "joined_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "conversation_members_conversation_id_conversations_id_fk": { + "name": "conversation_members_conversation_id_conversations_id_fk", + "tableFrom": "conversation_members", + "tableTo": "conversations", + "columnsFrom": [ + "conversation_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "conversation_members_user_id_users_id_fk": { + "name": "conversation_members_user_id_users_id_fk", + "tableFrom": "conversation_members", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "conversation_members_last_read_message_id_messages_id_fk": { + "name": "conversation_members_last_read_message_id_messages_id_fk", + "tableFrom": "conversation_members", + "tableTo": "messages", + "columnsFrom": [ + "last_read_message_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.conversations": { + "name": "conversations", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "type": { + "name": "type", + "type": "conversation_type", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'dm'" + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "avatar_url": { + "name": "avatar_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "epoch": { + "name": "epoch", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.device_prekeys": { + "name": "device_prekeys", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "device_id": { + "name": "device_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "key_type": { + "name": "key_type", + "type": "prekey_type", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "key_id": { + "name": "key_id", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "public_key": { + "name": "public_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "signature": { + "name": "signature", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "consumed": { + "name": "consumed", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "device_prekeys_device_type_keyid_idx": { + "name": "device_prekeys_device_type_keyid_idx", + "columns": [ + { + "expression": "device_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "key_type", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "key_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "device_prekeys_signed_device_idx": { + "name": "device_prekeys_signed_device_idx", + "columns": [ + { + "expression": "device_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"device_prekeys\".\"key_type\" = 'signed'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "device_prekeys_one_time_available_idx": { + "name": "device_prekeys_one_time_available_idx", + "columns": [ + { + "expression": "device_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"device_prekeys\".\"key_type\" = 'one_time' AND \"device_prekeys\".\"consumed\" = false", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "device_prekeys_device_id_devices_id_fk": { + "name": "device_prekeys_device_id_devices_id_fk", + "tableFrom": "device_prekeys", + "tableTo": "devices", + "columnsFrom": [ + "device_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": { + "device_prekeys_signed_requires_signature": { + "name": "device_prekeys_signed_requires_signature", + "value": "\"device_prekeys\".\"key_type\" <> 'signed' OR \"device_prekeys\".\"signature\" IS NOT NULL" + } + }, + "isRLSEnabled": false + }, + "public.devices": { + "name": "devices", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "identity_public_key": { + "name": "identity_public_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "registration_id": { + "name": "registration_id", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "device_name": { + "name": "device_name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "platform": { + "name": "platform", + "type": "device_platform", + "typeSchema": "public", + "primaryKey": false, + "notNull": false + }, + "last_seen_at": { + "name": "last_seen_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "push_enabled": { + "name": "push_enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "revoked_at": { + "name": "revoked_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "devices_user_identity_idx": { + "name": "devices_user_identity_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "identity_public_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "devices_user_id_active_idx": { + "name": "devices_user_id_active_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"devices\".\"revoked_at\" IS NULL", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "devices_user_id_users_id_fk": { + "name": "devices_user_id_users_id_fk", + "tableFrom": "devices", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.files": { + "name": "files", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "uploader_id": { + "name": "uploader_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "conversation_id": { + "name": "conversation_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "file_status", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "size": { + "name": "size", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "mime_type": { + "name": "mime_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "sha256": { + "name": "sha256", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "storage_key": { + "name": "storage_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "is_thumbnail": { + "name": "is_thumbnail", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "hard_deleted_at": { + "name": "hard_deleted_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "files_uploader_id_users_id_fk": { + "name": "files_uploader_id_users_id_fk", + "tableFrom": "files", + "tableTo": "users", + "columnsFrom": [ + "uploader_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "files_conversation_id_conversations_id_fk": { + "name": "files_conversation_id_conversations_id_fk", + "tableFrom": "files", + "tableTo": "conversations", + "columnsFrom": [ + "conversation_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "files_storage_key_unique": { + "name": "files_storage_key_unique", + "nullsNotDistinct": false, + "columns": [ + "storage_key" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.group_control_events": { + "name": "group_control_events", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "conversation_id": { + "name": "conversation_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "sequence": { + "name": "sequence", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "epoch": { + "name": "epoch", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "event_type": { + "name": "event_type", + "type": "group_control_event_type", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "actor_user_id": { + "name": "actor_user_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "target_user_id": { + "name": "target_user_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "message_id": { + "name": "message_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "payload": { + "name": "payload", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "group_control_conversation_sequence_idx": { + "name": "group_control_conversation_sequence_idx", + "columns": [ + { + "expression": "conversation_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "sequence", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "group_control_events_conversation_id_conversations_id_fk": { + "name": "group_control_events_conversation_id_conversations_id_fk", + "tableFrom": "group_control_events", + "tableTo": "conversations", + "columnsFrom": [ + "conversation_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "group_control_events_actor_user_id_users_id_fk": { + "name": "group_control_events_actor_user_id_users_id_fk", + "tableFrom": "group_control_events", + "tableTo": "users", + "columnsFrom": [ + "actor_user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + }, + "group_control_events_target_user_id_users_id_fk": { + "name": "group_control_events_target_user_id_users_id_fk", + "tableFrom": "group_control_events", + "tableTo": "users", + "columnsFrom": [ + "target_user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + }, + "group_control_events_message_id_messages_id_fk": { + "name": "group_control_events_message_id_messages_id_fk", + "tableFrom": "group_control_events", + "tableTo": "messages", + "columnsFrom": [ + "message_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.message_envelopes": { + "name": "message_envelopes", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "message_id": { + "name": "message_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "recipient_device_id": { + "name": "recipient_device_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "recipient_user_id": { + "name": "recipient_user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "ciphertext": { + "name": "ciphertext", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "delivered_at": { + "name": "delivered_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "read_at": { + "name": "read_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "me_recipient_device_created_idx": { + "name": "me_recipient_device_created_idx", + "columns": [ + { + "expression": "recipient_device_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "me_message_idx": { + "name": "me_message_idx", + "columns": [ + { + "expression": "message_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "message_envelopes_message_id_messages_id_fk": { + "name": "message_envelopes_message_id_messages_id_fk", + "tableFrom": "message_envelopes", + "tableTo": "messages", + "columnsFrom": [ + "message_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "message_envelopes_recipient_device_id_devices_id_fk": { + "name": "message_envelopes_recipient_device_id_devices_id_fk", + "tableFrom": "message_envelopes", + "tableTo": "devices", + "columnsFrom": [ + "recipient_device_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "message_envelopes_recipient_user_id_users_id_fk": { + "name": "message_envelopes_recipient_user_id_users_id_fk", + "tableFrom": "message_envelopes", + "tableTo": "users", + "columnsFrom": [ + "recipient_user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.messages": { + "name": "messages", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "conversation_id": { + "name": "conversation_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "sender_id": { + "name": "sender_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "sender_device_id": { + "name": "sender_device_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "content_type": { + "name": "content_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'text'" + }, + "ciphertext": { + "name": "ciphertext", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "file_id": { + "name": "file_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "edits_message_id": { + "name": "edits_message_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "messages_conversation_created_idx": { + "name": "messages_conversation_created_idx", + "columns": [ + { + "expression": "conversation_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "messages_conversation_id_conversations_id_fk": { + "name": "messages_conversation_id_conversations_id_fk", + "tableFrom": "messages", + "tableTo": "conversations", + "columnsFrom": [ + "conversation_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "messages_sender_id_users_id_fk": { + "name": "messages_sender_id_users_id_fk", + "tableFrom": "messages", + "tableTo": "users", + "columnsFrom": [ + "sender_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "messages_sender_device_id_devices_id_fk": { + "name": "messages_sender_device_id_devices_id_fk", + "tableFrom": "messages", + "tableTo": "devices", + "columnsFrom": [ + "sender_device_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + }, + "messages_file_id_files_id_fk": { + "name": "messages_file_id_files_id_fk", + "tableFrom": "messages", + "tableTo": "files", + "columnsFrom": [ + "file_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + }, + "messages_edits_message_id_messages_id_fk": { + "name": "messages_edits_message_id_messages_id_fk", + "tableFrom": "messages", + "tableTo": "messages", + "columnsFrom": [ + "edits_message_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.proposal_votes": { + "name": "proposal_votes", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "treasury_proposal_id": { + "name": "treasury_proposal_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "vote": { + "name": "vote", + "type": "proposal_vote_type", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "signature": { + "name": "signature", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "proposal_votes_proposal_user_unique": { + "name": "proposal_votes_proposal_user_unique", + "columns": [ + { + "expression": "treasury_proposal_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "proposal_votes_treasury_proposal_id_treasury_proposals_id_fk": { + "name": "proposal_votes_treasury_proposal_id_treasury_proposals_id_fk", + "tableFrom": "proposal_votes", + "tableTo": "treasury_proposals", + "columnsFrom": [ + "treasury_proposal_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "proposal_votes_user_id_users_id_fk": { + "name": "proposal_votes_user_id_users_id_fk", + "tableFrom": "proposal_votes", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.push_subscriptions": { + "name": "push_subscriptions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "device_id": { + "name": "device_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "endpoint": { + "name": "endpoint", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "p256dh": { + "name": "p256dh", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "auth": { + "name": "auth", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "last_used_at": { + "name": "last_used_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "disabled_at": { + "name": "disabled_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "push_subscriptions_device_id_devices_id_fk": { + "name": "push_subscriptions_device_id_devices_id_fk", + "tableFrom": "push_subscriptions", + "tableTo": "devices", + "columnsFrom": [ + "device_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "push_subscriptions_endpoint_unique": { + "name": "push_subscriptions_endpoint_unique", + "nullsNotDistinct": false, + "columns": [ + "endpoint" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.token_transfers": { + "name": "token_transfers", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "conversation_id": { + "name": "conversation_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "sender_id": { + "name": "sender_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "recipient_address": { + "name": "recipient_address", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "amount": { + "name": "amount", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "token_contract_id": { + "name": "token_contract_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "tx_hash": { + "name": "tx_hash", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "memo": { + "name": "memo", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "token_transfers_conversation_id_conversations_id_fk": { + "name": "token_transfers_conversation_id_conversations_id_fk", + "tableFrom": "token_transfers", + "tableTo": "conversations", + "columnsFrom": [ + "conversation_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "token_transfers_sender_id_users_id_fk": { + "name": "token_transfers_sender_id_users_id_fk", + "tableFrom": "token_transfers", + "tableTo": "users", + "columnsFrom": [ + "sender_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "token_transfers_tx_hash_unique": { + "name": "token_transfers_tx_hash_unique", + "nullsNotDistinct": false, + "columns": [ + "tx_hash" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.treasury_proposals": { + "name": "treasury_proposals", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "contract_id": { + "name": "contract_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "proposal_id": { + "name": "proposal_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "conversation_id": { + "name": "conversation_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "status": { + "name": "status", + "type": "treasury_proposal_status", + "typeSchema": "public", + "primaryKey": false, + "notNull": true, + "default": "'active'" + }, + "approvals_count": { + "name": "approvals_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "rejections_count": { + "name": "rejections_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "recipient": { + "name": "recipient", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "amount": { + "name": "amount", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "token": { + "name": "token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "threshold": { + "name": "threshold", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 3 + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "treasury_proposals_contract_proposal_idx": { + "name": "treasury_proposals_contract_proposal_idx", + "columns": [ + { + "expression": "contract_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "proposal_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "treasury_proposals_conversation_id_conversations_id_fk": { + "name": "treasury_proposals_conversation_id_conversations_id_fk", + "tableFrom": "treasury_proposals", + "tableTo": "conversations", + "columnsFrom": [ + "conversation_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.users": { + "name": "users", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "username": { + "name": "username", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "avatar_url": { + "name": "avatar_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "presence_visible": { + "name": "presence_visible", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "send_read_receipts": { + "name": "send_read_receipts", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "users_username_unique": { + "name": "users_username_unique", + "nullsNotDistinct": false, + "columns": [ + "username" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.wallets": { + "name": "wallets", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "address": { + "name": "address", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "is_primary": { + "name": "is_primary", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "wallets_user_id_users_id_fk": { + "name": "wallets_user_id_users_id_fk", + "tableFrom": "wallets", + "tableTo": "users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "wallets_address_unique": { + "name": "wallets_address_unique", + "nullsNotDistinct": false, + "columns": [ + "address" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": { + "public.content_type": { + "name": "content_type", + "schema": "public", + "values": [ + "text", + "file", + "image", + "video", + "audio", + "system" + ] + }, + "public.conversation_type": { + "name": "conversation_type", + "schema": "public", + "values": [ + "dm", + "group" + ] + }, + "public.device_platform": { + "name": "device_platform", + "schema": "public", + "values": [ + "web", + "ios", + "android" + ] + }, + "public.file_status": { + "name": "file_status", + "schema": "public", + "values": [ + "pending", + "ready", + "deleted" + ] + }, + "public.group_control_event_type": { + "name": "group_control_event_type", + "schema": "public", + "values": [ + "member_added", + "member_removed", + "member_left", + "commit" + ] + }, + "public.prekey_type": { + "name": "prekey_type", + "schema": "public", + "values": [ + "signed", + "one_time" + ] + }, + "public.proposal_vote_type": { + "name": "proposal_vote_type", + "schema": "public", + "values": [ + "approve", + "reject" + ] + }, + "public.treasury_proposal_status": { + "name": "treasury_proposal_status", + "schema": "public", + "values": [ + "active", + "approved", + "rejected", + "executed", + "expired" + ] + } + }, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/apps/backend/drizzle/meta/_journal.json b/apps/backend/drizzle/meta/_journal.json index 52ed62d6..ee4945a9 100644 --- a/apps/backend/drizzle/meta/_journal.json +++ b/apps/backend/drizzle/meta/_journal.json @@ -8,6 +8,13 @@ "when": 1784899426825, "tag": "0000_stale_mandarin", "breakpoints": true + }, + { + "idx": 1, + "version": "7", + "when": 1785289957784, + "tag": "0001_group_control_events", + "breakpoints": true } ] } \ No newline at end of file diff --git a/apps/backend/src/__tests__/conversations.routes.test.ts b/apps/backend/src/__tests__/conversations.routes.test.ts index db1c9f44..3f8c5b1d 100644 --- a/apps/backend/src/__tests__/conversations.routes.test.ts +++ b/apps/backend/src/__tests__/conversations.routes.test.ts @@ -35,6 +35,33 @@ vi.mock('../lib/redis.js', () => ({ convCacheKey: (userId: string) => `conversations:${userId}`, })); +// Membership changes now run inside a transaction together with their +// group-control event (#369), so the mock db hands out a transaction handle +// that behaves like the top-level one. Sequencing itself is covered by +// groupControl.test.ts; here it is stubbed so these tests stay about routing. +const mockTransaction = vi.fn((fn: (tx: unknown) => unknown) => + fn({ delete: mockDelete, insert: mockInsert, update: mockUpdate }), +); + +const mockAppendGroupControlEvent = vi.fn(async () => ({ + event: { conversationId: 'conv-1', epoch: 1, sequence: 1, eventType: 'member_added' }, + systemMessage: null, +})); +const mockBroadcastGroupControlEvent = vi.fn(); +const mockGetGroupState = vi.fn(); +const mockReadGroupControlEvents = vi.fn(); + +vi.mock('../services/groupControl.js', () => ({ + appendGroupControlEvent: mockAppendGroupControlEvent, + broadcastGroupControlEvent: mockBroadcastGroupControlEvent, + getGroupState: mockGetGroupState, + readGroupControlEvents: mockReadGroupControlEvents, + serializeGroupControlEvent: (event: unknown) => event, + DEFAULT_GROUP_CONTROL_PAGE_SIZE: 100, + MAX_GROUP_CONTROL_PAGE_SIZE: 500, + MAX_GROUP_CONTROL_PAYLOAD_BYTES: 65536, +})); + vi.mock('../db/index.js', () => ({ db: { query: { @@ -44,6 +71,7 @@ vi.mock('../db/index.js', () => ({ delete: mockDelete, insert: mockInsert, update: mockUpdate, + transaction: mockTransaction, }, })); @@ -93,6 +121,12 @@ function makeApp() { beforeEach(() => { vi.clearAllMocks(); + // clearAllMocks keeps implementations, so restate the group-control default + // rather than letting one test's mockResolvedValue leak into the next. + mockAppendGroupControlEvent.mockResolvedValue({ + event: { conversationId: 'conv-1', epoch: 1, sequence: 1, eventType: 'member_added' }, + systemMessage: null, + }); }); describe('GET /conversations/:id', () => { @@ -295,7 +329,20 @@ describe('POST /conversations/:id/members', () => { conversationId: 'conv-1', userId: 'user-2', joinedAt: joinedAt.toISOString(), + // The join is sequenced, so the caller learns the new epoch (#369). + epoch: 1, + sequence: 1, }); + expect(mockAppendGroupControlEvent).toHaveBeenCalledWith( + expect.objectContaining({ + conversationId: 'conv-1', + eventType: 'member_added', + actorUserId: 'user-1', + targetUserId: 'user-2', + }), + expect.anything(), + ); + expect(mockBroadcastGroupControlEvent).toHaveBeenCalled(); }); }); @@ -442,5 +489,167 @@ describe('DELETE /conversations/:id/leave', () => { expect(res.status).toBe(204); expect(mockDelete).toHaveBeenCalledWith(conversationMembersTable); expect(deleteWhere).toHaveBeenCalled(); + expect(mockAppendGroupControlEvent).toHaveBeenCalledWith( + expect.objectContaining({ + conversationId: 'conv-1', + eventType: 'member_left', + actorUserId: 'user-1', + }), + expect.anything(), + ); + expect(mockBroadcastGroupControlEvent).toHaveBeenCalled(); + }); + + it('does not sequence an event when the conversation itself is deleted', async () => { + const deleteWhere = vi.fn().mockResolvedValue(undefined); + mockDelete.mockReturnValue({ where: deleteWhere }); + mockFindConversation.mockResolvedValue({ id: 'conv-1', type: 'group' }); + mockFindMember.mockResolvedValue({ id: 'member-1' }); + mockFindMany.mockResolvedValue([{ userId: 'user-1' }]); + + await request(makeApp()).delete('/conversations/conv-1/leave'); + + // Nothing survives to reconcile against, so no epoch bump is recorded. + expect(mockAppendGroupControlEvent).not.toHaveBeenCalled(); + }); +}); + +// ── Group control endpoints (#369) ─────────────────────────────────────────── + +describe('GET /conversations/:id/epoch', () => { + it('returns 403 for a non-member', async () => { + mockFindMember.mockResolvedValue(undefined); + + const res = await request(makeApp()).get('/conversations/conv-1/epoch'); + + expect(res.status).toBe(403); + }); + + it('reports the current epoch and latest sequence', async () => { + mockFindMember.mockResolvedValue({ id: 'member-1' }); + mockGetGroupState.mockResolvedValue({ epoch: 4, latestSequence: 7 }); + + const res = await request(makeApp()).get('/conversations/conv-1/epoch'); + + expect(res.status).toBe(200); + expect(res.body).toEqual({ conversationId: 'conv-1', epoch: 4, latestSequence: 7 }); + }); + + it('returns 404 when the conversation is gone', async () => { + mockFindMember.mockResolvedValue({ id: 'member-1' }); + mockGetGroupState.mockResolvedValue(null); + + const res = await request(makeApp()).get('/conversations/conv-1/epoch'); + + expect(res.status).toBe(404); + }); +}); + +describe('GET /conversations/:id/group-control', () => { + const events = [ + { id: 'evt-3', sequence: 3, epoch: 3, eventType: 'member_added' }, + { id: 'evt-4', sequence: 4, epoch: 4, eventType: 'commit' }, + ]; + + it('returns 403 for a non-member', async () => { + mockFindMember.mockResolvedValue(undefined); + + const res = await request(makeApp()).get('/conversations/conv-1/group-control'); + + expect(res.status).toBe(403); + }); + + it('AC1 — returns the missed events in order with a cursor for the next page', async () => { + mockFindMember.mockResolvedValue({ id: 'member-1' }); + mockGetGroupState.mockResolvedValue({ epoch: 5, latestSequence: 5 }); + mockReadGroupControlEvents.mockResolvedValue({ events, hasMore: true }); + + const res = await request(makeApp()).get( + '/conversations/conv-1/group-control?sinceSequence=2&limit=2', + ); + + expect(res.status).toBe(200); + expect(res.body.events.map((e: { sequence: number }) => e.sequence)).toEqual([3, 4]); + expect(res.body.currentEpoch).toBe(5); + expect(res.body.latestSequence).toBe(5); + expect(res.body.nextSequence).toBe(4); + expect(res.body.hasMore).toBe(true); + expect(mockReadGroupControlEvents).toHaveBeenCalledWith({ + conversationId: 'conv-1', + sinceSequence: 2, + limit: 2, + }); + }); + + it('keeps the cursor where it was when nothing new arrived', async () => { + mockFindMember.mockResolvedValue({ id: 'member-1' }); + mockGetGroupState.mockResolvedValue({ epoch: 5, latestSequence: 5 }); + mockReadGroupControlEvents.mockResolvedValue({ events: [], hasMore: false }); + + const res = await request(makeApp()).get('/conversations/conv-1/group-control?sinceSequence=5'); + + expect(res.body.nextSequence).toBe(5); + expect(res.body.hasMore).toBe(false); + }); + + it('rejects a negative or non-numeric cursor', async () => { + mockFindMember.mockResolvedValue({ id: 'member-1' }); + + for (const cursor of ['-1', 'abc']) { + const res = await request(makeApp()).get( + `/conversations/conv-1/group-control?sinceSequence=${cursor}`, + ); + expect(res.status).toBe(400); + } + }); +}); + +describe('POST /conversations/:id/group-control', () => { + it('rejects a missing or empty payload', async () => { + const res = await request(makeApp()).post('/conversations/conv-1/group-control').send({}); + + expect(res.status).toBe(400); + }); + + it('rejects a payload beyond the size cap', async () => { + const res = await request(makeApp()) + .post('/conversations/conv-1/group-control') + .send({ payload: 'x'.repeat(65537) }); + + expect(res.status).toBe(413); + }); + + it('returns 403 for a non-member', async () => { + mockFindMember.mockResolvedValue(undefined); + + const res = await request(makeApp()) + .post('/conversations/conv-1/group-control') + .send({ payload: 'opaque-commit' }); + + expect(res.status).toBe(403); + }); + + it('sequences a member commit and broadcasts it', async () => { + mockFindMember.mockResolvedValue({ id: 'member-1' }); + mockAppendGroupControlEvent.mockResolvedValue({ + event: { conversationId: 'conv-1', epoch: 6, sequence: 6, eventType: 'commit' }, + systemMessage: null, + }); + + const res = await request(makeApp()) + .post('/conversations/conv-1/group-control') + .send({ payload: 'opaque-commit' }); + + expect(res.status).toBe(201); + expect(res.body).toMatchObject({ epoch: 6, sequence: 6, eventType: 'commit' }); + expect(mockAppendGroupControlEvent).toHaveBeenCalledWith( + expect.objectContaining({ + conversationId: 'conv-1', + eventType: 'commit', + actorUserId: 'user-1', + payload: 'opaque-commit', + }), + ); + expect(mockBroadcastGroupControlEvent).toHaveBeenCalled(); }); }); diff --git a/apps/backend/src/__tests__/groupControl.test.ts b/apps/backend/src/__tests__/groupControl.test.ts new file mode 100644 index 00000000..24c7b6a9 --- /dev/null +++ b/apps/backend/src/__tests__/groupControl.test.ts @@ -0,0 +1,462 @@ +/** + * #369 — epoch sequencing, system events, and the catch-up path. + * + * The invariants under test: every group-control event bumps the epoch and + * takes the next gap-free sequence, a client can fetch what it missed in + * order, and two clients replaying the same log land on the same epoch. + */ +import { describe, it, expect, vi, beforeEach } from 'vitest'; + +// ── an in-memory stand-in for the two tables under test ───────────────────── + +interface ControlRow { + id: string; + conversationId: string; + sequence: number; + epoch: number; + eventType: string; + actorUserId: string | null; + targetUserId: string | null; + messageId: string | null; + payload: string | null; + createdAt: Date; +} + +const store = { + epochs: new Map(), + control: [] as ControlRow[], + messages: [] as Array>, +}; + +let nextId = 0; +const makeId = () => `id-${++nextId}`; + +/** + * `select()` serves two shapes: `await ...where(...)` for the max-sequence + * aggregate, and `...where(...).orderBy(...).limit(n)` for the ordered read. + * The returned builder is thenable so both work off one implementation. + */ +function selectBuilder() { + return { + from: (_table: unknown) => ({ + where: (where: { conversationId?: string }) => { + const conversationId = where.conversationId ?? currentQuery.conversationId; + const forConversation = store.control.filter( + (row) => row.conversationId === conversationId, + ); + const maxSequence = + forConversation.length > 0 ? Math.max(...forConversation.map((r) => r.sequence)) : 0; + + return { + orderBy: (_order: unknown) => ({ + limit: async (limit: number) => + forConversation + .filter((row) => row.sequence > currentQuery.sinceSequence) + .sort((a, b) => a.sequence - b.sequence) + .slice(0, limit), + }), + // Both aliases the service selects the aggregate under. + then: (resolve: (rows: unknown) => unknown, reject?: (err: unknown) => unknown) => + Promise.resolve([{ maxSequence, latestSequence: maxSequence }]).then(resolve, reject), + }; + }, + }), + }; +} + +/** Captures which table a chained builder is operating on. */ +function makeExecutor() { + return { + update: (_table: unknown) => ({ + set: (_values: unknown) => ({ + where: (where: { conversationId: string }) => ({ + returning: async () => { + const current = store.epochs.get(where.conversationId); + if (current === undefined) return []; + const next = current + 1; + store.epochs.set(where.conversationId, next); + return [{ epoch: next }]; + }, + }), + }), + }), + select: (_columns: unknown) => selectBuilder(), + insert: (table: { __name: string }) => ({ + values: (values: Record) => ({ + returning: async () => { + if (table.__name === 'messages') { + const row = { id: makeId(), createdAt: new Date(), ...values }; + store.messages.push(row); + return [row]; + } + const row = { + id: makeId(), + createdAt: new Date(), + ...values, + } as unknown as ControlRow; + store.control.push(row); + return [row]; + }, + }), + }), + }; +} + +const mockConversationFindFirst = vi.fn(); + +vi.mock('../db/index.js', () => ({ + db: { + query: { + conversations: { findFirst: mockConversationFindFirst }, + }, + transaction: async (fn: (tx: unknown) => unknown) => fn(makeExecutor()), + update: () => makeExecutor().update(null), + insert: (table: { __name: string }) => makeExecutor().insert(table), + select: () => selectBuilder(), + }, +})); + +/** The read path builds its filter through mocked drizzle helpers. */ +const currentQuery = { conversationId: '', sinceSequence: 0 }; + +vi.mock('../db/schema.js', () => ({ + conversations: { __name: 'conversations', id: 'id', epoch: 'epoch' }, + groupControlEvents: { + __name: 'group_control_events', + conversationId: 'conversationId', + sequence: 'sequence', + }, + messages: { __name: 'messages', id: 'id' }, +})); + +vi.mock('drizzle-orm', () => ({ + and: (...args: unknown[]) => Object.assign({}, ...args.filter(Boolean)), + asc: vi.fn(), + eq: (col: unknown, val: unknown) => { + if (col === 'conversationId' || col === 'id') { + currentQuery.conversationId = String(val); + return { conversationId: String(val) }; + } + return {}; + }, + gt: (col: unknown, val: unknown) => { + if (col === 'sequence') currentQuery.sinceSequence = Number(val); + return {}; + }, + sql: Object.assign( + (strings: TemplateStringsArray, ...values: unknown[]) => ({ strings, values }), + { raw: vi.fn() }, + ), +})); + +const mockEmit = vi.fn(); +const mockTo = vi.fn(() => ({ emit: mockEmit })); +let socketServer: unknown = { to: mockTo }; + +vi.mock('../lib/socket.js', () => ({ + getSocketServer: () => socketServer, +})); + +vi.mock('../services/roomManager.js', () => ({ + conversationRoom: (id: string) => `conv:${id}`, +})); + +const { + appendGroupControlEvent, + broadcastGroupControlEvent, + readGroupControlEvents, + getGroupState, + buildSystemEventBody, + serializeGroupControlEvent, +} = await import('../services/groupControl.js'); + +const CONV = 'conv-1'; + +beforeEach(() => { + vi.clearAllMocks(); + store.epochs = new Map([[CONV, 0]]); + store.control = []; + store.messages = []; + nextId = 0; + socketServer = { to: mockTo }; + currentQuery.conversationId = ''; + currentQuery.sinceSequence = 0; + mockConversationFindFirst.mockImplementation(async () => ({ epoch: store.epochs.get(CONV) })); +}); + +// ─── appending ──────────────────────────────────────────────────────────────── + +describe('appendGroupControlEvent', () => { + it('AC2 — bumps the epoch and records what changed', async () => { + const { event } = await appendGroupControlEvent({ + conversationId: CONV, + eventType: 'member_added', + actorUserId: 'user-1', + targetUserId: 'user-2', + }); + + expect(event).toMatchObject({ + conversationId: CONV, + sequence: 1, + epoch: 1, + eventType: 'member_added', + actorUserId: 'user-1', + targetUserId: 'user-2', + }); + expect(store.epochs.get(CONV)).toBe(1); + }); + + it('assigns gap-free sequences and a monotonic epoch across events', async () => { + const first = await appendGroupControlEvent({ + conversationId: CONV, + eventType: 'member_added', + actorUserId: 'user-1', + targetUserId: 'user-2', + }); + const second = await appendGroupControlEvent({ + conversationId: CONV, + eventType: 'member_left', + actorUserId: 'user-2', + targetUserId: 'user-2', + }); + const third = await appendGroupControlEvent({ + conversationId: CONV, + eventType: 'commit', + actorUserId: 'user-1', + payload: 'opaque-mls-commit', + }); + + expect([first, second, third].map((r) => r.event.sequence)).toEqual([1, 2, 3]); + expect([first, second, third].map((r) => r.event.epoch)).toEqual([1, 2, 3]); + }); + + it('AC2 — writes a content_type=system message describing the change', async () => { + const { systemMessage, event } = await appendGroupControlEvent({ + conversationId: CONV, + eventType: 'member_added', + actorUserId: 'user-1', + targetUserId: 'user-2', + }); + + expect(systemMessage).toMatchObject({ + conversationId: CONV, + senderId: 'user-1', + contentType: 'system', + }); + expect(JSON.parse(String(systemMessage!['ciphertext']))).toEqual({ + type: 'group_control', + eventType: 'member_added', + conversationId: CONV, + epoch: 1, + sequence: 1, + actorUserId: 'user-1', + targetUserId: 'user-2', + }); + // The control row and the timeline entry point at each other, so the two + // views of the change can never disagree. + expect(event.messageId).toBe(systemMessage!['id']); + }); + + it('stores a client commit payload verbatim without interpreting it', async () => { + const payload = JSON.stringify({ anything: 'the server does not parse' }); + + const { event } = await appendGroupControlEvent({ + conversationId: CONV, + eventType: 'commit', + actorUserId: 'user-1', + payload, + }); + + expect(event.payload).toBe(payload); + }); + + it('fails loudly when the conversation is gone rather than losing the event', async () => { + await expect( + appendGroupControlEvent({ + conversationId: 'missing-conversation', + eventType: 'commit', + actorUserId: 'user-1', + }), + ).rejects.toThrow(/not found/); + }); +}); + +// ─── broadcasting ───────────────────────────────────────────────────────────── + +describe('broadcastGroupControlEvent', () => { + it('AC2 — emits the system event, the epoch change and the timeline entry', async () => { + const appended = await appendGroupControlEvent({ + conversationId: CONV, + eventType: 'member_added', + actorUserId: 'user-1', + targetUserId: 'user-2', + }); + + broadcastGroupControlEvent(appended); + + // Both the optimized room and the plain conversation id, for compatibility. + expect(mockTo).toHaveBeenCalledWith(`conv:${CONV}`); + expect(mockTo).toHaveBeenCalledWith(CONV); + + const emitted = mockEmit.mock.calls.map(([name]) => name); + expect(emitted).toContain('group_system_event'); + expect(emitted).toContain('epoch_changed'); + expect(emitted).toContain('new_message'); + + expect(mockEmit).toHaveBeenCalledWith('epoch_changed', { + conversationId: CONV, + epoch: 1, + sequence: 1, + }); + }); + + it('is a no-op without a socket server, so the durable log still stands', async () => { + socketServer = null; + const appended = await appendGroupControlEvent({ + conversationId: CONV, + eventType: 'commit', + actorUserId: 'user-1', + }); + + expect(() => broadcastGroupControlEvent(appended)).not.toThrow(); + expect(mockEmit).not.toHaveBeenCalled(); + }); + + it('describes the change in one stable shape', async () => { + const { event } = await appendGroupControlEvent({ + conversationId: CONV, + eventType: 'member_left', + actorUserId: 'user-2', + targetUserId: 'user-2', + }); + + expect(buildSystemEventBody(event)).toEqual({ + type: 'group_control', + eventType: 'member_left', + conversationId: CONV, + epoch: 1, + sequence: 1, + actorUserId: 'user-2', + targetUserId: 'user-2', + }); + expect(serializeGroupControlEvent(event)).toMatchObject({ + sequence: 1, + epoch: 1, + eventType: 'member_left', + }); + }); +}); + +// ─── catch-up ───────────────────────────────────────────────────────────────── + +describe('AC1 — missed commits are retrievable in order', () => { + async function seed(count: number) { + for (let i = 0; i < count; i++) { + await appendGroupControlEvent({ + conversationId: CONV, + eventType: 'commit', + actorUserId: 'user-1', + payload: `commit-${i + 1}`, + }); + } + } + + it('returns everything after the cursor, oldest first', async () => { + await seed(5); + + const { events, hasMore } = await readGroupControlEvents({ + conversationId: CONV, + sinceSequence: 2, + }); + + expect(events.map((e) => e.sequence)).toEqual([3, 4, 5]); + expect(hasMore).toBe(false); + }); + + it('treats the cursor as exclusive, so replaying never re-applies an event', async () => { + await seed(3); + + const { events } = await readGroupControlEvents({ + conversationId: CONV, + sinceSequence: 3, + }); + + expect(events).toEqual([]); + }); + + it('returns the whole log for a client that has never synced', async () => { + await seed(3); + + const { events } = await readGroupControlEvents({ conversationId: CONV }); + + expect(events.map((e) => e.sequence)).toEqual([1, 2, 3]); + }); + + it('pages without dropping an event between pages', async () => { + await seed(5); + + const first = await readGroupControlEvents({ conversationId: CONV, limit: 2 }); + expect(first.events.map((e) => e.sequence)).toEqual([1, 2]); + expect(first.hasMore).toBe(true); + + const second = await readGroupControlEvents({ + conversationId: CONV, + sinceSequence: first.events[first.events.length - 1]!.sequence, + limit: 2, + }); + expect(second.events.map((e) => e.sequence)).toEqual([3, 4]); + + const third = await readGroupControlEvents({ + conversationId: CONV, + sinceSequence: second.events[second.events.length - 1]!.sequence, + limit: 2, + }); + expect(third.events.map((e) => e.sequence)).toEqual([5]); + expect(third.hasMore).toBe(false); + }); + + it('AC3 — a client that missed everything converges on the current epoch', async () => { + await seed(4); + + const state = await getGroupState(CONV); + const { events } = await readGroupControlEvents({ conversationId: CONV, sinceSequence: 0 }); + + // Replay in order, exactly as a catching-up client would. + let clientEpoch = 0; + let clientSequence = 0; + for (const event of events) { + expect(event.sequence).toBe(clientSequence + 1); // gap-free + clientSequence = event.sequence; + clientEpoch = event.epoch; + } + + expect(clientEpoch).toBe(state!.epoch); + expect(clientSequence).toBe(state!.latestSequence); + }); + + it('AC3 — a client that saw the live events lands on the same epoch as one that replayed', async () => { + await seed(3); + + const live = store.control[store.control.length - 1]!.epoch; + const { events } = await readGroupControlEvents({ conversationId: CONV, sinceSequence: 0 }); + const replayed = events[events.length - 1]!.epoch; + + expect(replayed).toBe(live); + }); + + it('reports where the group stands for an "am I behind?" check', async () => { + await seed(2); + + expect(await getGroupState(CONV)).toEqual({ epoch: 2, latestSequence: 2 }); + }); + + it('reports a fresh conversation as epoch zero with nothing to replay', async () => { + expect(await getGroupState(CONV)).toEqual({ epoch: 0, latestSequence: 0 }); + }); + + it('returns null for a conversation that does not exist', async () => { + mockConversationFindFirst.mockResolvedValue(undefined); + + expect(await getGroupState('missing-conversation')).toBeNull(); + }); +}); diff --git a/apps/backend/src/db/schema.ts b/apps/backend/src/db/schema.ts index 46e079c8..15440fd1 100644 --- a/apps/backend/src/db/schema.ts +++ b/apps/backend/src/db/schema.ts @@ -43,6 +43,10 @@ export const conversations = pgTable('conversations', { type: conversationTypeEnum('type').notNull().default('dm'), name: text('name'), avatarUrl: text('avatar_url'), + // Group epoch (#369). Incremented by every group-control event; the row is + // also the serialization point for sequencing those events, so a concurrent + // join and leave can never be assigned the same sequence number. + epoch: integer('epoch').notNull().default(0), createdAt: timestamp('created_at').notNull().defaultNow(), }); @@ -347,6 +351,65 @@ export const pushSubscriptions = pgTable('push_subscriptions', { export type PushSubscription = typeof pushSubscriptions.$inferSelect; export type NewPushSubscription = typeof pushSubscriptions.$inferInsert; +// ─── Group control messages (#369) ──────────────────────────────────────────── +// +// The ordered, gap-free log of everything that changes a group's membership or +// its epoch: joins, leaves, removals, and client-submitted MLS commits. +// +// Chat messages are ordered by (createdAt, id) — good enough for a timeline, +// but useless for group state, where a client that applies two commits in the +// wrong order derives a different key schedule and can no longer decrypt. +// Group control therefore gets its own strictly monotonic per-conversation +// `sequence`, and "catch up" is `sequence > mine`, which cannot silently skip +// an event the way a timestamp cursor can. +// +// `epoch` is the value *after* the event was applied, so a client can compare +// its own epoch against the newest row and know exactly how far behind it is. +// The `messageId` links to the `content_type='system'` message emitted for the +// same event, so the timeline and the control log never disagree. +// +// `payload` is opaque to the server: an MLS commit or welcome blob, stored and +// relayed byte-for-byte. The server sequences group control, it does not +// interpret it. + +export const groupControlEventTypeEnum = pgEnum('group_control_event_type', [ + 'member_added', + 'member_removed', + 'member_left', + 'commit', +]); + +export type GroupControlEventType = (typeof groupControlEventTypeEnum.enumValues)[number]; + +export const groupControlEvents = pgTable( + 'group_control_events', + { + id: uuid('id').primaryKey().defaultRandom(), + conversationId: uuid('conversation_id') + .notNull() + .references(() => conversations.id, { onDelete: 'cascade' }), + /** Strictly increasing from 1, gap-free within a conversation. */ + sequence: integer('sequence').notNull(), + /** Group epoch after this event was applied. */ + epoch: integer('epoch').notNull(), + eventType: groupControlEventTypeEnum('event_type').notNull(), + actorUserId: uuid('actor_user_id').references(() => users.id, { onDelete: 'set null' }), + targetUserId: uuid('target_user_id').references(() => users.id, { onDelete: 'set null' }), + /** The system message emitted for this event, when one was created. */ + messageId: uuid('message_id').references(() => messages.id, { onDelete: 'set null' }), + /** Opaque client-supplied MLS commit/welcome material. Never parsed here. */ + payload: text('payload'), + createdAt: timestamp('created_at').notNull().defaultNow(), + }, + (table) => [ + // Both the gap-free guarantee and the catch-up read path. + uniqueIndex('group_control_conversation_sequence_idx').on(table.conversationId, table.sequence), + ], +); + +export type GroupControlEvent = typeof groupControlEvents.$inferSelect; +export type NewGroupControlEvent = typeof groupControlEvents.$inferInsert; + // ─── Relations ──────────────────────────────────────────────────────────────── export const usersRelations = relations(users, ({ many }) => ({ @@ -368,6 +431,17 @@ export const conversationsRelations = relations(conversations, ({ many }) => ({ transfers: many(tokenTransfers), treasuryProposals: many(treasuryProposals), files: many(files), + groupControlEvents: many(groupControlEvents), +})); + +export const groupControlEventsRelations = relations(groupControlEvents, ({ one }) => ({ + conversation: one(conversations, { + fields: [groupControlEvents.conversationId], + references: [conversations.id], + }), + actor: one(users, { fields: [groupControlEvents.actorUserId], references: [users.id] }), + target: one(users, { fields: [groupControlEvents.targetUserId], references: [users.id] }), + message: one(messages, { fields: [groupControlEvents.messageId], references: [messages.id] }), })); export const filesRelations = relations(files, ({ one, many }) => ({ diff --git a/apps/backend/src/lib/eventEnvelope.ts b/apps/backend/src/lib/eventEnvelope.ts index 7dcdd955..ca243693 100644 --- a/apps/backend/src/lib/eventEnvelope.ts +++ b/apps/backend/src/lib/eventEnvelope.ts @@ -25,6 +25,9 @@ export const KNOWN_EVENT_TYPES = new Set([ 'ephemeral_replay', 'resume_complete', 'device_envelope', + // Group epoch + membership reconciliation (#369). + 'group_system_event', + 'epoch_changed', 'error', ]); diff --git a/apps/backend/src/routes/conversations.ts b/apps/backend/src/routes/conversations.ts index 1d758f6c..39199535 100644 --- a/apps/backend/src/routes/conversations.ts +++ b/apps/backend/src/routes/conversations.ts @@ -16,6 +16,16 @@ import { invalidateConversationCaches } from '../lib/conversationCache.js'; import { serializeMessage } from '../lib/messages.js'; import { getSocketServer } from '../lib/socket.js'; import { MAX_MESSAGES_LIMIT, DEFAULT_MESSAGES_LIMIT } from '../constants.js'; +import { + DEFAULT_GROUP_CONTROL_PAGE_SIZE, + MAX_GROUP_CONTROL_PAGE_SIZE, + MAX_GROUP_CONTROL_PAYLOAD_BYTES, + appendGroupControlEvent, + broadcastGroupControlEvent, + getGroupState, + readGroupControlEvents, + serializeGroupControlEvent, +} from '../services/groupControl.js'; export const conversationsRouter: IRouter = Router(); @@ -313,16 +323,40 @@ conversationsRouter.post('/:id/members', async (req: AuthRequest, res) => { } try { - const [newMembership] = await db - .insert(conversationMembers) - .values({ conversationId, userId: newUserId }) - .returning(); + // The membership row and its group-control event are written together + // (#369). A member committed without the epoch bump that announces them + // would leave every other client unaware of someone who can now decrypt — + // exactly the divergence the control log exists to prevent. + const result = await db.transaction(async (tx) => { + const [newMembership] = await tx + .insert(conversationMembers) + .values({ conversationId, userId: newUserId }) + .returning(); + + if (!newMembership) { + return null; + } + + const appended = await appendGroupControlEvent( + { + conversationId, + eventType: 'member_added', + actorUserId: requesterId, + targetUserId: newUserId, + }, + tx, + ); + + return { newMembership, appended }; + }); - if (!newMembership) { + if (!result) { res.status(500).json({ error: 'Failed to add conversation member' }); return; } + const { newMembership, appended } = result; + const members = await db.query.conversationMembers.findMany({ where: eq(conversationMembers.conversationId, conversationId), columns: { userId: true }, @@ -335,11 +369,17 @@ conversationsRouter.post('/:id/members', async (req: AuthRequest, res) => { conversationId, }); + // Fanned out only once the transaction has committed, so a client that + // reacts to the event always finds the member already present. + broadcastGroupControlEvent(appended); + res.status(201).json({ id: newMembership.id, conversationId: newMembership.conversationId, userId: newMembership.userId, joinedAt: newMembership.joinedAt, + epoch: appended.event.epoch, + sequence: appended.event.sequence, }); } catch { res.status(409).json({ error: 'Database conflict or validation error' }); @@ -725,10 +765,21 @@ conversationsRouter.delete('/:id/leave', async (req: AuthRequest, res) => { columns: { userId: true }, }); - if (members.length === 1) { + const isLastMember = members.length === 1; + + if (isLastMember) { + // The conversation row — and with it the whole control log — goes away, + // so there is nobody left to reconcile and nothing to reconcile against. await db.delete(conversations).where(eq(conversations.id, conversationId)); - } else { - await db + await invalidateConversationCaches(members.map((member) => member.userId)); + res.status(204).send(); + return; + } + + // Departure and its epoch bump commit together, so remaining members can + // never observe a membership set that no control event accounts for (#369). + const appended = await db.transaction(async (tx) => { + await tx .delete(conversationMembers) .where( and( @@ -736,13 +787,182 @@ conversationsRouter.delete('/:id/leave', async (req: AuthRequest, res) => { eq(conversationMembers.userId, userId), ), ); - } + + return appendGroupControlEvent( + { + conversationId, + eventType: 'member_left', + actorUserId: userId, + targetUserId: userId, + }, + tx, + ); + }); await invalidateConversationCaches(members.map((member) => member.userId)); + broadcastGroupControlEvent(appended); + res.status(204).send(); }); +// ── Group control log (#369) ───────────────────────────────────────────────── +// +// The ordered sequence of everything that changed group membership or the +// epoch. A client that missed commits — offline, or reconnected mid-change — +// replays from its last applied sequence and converges on the current epoch. + +// GET /conversations/:id/epoch — cheap "am I behind?" check. +conversationsRouter.get('/:id/epoch', async (req: AuthRequest, res) => { + const userId = req.auth!.userId; + const conversationId = req.params['id'] as string | undefined; + + if (!conversationId) { + res.status(400).json({ error: 'Conversation id is required' }); + return; + } + + const membership = await db.query.conversationMembers.findFirst({ + where: and( + eq(conversationMembers.conversationId, conversationId), + eq(conversationMembers.userId, userId), + ), + }); + + if (!membership) { + res.status(403).json({ error: 'Not a member of this conversation' }); + return; + } + + const state = await getGroupState(conversationId); + + if (!state) { + res.status(404).json({ error: 'Conversation not found' }); + return; + } + + res.json({ conversationId, ...state }); +}); + +// GET /conversations/:id/group-control?sinceSequence=&limit= +// Ordered, gap-free catch-up. `sinceSequence` is exclusive, so replaying with +// the same cursor never re-applies an event the client already has. +conversationsRouter.get('/:id/group-control', async (req: AuthRequest, res) => { + const userId = req.auth!.userId; + const conversationId = req.params['id'] as string | undefined; + + if (!conversationId) { + res.status(400).json({ error: 'Conversation id is required' }); + return; + } + + const rawSince = req.query['sinceSequence']; + const sinceSequence = rawSince === undefined ? 0 : Number.parseInt(String(rawSince), 10); + + if (!Number.isFinite(sinceSequence) || sinceSequence < 0) { + res.status(400).json({ error: 'sinceSequence must be a non-negative integer' }); + return; + } + + const rawLimit = Number.parseInt(req.query['limit'] as string, 10); + const limit = + Number.isFinite(rawLimit) && rawLimit > 0 + ? Math.min(rawLimit, MAX_GROUP_CONTROL_PAGE_SIZE) + : DEFAULT_GROUP_CONTROL_PAGE_SIZE; + + const membership = await db.query.conversationMembers.findFirst({ + where: and( + eq(conversationMembers.conversationId, conversationId), + eq(conversationMembers.userId, userId), + ), + }); + + if (!membership) { + res.status(403).json({ error: 'Not a member of this conversation' }); + return; + } + + const state = await getGroupState(conversationId); + + if (!state) { + res.status(404).json({ error: 'Conversation not found' }); + return; + } + + const { events, hasMore } = await readGroupControlEvents({ + conversationId, + sinceSequence, + limit, + }); + + const lastSequence = events[events.length - 1]?.sequence ?? sinceSequence; + + res.json({ + conversationId, + // Where the group is now, so a client knows whether this page finished + // the catch-up even before it looks at `hasMore`. + currentEpoch: state.epoch, + latestSequence: state.latestSequence, + events: events.map(serializeGroupControlEvent), + // Feed straight back as `sinceSequence` for the next page. + nextSequence: lastSequence, + hasMore, + }); +}); + +// POST /conversations/:id/group-control — submit an MLS commit for sequencing. +// The payload is opaque: the server orders group control, it does not +// interpret it. +conversationsRouter.post('/:id/group-control', async (req: AuthRequest, res) => { + const userId = req.auth!.userId; + const conversationId = req.params['id'] as string | undefined; + + if (!conversationId) { + res.status(400).json({ error: 'Conversation id is required' }); + return; + } + + const { payload } = req.body as { payload?: unknown }; + + if (typeof payload !== 'string' || payload.length === 0) { + res.status(400).json({ error: 'payload must be a non-empty string' }); + return; + } + + if (Buffer.byteLength(payload, 'utf8') > MAX_GROUP_CONTROL_PAYLOAD_BYTES) { + res.status(413).json({ + error: `payload exceeds ${MAX_GROUP_CONTROL_PAYLOAD_BYTES} bytes`, + }); + return; + } + + const membership = await db.query.conversationMembers.findFirst({ + where: and( + eq(conversationMembers.conversationId, conversationId), + eq(conversationMembers.userId, userId), + ), + }); + + if (!membership) { + res.status(403).json({ error: 'Not a member of this conversation' }); + return; + } + + try { + const appended = await appendGroupControlEvent({ + conversationId, + eventType: 'commit', + actorUserId: userId, + payload, + }); + broadcastGroupControlEvent(appended); + + res.status(201).json(serializeGroupControlEvent(appended.event)); + } catch { + res.status(500).json({ error: 'Failed to append group control event' }); + } +}); + // ── GET /conversations/:id/devices ───────────────────────────────────────────── // Returns the full active (non-revoked) device set for all members of a // conversation. The web client calls this before encrypting a message so it diff --git a/apps/backend/src/services/groupControl.ts b/apps/backend/src/services/groupControl.ts new file mode 100644 index 00000000..e228c3f5 --- /dev/null +++ b/apps/backend/src/services/groupControl.ts @@ -0,0 +1,282 @@ +/** + * Group epoch sequencing and system events (#369). + * + * Group state — who is a member, which epoch's keys are current — is only + * usable if every client applies the same changes in the same order. Chat + * messages are ordered by `(createdAt, id)`, which is fine for a timeline but + * not for group control: two clients that apply a join and a leave in + * different orders derive different state, and a timestamp cursor can silently + * skip an event written slightly out of clock order. + * + * So group control gets its own log with a strictly monotonic, gap-free + * `sequence` per conversation. A client that missed commits asks for + * everything after the last sequence it applied and replays in order; "am I + * behind?" is an integer comparison, not a guess. + * + * Serialization: `conversations.epoch` is bumped with an `UPDATE ... RETURNING` + * inside the same transaction that assigns the sequence. That update takes a + * row lock on the conversation, so a concurrent join and leave are forced into + * a real order rather than racing for the same sequence number — the unique + * index on `(conversationId, sequence)` is the backstop, not the mechanism. + * + * Every event also persists a `content_type='system'` message so the change + * appears in the conversation timeline, and is fanned out live as + * `group_system_event` + `epoch_changed`. + */ +import { and, asc, eq, gt, sql } from 'drizzle-orm'; +import { db } from '../db/index.js'; +import { + conversations, + groupControlEvents, + messages, + type GroupControlEvent, + type GroupControlEventType, + type Message, +} from '../db/schema.js'; +import { getSocketServer } from '../lib/socket.js'; +import { conversationRoom } from './roomManager.js'; + +/** Maximum size of an opaque client-submitted commit payload. */ +export const MAX_GROUP_CONTROL_PAYLOAD_BYTES = 64 * 1024; + +/** Default and maximum page sizes for the catch-up endpoint. */ +export const DEFAULT_GROUP_CONTROL_PAGE_SIZE = 100; +export const MAX_GROUP_CONTROL_PAGE_SIZE = 500; + +export interface AppendGroupControlInput { + conversationId: string; + eventType: GroupControlEventType; + actorUserId?: string | null; + targetUserId?: string | null; + /** Opaque MLS material. Never inspected. */ + payload?: string | null; +} + +/** + * The body of the `content_type='system'` message written for an event. Kept + * as one shape so a client can parse any system message the same way, and + * deliberately free of anything private — it is stored unencrypted, exactly + * like the existing device-change system messages. + */ +export interface GroupSystemEventBody { + type: 'group_control'; + eventType: GroupControlEventType; + conversationId: string; + epoch: number; + sequence: number; + actorUserId: string | null; + targetUserId: string | null; +} + +export function buildSystemEventBody(event: GroupControlEvent): GroupSystemEventBody { + return { + type: 'group_control', + eventType: event.eventType, + conversationId: event.conversationId, + epoch: event.epoch, + sequence: event.sequence, + actorUserId: event.actorUserId, + targetUserId: event.targetUserId, + }; +} + +export interface AppendedGroupControl { + event: GroupControlEvent; + /** The timeline entry written for the event, when the event had an actor. */ + systemMessage: Message | null; +} + +/** The transaction handle drizzle hands to a `db.transaction` callback. */ +export type GroupControlTx = Parameters[0]>[0]; + +/** + * Append one group-control event: bump the epoch, assign the next sequence, + * write the system message, and link the two. + * + * Pass `existingTx` to run inside a caller's transaction — membership routes + * do, so a join either lands with its epoch bump or not at all. A membership + * row committed without its control event would leave every other client + * permanently unaware of a member who can now decrypt, which is exactly the + * divergence this log exists to prevent. + */ +export async function appendGroupControlEvent( + input: AppendGroupControlInput, + existingTx?: GroupControlTx, +): Promise { + const { conversationId, eventType } = input; + const actorUserId = input.actorUserId ?? null; + const targetUserId = input.targetUserId ?? null; + const payload = input.payload ?? null; + + const run = async (tx: GroupControlTx): Promise => { + // Row-locks the conversation for the rest of the transaction, which is + // what makes the sequence assignment below safe under concurrency. + const [updated] = await tx + .update(conversations) + .set({ epoch: sql`${conversations.epoch} + 1` }) + .where(eq(conversations.id, conversationId)) + .returning({ epoch: conversations.epoch }); + + if (!updated) { + throw new Error(`Conversation ${conversationId} not found`); + } + + const [maxRow] = await tx + .select({ maxSequence: sql`coalesce(max(${groupControlEvents.sequence}), 0)` }) + .from(groupControlEvents) + .where(eq(groupControlEvents.conversationId, conversationId)); + + const sequence = Number(maxRow?.maxSequence ?? 0) + 1; + + // The system message is written first so the control row can point at it; + // an actor-less event (a server-driven change) still needs a sender, so + // those are not given a timeline entry. + let systemMessage: Message | null = null; + if (actorUserId) { + const [inserted] = await tx + .insert(messages) + .values({ + conversationId, + senderId: actorUserId, + contentType: 'system', + ciphertext: JSON.stringify({ + type: 'group_control', + eventType, + conversationId, + epoch: updated.epoch, + sequence, + actorUserId, + targetUserId, + } satisfies GroupSystemEventBody), + }) + .returning(); + systemMessage = inserted ?? null; + } + + const [event] = await tx + .insert(groupControlEvents) + .values({ + conversationId, + sequence, + epoch: updated.epoch, + eventType, + actorUserId, + targetUserId, + messageId: systemMessage?.id ?? null, + payload, + }) + .returning(); + + if (!event) { + throw new Error('Failed to append group control event'); + } + + return { event, systemMessage }; + }; + + return existingTx ? run(existingTx) : db.transaction(run); +} + +/** + * Fan an appended event out to everyone currently connected. Best-effort and + * deliberately after the transaction commits: the durable log is the source of + * truth, and a client that misses the live event catches up through + * `readGroupControlEvents`. + */ +export function broadcastGroupControlEvent({ event, systemMessage }: AppendedGroupControl): void { + const io = getSocketServer(); + if (!io) return; + + const body = buildSystemEventBody(event); + // Both the optimized fan-out room and the plain conversation id, matching + // how the rest of the gateway emits for backward compatibility. + const rooms = [conversationRoom(event.conversationId), event.conversationId]; + + for (const room of rooms) { + io.to(room).emit('group_system_event', { id: event.id, ...body, createdAt: event.createdAt }); + io.to(room).emit('epoch_changed', { + conversationId: event.conversationId, + epoch: event.epoch, + sequence: event.sequence, + }); + // Existing clients render the timeline from `new_message`, so the system + // entry has to arrive on that channel too. + if (systemMessage) { + io.to(room).emit('new_message', systemMessage); + } + } +} + +/** + * Ordered catch-up read. Returns every event after `sinceSequence`, oldest + * first — the order a client must replay them in. `sinceSequence` is + * exclusive, so re-issuing the same cursor never re-delivers an applied event. + */ +export async function readGroupControlEvents({ + conversationId, + sinceSequence = 0, + limit = DEFAULT_GROUP_CONTROL_PAGE_SIZE, +}: { + conversationId: string; + sinceSequence?: number; + limit?: number; +}): Promise<{ events: GroupControlEvent[]; hasMore: boolean }> { + const pageSize = Math.min(Math.max(1, limit), MAX_GROUP_CONTROL_PAGE_SIZE); + + const rows = await db + .select() + .from(groupControlEvents) + .where( + and( + eq(groupControlEvents.conversationId, conversationId), + gt(groupControlEvents.sequence, sinceSequence), + ), + ) + .orderBy(asc(groupControlEvents.sequence)) + .limit(pageSize + 1); + + const hasMore = rows.length > pageSize; + + return { events: hasMore ? rows.slice(0, pageSize) : rows, hasMore }; +} + +/** + * Where the conversation currently stands. `latestSequence` is what a client + * compares its own cursor against to decide whether it needs to catch up. + */ +export async function getGroupState( + conversationId: string, +): Promise<{ epoch: number; latestSequence: number } | null> { + const conversation = await db.query.conversations.findFirst({ + where: eq(conversations.id, conversationId), + columns: { epoch: true }, + }); + + if (!conversation) return null; + + const [row] = await db + .select({ latestSequence: sql`coalesce(max(${groupControlEvents.sequence}), 0)` }) + .from(groupControlEvents) + .where(eq(groupControlEvents.conversationId, conversationId)); + + return { + epoch: conversation.epoch, + latestSequence: Number(row?.latestSequence ?? 0), + }; +} + +/** Shape returned to clients. Keeps the wire format in one place. */ +export function serializeGroupControlEvent(event: GroupControlEvent) { + return { + id: event.id, + conversationId: event.conversationId, + sequence: event.sequence, + epoch: event.epoch, + eventType: event.eventType, + actorUserId: event.actorUserId, + targetUserId: event.targetUserId, + messageId: event.messageId, + payload: event.payload, + createdAt: event.createdAt, + }; +} diff --git a/docs/group-epoch-sync.md b/docs/group-epoch-sync.md new file mode 100644 index 00000000..f9979074 --- /dev/null +++ b/docs/group-epoch-sync.md @@ -0,0 +1,155 @@ +# Group epoch sync and system events + +Group state — who is a member, whose keys are current — is only usable if every +client applies the same changes in the same order. This document describes the +ordered group-control log that makes that true, and how a client that missed +commits catches up. + +Implementation: `apps/backend/src/services/groupControl.ts`, the +`group_control_events` table and `conversations.epoch`, migration +`0001_group_control_events.sql`. + +## Why a separate log + +Chat messages are ordered by `(createdAt, id)`. That is fine for a timeline, +but wrong for group control: + +- A client that applies a join and a leave in the wrong order derives a + different key schedule and can no longer decrypt. +- A timestamp cursor can silently _skip_ an event written slightly out of clock + order — and a skipped membership change is indistinguishable, to the client, + from no change at all. + +So group control gets its own log with a strictly monotonic, gap-free +`sequence` per conversation. "Am I behind?" becomes an integer comparison, and +"catch me up" becomes `sequence > mine`, which cannot skip. + +## The log + +| Column | Meaning | +| ------------------------------ | --------------------------------------------------------------- | +| `sequence` | Strictly increasing from 1, gap-free within a conversation | +| `epoch` | The group epoch **after** this event was applied | +| `eventType` | `member_added`, `member_removed`, `member_left`, `commit` | +| `actorUserId` / `targetUserId` | Who made the change, and who it was about | +| `messageId` | The `content_type='system'` message emitted for the same event | +| `payload` | Opaque client-supplied MLS material, never parsed by the server | + +Because `epoch` is the value _after_ the event, a client compares its own epoch +against the newest row and knows exactly how far behind it is. + +### Ordering under concurrency + +`conversations.epoch` is bumped with `UPDATE ... RETURNING` inside the same +transaction that assigns the sequence. That update takes a row lock on the +conversation, so a concurrent join and leave are forced into a real order +rather than racing for the same sequence number. The unique index on +`(conversationId, sequence)` is the backstop, not the mechanism. + +### Atomicity with the membership change + +The membership row and its control event are written in one transaction. A +member committed without the epoch bump that announces them would leave every +other client unaware of someone who can now decrypt — precisely the divergence +this log exists to prevent. The live broadcast happens only after the commit, +so a client reacting to the event always finds the membership already in place. + +## System events + +Every control event also persists a `content_type='system'` message, so the +change appears in the conversation timeline alongside chat. Its body is one +stable shape: + +```json +{ + "type": "group_control", + "eventType": "member_added", + "conversationId": "…", + "epoch": 4, + "sequence": 7, + "actorUserId": "…", + "targetUserId": "…" +} +``` + +Like the existing device-change system messages, this is stored unencrypted and +contains no private content — only who changed what, and the resulting epoch. + +Live, each event is fanned out to the conversation room as: + +- `group_system_event` — the full control event. +- `epoch_changed` — `{ conversationId, epoch, sequence }`, for clients that + only need to know they must reconcile. +- `new_message` — the system message, so existing timeline rendering picks it + up with no client change. + +Both the optimized fan-out room and the plain conversation id receive them, in +line with how the rest of the gateway emits. + +## Catching up + +### Am I behind? + +``` +GET /conversations/:id/epoch +→ { "conversationId": "…", "epoch": 4, "latestSequence": 7 } +``` + +Compare `latestSequence` against the last sequence you applied. + +### Fetch what you missed, in order + +``` +GET /conversations/:id/group-control?sinceSequence=&limit= + +→ { + "conversationId": "…", + "currentEpoch": 4, + "latestSequence": 7, + "events": [ … ], // ascending by sequence — the order to replay in + "nextSequence": 7, // feed straight back as sinceSequence + "hasMore": false + } +``` + +`sinceSequence` is **exclusive**, so replaying with the same cursor never +re-applies an event you already have. Omit it to fetch the whole log — the path +a client takes on first sync or after a long absence. Page size defaults to 100 +and is clamped to 500. + +`currentEpoch` and `latestSequence` describe where the group is _now_, so a +client can tell whether this page finished the catch-up even before it looks at +`hasMore`. + +The result: a client that replays the log from 0 and a client that saw every +event live end up on the same epoch, because both applied the same events in +the same order. + +### Submitting a commit + +``` +POST /conversations/:id/group-control +{ "payload": "" } + +→ 201 { "sequence": 8, "epoch": 5, "eventType": "commit", … } +``` + +Members only. The payload is opaque to the server — it is stored and relayed +byte for byte, capped at 64 KiB. The server sequences group control; it does +not interpret it. + +## Membership changes that emit events + +| Route | Event | +| --------------------------------------- | -------------- | +| `POST /conversations/:id/members` | `member_added` | +| `DELETE /conversations/:id/leave` | `member_left` | +| `POST /conversations/:id/group-control` | `commit` | + +The response to a join now also carries the resulting `epoch` and `sequence`, +so the caller does not need a follow-up request to learn where the group +landed. + +One case emits nothing: the **last** member leaving. That deletes the +conversation, and the control log with it, so there is nobody left to reconcile +and nothing to reconcile against.