wip attempt better sync history

compose.yaml

@@ -261,6 +261,12 @@ services:
     build:
       context: ./src/frontend/servers/collaboration
       dockerfile: Dockerfile
+    # Bind-mount the relay source so edits on the host hot-reload through
+    # `tsx watch` without rebuilding the image. The host's node_modules
+    # (populated by `npm install` in the relay dir) is used directly.
+    volumes:
+      - ./src/frontend/servers/collaboration:/app
+    command: ["npx", "tsx", "watch", "relay.ts"]
     ports:
       - "4100:4100"
     environment:

src/backend/core/api/utils.py

@@ -101,8 +101,13 @@ def generate_s3_authorization_headers(key):
     return request
 
 
-def generate_upload_policy_for_key(key, content_type=None):
-    """Generate a presigned S3 PUT URL for a given key."""
+def generate_upload_policy_for_key(key, content_type=None, metadata=None):
+    """Generate a presigned S3 PUT URL for a given key.
+
+    When `metadata` is provided, each key is signed as an `x-amz-meta-*`
+    header. The client MUST send the corresponding headers verbatim on the
+    PUT request or S3 will reject the signature.
+    """
     if settings.AWS_S3_DOMAIN_REPLACE:
         s3_client = boto3.client(
             "s3",
@@ -120,6 +125,8 @@ def generate_upload_policy_for_key(key, content_type=None):
     params = {"Bucket": default_storage.bucket_name, "Key": key, "ACL": "private"}
     if content_type:
         params["ContentType"] = content_type
+    if metadata:
+        params["Metadata"] = {str(k): str(v) for k, v in metadata.items()}
 
     return s3_client.generate_presigned_url(
         ClientMethod="put_object",

src/backend/core/api/viewsets.py

@@ -1508,12 +1508,41 @@ class ItemViewSet(
         else:
             key = item.file_key
 
+        # epoch_ms is mandatory: the snapshot epoch anchors the relay history
+        # replay model. Missing or invalid values would break joiners that
+        # rely on x-amz-meta-epoch to decide which events to replay, so we
+        # refuse the upload rather than silently degrading.
+        epoch_ms = request.data.get("epoch_ms")
+        if epoch_ms is None:
+            return drf.response.Response(
+                {"detail": _("epoch_ms is required.")},
+                status=drf.status.HTTP_400_BAD_REQUEST,
+            )
+        try:
+            epoch_int = int(epoch_ms)
+        except (TypeError, ValueError):
+            return drf.response.Response(
+                {"detail": _("epoch_ms must be an integer.")},
+                status=drf.status.HTTP_400_BAD_REQUEST,
+            )
+        if epoch_int <= 0:
+            return drf.response.Response(
+                {"detail": _("epoch_ms must be a positive integer.")},
+                status=drf.status.HTTP_400_BAD_REQUEST,
+            )
+
         upload_url = utils.generate_upload_policy_for_key(
-            key, content_type="application/octet-stream"
+            key,
+            content_type="application/octet-stream",
+            metadata={"epoch": str(epoch_int)},
         )
 
         return drf.response.Response(
-            {"upload_url": upload_url, "filename": new_filename or item.filename},
+            {
+                "upload_url": upload_url,
+                "filename": new_filename or item.filename,
+                "required_headers": {"x-amz-meta-epoch": str(epoch_int)},
+            },
             status=drf.status.HTTP_200_OK,
         )
 

src/frontend/apps/drive/src/features/encryption/oo-bridge/OOEditor.tsx

@@ -37,6 +37,7 @@ import {
 } from './participants';
 import { resetPatchIndex } from './changesPipeline';
 import { EncryptedRelay } from './encryptedRelay';
+import { withIncomingOTGate } from './incomingOtGate';
 import {
   acquireCellLock,
   releaseCellLock,
@@ -130,6 +131,34 @@ export const OOEditor = ({ item }: OOEditorProps) => {
   // Reference to the inner OO iframe's window — used to register media in
   // g_oDocumentUrls when inbound saveChanges envelopes carry inline images.
   const innerWindowRef = useRef<any>(null);
+  /**
+   * Remote saveChanges envelopes that arrived before OO finished loading
+   * the base document. OO silently drops external saveChanges that land
+   * before `onDocumentReady`, so we hold them in order and drain once the
+   * editor is ready. Critical for the history-replay path on join: the
+   * relay dumps every pending change immediately after WS open, often
+   * before OO has mounted the document.
+   */
+  const pendingRemoteChangesRef = useRef<Array<() => void>>([]);
+  const documentReadyRef = useRef(false);
+
+  const drainPendingRemoteChanges = useCallback(() => {
+    const queue = pendingRemoteChangesRef.current;
+    if (queue.length === 0) return;
+    console.log(
+      '[OOEditor] draining',
+      queue.length,
+      'remote changes queued before document ready',
+    );
+    pendingRemoteChangesRef.current = [];
+    for (const fn of queue) {
+      try {
+        fn();
+      } catch (e) {
+        console.warn('[OOEditor] pending remote change drain error', e);
+      }
+    }
+  }, []);
   const canEdit = !!item.abilities?.partial_update;
 
   const mime = item.mimetype || '';
@@ -163,7 +192,11 @@ export const OOEditor = ({ item }: OOEditorProps) => {
    * Upload encrypted content to S3.
    */
   const uploadEncrypted = useCallback(
-    async (content: ArrayBuffer, _format: string): Promise<void> => {
+    async (
+      content: ArrayBuffer,
+      _format: string,
+      epochMs: number,
+    ): Promise<void> => {
       const vaultClient = window.__driveVaultClient;
       if (!vaultClient) {
         throw new Error('Vault client not available');
@@ -188,21 +221,34 @@ export const OOEditor = ({ item }: OOEditorProps) => {
         return bytes.buffer;
       });
 
-      // Encrypt via vault
+      // Encrypt via vault. The vault worker takes ownership of every buffer
+      // passed through postMessage (transfer list), so hand it fresh copies
+      // of the key material — otherwise the canonical keys held by the
+      // relay / init closure get detached and subsequent encrypt/decrypt
+      // calls fail with "wrong secret key for the given ciphertext".
       const { encryptedData } = await vaultClient.encryptWithKey(
         content,
-        entryKeyBytes.buffer,
-        encryptedKeyChain.length > 0 ? encryptedKeyChain : undefined
+        entryKeyBytes.buffer.slice(0),
+        encryptedKeyChain.length > 0
+          ? encryptedKeyChain.map(k => k.slice(0))
+          : undefined
       );
 
       // Get a presigned S3 upload URL for the existing file key
       // S3 versioning keeps previous versions — no new filename needed
       const urlResponse = await fetchAPI(
         `items/${item.id}/encryption-upload-url/`,
-        { method: 'POST' },
+        {
+          method: 'POST',
+          body: JSON.stringify({ epoch_ms: epochMs }),
+          headers: { 'Content-Type': 'application/json' },
+        },
         { redirectOn40x: false },
       );
-      const { upload_url: uploadUrl } = await urlResponse.json();
+      const {
+        upload_url: uploadUrl,
+        required_headers: requiredHeaders = {},
+      } = await urlResponse.json();
 
       // Upload encrypted content to S3 via presigned URL (XHR like regular Drive uploads)
       const encryptedBytes = new Uint8Array(encryptedData);
@@ -211,6 +257,13 @@ export const OOEditor = ({ item }: OOEditorProps) => {
         xhr.open('PUT', uploadUrl);
         xhr.setRequestHeader('X-amz-acl', 'private');
         xhr.setRequestHeader('Content-Type', 'application/octet-stream');
+        // Every header that was bound into the presigned signature MUST
+        // be sent verbatim or S3 returns SignatureDoesNotMatch.
+        for (const [name, value] of Object.entries(
+          requiredHeaders as Record<string, string>,
+        )) {
+          xhr.setRequestHeader(name, value);
+        }
         xhr.addEventListener('error', () => reject(new Error('S3 upload network error')));
         xhr.addEventListener('abort', () => reject(new Error('S3 upload aborted')));
         xhr.addEventListener('readystatechange', () => {
@@ -225,6 +278,9 @@ export const OOEditor = ({ item }: OOEditorProps) => {
         xhr.send(encryptedBytes);
       });
 
+      // Tell the relay this epoch is now durable in S3 — it will schedule
+      // a delayed purge of older history entries after the grace window.
+      relayRef.current?.sendSaveCommitted(epochMs);
     },
     [item.id]
   );
@@ -266,6 +322,11 @@ export const OOEditor = ({ item }: OOEditorProps) => {
           cache: 'no-store',
         });
         if (!response.ok) throw new Error(`Fetch failed: ${response.status}`);
+        // Read the snapshot epoch written at save time. Requires nginx to
+        // forward the upstream S3 `x-amz-meta-epoch` header to the browser.
+        // Missing (pre-epoch file) → 0 = "no snapshot anchor, replay all".
+        const rawEpoch = response.headers.get('x-amz-meta-epoch');
+        const snapshotEpochMs = rawEpoch ? Number(rawEpoch) : 0;
         const encryptedBuffer = await response.arrayBuffer();
 
         const vaultClient = window.__driveVaultClient;
@@ -287,10 +348,15 @@ export const OOEditor = ({ item }: OOEditorProps) => {
           return bytes.buffer;
         });
 
+        // Clone the key material — the vault worker transfers its inputs,
+        // detaching the originals. We still need these buffers intact for
+        // the EncryptedRelay constructor below.
         const { data: decryptedBuffer } = await vaultClient.decryptWithKey(
           encryptedBuffer,
-          entryKeyBytes.buffer,
-          encryptedKeyChain.length > 0 ? encryptedKeyChain : undefined
+          entryKeyBytes.buffer.slice(0),
+          encryptedKeyChain.length > 0
+            ? encryptedKeyChain.map(k => k.slice(0))
+            : undefined
         );
         if (cancelled) return;
 
@@ -426,6 +492,10 @@ export const OOEditor = ({ item }: OOEditorProps) => {
             },
             onDocumentReady: () => {
               setState('ready');
+              documentReadyRef.current = true;
+              // OO can now accept external saveChanges. Drain anything the
+              // relay replayed while the document was still loading.
+              drainPendingRemoteChanges();
               try {
                 const ooIframe = document.querySelector(
                   'iframe[name="frameEditor"]',
@@ -1009,30 +1079,45 @@ export const OOEditor = ({ item }: OOEditorProps) => {
             encryptedSymmetricKey: entryKeyBytes.buffer,
             encryptedKeyChain:
               encryptedKeyChain.length > 0 ? encryptedKeyChain : [],
+            sinceTimestampMs: Number.isFinite(snapshotEpochMs)
+              ? snapshotEpochMs
+              : 0,
             callbacks: {
               onSaveChanges: (_userId, message, media) => {
-                // Register any inline media in g_oDocumentUrls BEFORE we
-                // hand the change to OO. The change carries image
-                // references by name; without the bytes registered first,
-                // OO falls back to fetching `<name>` from the editor app
-                // dir and 404s.
-                if (media) {
-                  try {
-                    const docUrls =
-                      innerWindowRef.current?.AscCommon?.g_oDocumentUrls;
-                    if (docUrls) {
-                      for (const [name, url] of Object.entries(media)) {
-                        docUrls.addImageUrl(name, url);
+                const apply = () => {
+                  // Register any inline media in g_oDocumentUrls BEFORE we
+                  // hand the change to OO. The change carries image
+                  // references by name; without the bytes registered first,
+                  // OO falls back to fetching `<name>` from the editor app
+                  // dir and 404s.
+                  if (media) {
+                    try {
+                      const docUrls =
+                        innerWindowRef.current?.AscCommon?.g_oDocumentUrls;
+                      if (docUrls) {
+                        for (const [name, url] of Object.entries(media)) {
+                          docUrls.addImageUrl(name, url);
+                        }
                       }
+                    } catch (e) {
+                      console.warn(
+                        '[OOEditor] failed to register inbound media',
+                        e,
+                      );
                     }
-                  } catch (e) {
-                    console.warn(
-                      '[OOEditor] failed to register inbound media',
-                      e,
-                    );
                   }
+                  sendToEditorGuarded(message as any);
+                };
+                // Hold remote changes that arrive before the base document
+                // has finished loading — OO drops external saveChanges in
+                // that window. The drain runs inside onDocumentReady.
+                if (!documentReadyRef.current) {
+                  pendingRemoteChangesRef.current.push(() =>
+                    withIncomingOTGate(apply),
+                  );
+                  return;
                 }
-                sendToEditorGuarded(message as any);
+                withIncomingOTGate(apply);
               },
               onPeerJoin: (userId, userName, peerCanEdit) => {
                 if (peerCanEdit) editorPeersRef.current.add(userId);

src/frontend/apps/drive/src/features/encryption/oo-bridge/checkpointing.ts

@@ -9,6 +9,7 @@
 import { convertFromInternal } from './x2tConverter';
 import { getPatchIndex } from './changesPipeline';
 import { acquireSaveLock, releaseSaveLock, isSaveLocked } from './locks';
+import { pauseIncomingOT, resumeIncomingOT } from './incomingOtGate';
 
 const CHECKPOINT_CHANGES_THRESHOLD = 50;
 const CHECKPOINT_TIME_INTERVAL_MS = 15_000; // 15 seconds for testing
@@ -39,7 +40,11 @@ let autoSaveTimer: ReturnType<typeof setInterval> | null = null;
 
 /** Callback for uploading encrypted content */
 let uploadCallback:
-  | ((content: ArrayBuffer, format: string) => Promise<void>)
+  | ((
+      content: ArrayBuffer,
+      format: string,
+      epochMs: number,
+    ) => Promise<void>)
   | null = null;
 
 /**
@@ -55,7 +60,11 @@ export function initCheckpointing(opts: {
   format: string;
   type: string;
   userId: string;
-  onUpload: (content: ArrayBuffer, format: string) => Promise<void>;
+  onUpload: (
+    content: ArrayBuffer,
+    format: string,
+    epochMs: number,
+  ) => Promise<void>;
   /** Return true if this client should be responsible for saving.
    *  When absent, all clients save (single-user mode). */
   isSaveLeader?: () => boolean;
@@ -173,7 +182,20 @@ async function saveCheckpoint(): Promise<void> {
       return;
     }
 
-    const rawBin = innerEditor.asc_nativeGetFile();
+    // Capture the snapshot epoch and extract the native binary under the
+    // incoming-OT gate so no remote change can be applied between the two
+    // operations. Any remote change that arrives during this window is
+    // queued and drained after we release — those events have a relay
+    // timestamp > epochMs and will be replayed on joiners as "post-snapshot".
+    let rawBin: unknown;
+    let epochMs: number;
+    pauseIncomingOT();
+    try {
+      epochMs = Date.now();
+      rawBin = innerEditor.asc_nativeGetFile();
+    } finally {
+      resumeIncomingOT();
+    }
 
     if (!rawBin) {
       console.warn('Checkpoint: empty document, skipping save');
@@ -191,7 +213,7 @@ async function saveCheckpoint(): Promise<void> {
     } else if (rawBin instanceof ArrayBuffer) {
       binBuffer = rawBin;
     } else {
-      binBuffer = rawBin.buffer;
+      binBuffer = (rawBin as { buffer: ArrayBuffer }).buffer;
     }
 
     if (binBuffer.byteLength === 0) {
@@ -245,7 +267,7 @@ async function saveCheckpoint(): Promise<void> {
     // Upload (encryption happens in the callback). Note that the vault
     // transfers the buffer into a worker, so `converted.buffer` is detached
     // after this await — read the byte count from the local capture above.
-    await uploadCallback(converted.buffer, originalFormat);
+    await uploadCallback(converted.buffer, originalFormat, epochMs);
 
     lastCheckpointIndex = getPatchIndex();
     lastCheckpointTime = Date.now();
@@ -256,6 +278,8 @@ async function saveCheckpoint(): Promise<void> {
       originalFormat,
       '(images:',
       media.size,
+      ', epochMs:',
+      epochMs,
       ')',
     );
   } catch (error) {

src/frontend/apps/drive/src/features/encryption/oo-bridge/encryptedRelay.ts

@@ -156,6 +156,13 @@ export class EncryptedRelay {
   private maxReconnectAttempts = 3;
   private destroyed = false;
   private pendingOutgoing: ArrayBuffer[] = [];
+  /**
+   * Snapshot epoch (client UTC ms) baked into the S3 file this peer just
+   * loaded. The relay replays events strictly newer than this so we don't
+   * double-apply changes already in the snapshot. Updated after every
+   * successful local save.
+   */
+  private sinceTimestampMs: number;
 
   constructor(opts: {
     roomId: string;
@@ -165,6 +172,8 @@ export class EncryptedRelay {
     encryptedSymmetricKey: ArrayBuffer;
     encryptedKeyChain: ArrayBuffer[];
     callbacks: RelayCallbacks;
+    /** Snapshot epoch (ms) from S3 metadata; 0 means no snapshot. */
+    sinceTimestampMs?: number;
   }) {
     this.roomId = opts.roomId;
     this.userId = opts.userId;
@@ -173,13 +182,18 @@ export class EncryptedRelay {
     this.encryptedSymmetricKey = opts.encryptedSymmetricKey;
     this.encryptedKeyChain = opts.encryptedKeyChain;
     this.callbacks = opts.callbacks;
+    this.sinceTimestampMs = opts.sinceTimestampMs ?? 0;
   }
 
   /** Connect to the relay server */
   connect(): void {
     if (this.destroyed) return;
 
-    const url = `${RELAY_URL}?room=${this.roomId}`;
+    const params = new URLSearchParams({ room: this.roomId });
+    if (this.sinceTimestampMs > 0) {
+      params.set('since', String(this.sinceTimestampMs));
+    }
+    const url = `${RELAY_URL}?${params.toString()}`;
     this.ws = new WebSocket(url);
     // Don't set binaryType to 'arraybuffer' — we need to distinguish
     // text frames (JSON system messages) from binary frames (encrypted patches).
@@ -281,6 +295,18 @@ export class EncryptedRelay {
     });
   }
 
+  /**
+   * Notify the relay that a fresh S3 snapshot has been committed with the
+   * given epoch (client UTC ms). The relay schedules a delayed purge of
+   * every history entry whose timestamp is <= epochMs.
+   */
+  sendSaveCommitted(epochMs: number): void {
+    if (epochMs > this.sinceTimestampMs) {
+      this.sinceTimestampMs = epochMs;
+    }
+    this.sendSystem({ type: 'save:committed', epochMs });
+  }
+
   /** Disconnect and clean up */
   destroy(): void {
     this.destroyed = true;
@@ -302,6 +328,23 @@ export class EncryptedRelay {
 
   // --- Private ---
 
+  /**
+   * The vault worker takes ownership of every ArrayBuffer it receives via
+   * postMessage (transfer list), detaching the original. Our key material is
+   * long-lived and used across many encrypt/decrypt calls, so we hand the
+   * worker a FRESH copy each time — otherwise the first save() in the
+   * session detaches the key and every subsequent decrypt fails with
+   * "wrong secret key for the given ciphertext".
+   */
+  private cloneKey(): ArrayBuffer {
+    return this.encryptedSymmetricKey.slice(0);
+  }
+
+  private cloneKeyChain(): ArrayBuffer[] | undefined {
+    if (this.encryptedKeyChain.length === 0) return undefined;
+    return this.encryptedKeyChain.map(k => k.slice(0));
+  }
+
   private sendSystem(msg: object): void {
     if (this.ws?.readyState === WebSocket.OPEN) {
       this.ws.send(JSON.stringify(msg));
@@ -310,16 +353,34 @@ export class EncryptedRelay {
 
   /** Encrypt a system message and send as binary frame */
   private async sendEncryptedSystem(msg: object): Promise<void> {
+    const type = (msg as { type?: string }).type ?? '?';
     const plaintext = new TextEncoder().encode(JSON.stringify(msg)).buffer;
+    const plaintextSize = plaintext.byteLength;
 
     try {
       const { encryptedData } = await this.vaultClient.encryptWithKey(
         plaintext,
-        this.encryptedSymmetricKey,
-        this.encryptedKeyChain.length > 0 ? this.encryptedKeyChain : undefined,
+        this.cloneKey(),
+        this.cloneKeyChain(),
       );
 
       if (this.ws?.readyState === WebSocket.OPEN) {
+        const head = new Uint8Array(
+          encryptedData.slice(0, Math.min(16, encryptedData.byteLength)),
+        );
+        const headHex = Array.from(head)
+          .map(b => b.toString(16).padStart(2, '0'))
+          .join('');
+        console.log(
+          '[relay] send',
+          type,
+          'plaintext:',
+          plaintextSize,
+          'cipher:',
+          encryptedData.byteLength,
+          'head:',
+          headHex,
+        );
         this.ws.send(encryptedData);
       }
     } catch (err) {
@@ -354,11 +415,20 @@ export class EncryptedRelay {
         return;
       }
 
+      // Capture a fingerprint of the incoming frame BEFORE we hand the
+      // buffer to the vault worker (which transfers ownership, detaching
+      // the original). This lets us log size + leading bytes on failure.
+      const head = new Uint8Array(buffer.slice(0, Math.min(16, buffer.byteLength)));
+      const headHex = Array.from(head)
+        .map(b => b.toString(16).padStart(2, '0'))
+        .join('');
+      const size = buffer.byteLength;
+
       try {
         const { data: plaintext } = await this.vaultClient.decryptWithKey(
           buffer,
-          this.encryptedSymmetricKey,
-          this.encryptedKeyChain.length > 0 ? this.encryptedKeyChain : undefined,
+          this.cloneKey(),
+          this.cloneKeyChain(),
         );
 
         const json = new TextDecoder().decode(plaintext);
@@ -368,7 +438,14 @@ export class EncryptedRelay {
           this.handleSystemMessage(parsed as SystemMessage);
         }
       } catch (err) {
-        console.warn('[relay] Failed to decrypt incoming data:', err);
+        console.warn(
+          '[relay] decrypt failed — size:',
+          size,
+          'head:',
+          headHex,
+          'err:',
+          err,
+        );
       }
     }
   }

src/frontend/apps/drive/src/features/encryption/oo-bridge/incomingOtGate.ts

@@ -0,0 +1,43 @@
+/**
+ * Incoming OT gate.
+ *
+ * Lets the checkpointing flow pause the application of remote OnlyOffice
+ * changes while it captures the snapshot epoch and calls
+ * `asc_nativeGetFile()`. Any remote change that arrives during the window is
+ * queued and drained in order once the gate reopens.
+ *
+ * The epoch captured inside the gate is guaranteed to be strictly older than
+ * any queued remote change — so joiners can anchor on `epochMs` and replay
+ * the queued events as "post-snapshot" without duplication.
+ */
+
+type QueuedChange = () => void;
+
+let pauseCount = 0;
+const queue: QueuedChange[] = [];
+
+export function pauseIncomingOT(): void {
+  pauseCount++;
+}
+
+export function resumeIncomingOT(): void {
+  if (pauseCount > 0) pauseCount--;
+  if (pauseCount === 0 && queue.length > 0) {
+    const drain = queue.splice(0);
+    for (const fn of drain) {
+      try {
+        fn();
+      } catch (e) {
+        console.warn('[ot-gate] drain error', e);
+      }
+    }
+  }
+}
+
+export function withIncomingOTGate(apply: QueuedChange): void {
+  if (pauseCount > 0) {
+    queue.push(apply);
+  } else {
+    apply();
+  }
+}

src/frontend/servers/collaboration/package-lock.json

@@ -8,6 +8,7 @@
       "name": "@drive/collaboration-relay",
       "version": "0.1.0",
       "dependencies": {
+        "async-mutex": "^0.5.0",
         "ws": "^8.18.0"
       },
       "devDependencies": {
@@ -478,6 +479,15 @@
         "@types/node": "*"
       }
     },
+    "node_modules/async-mutex": {
+      "version": "0.5.0",
+      "resolved": "https://registry.npmjs.org/async-mutex/-/async-mutex-0.5.0.tgz",
+      "integrity": "sha512-1A94B18jkJ3DYq284ohPxoXbfTA5HsQ7/Mf4DEhcyLx3Bz27Rh59iScbB6EPiP+B+joue6YCxcMXSbFC1tZKwA==",
+      "license": "MIT",
+      "dependencies": {
+        "tslib": "^2.4.0"
+      }
+    },
     "node_modules/esbuild": {
       "version": "0.27.7",
       "resolved": "https://registry.npmjs.org/esbuild/-/esbuild-0.27.7.tgz",
@@ -558,6 +568,12 @@
         "url": "https://github.com/privatenumber/resolve-pkg-maps?sponsor=1"
       }
     },
+    "node_modules/tslib": {
+      "version": "2.8.1",
+      "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz",
+      "integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==",
+      "license": "0BSD"
+    },
     "node_modules/tsx": {
       "version": "4.21.0",
       "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.21.0.tgz",

src/frontend/servers/collaboration/package.json

@@ -8,6 +8,7 @@
     "build": "tsc"
   },
   "dependencies": {
+    "async-mutex": "^0.5.0",
     "ws": "^8.18.0"
   },
   "devDependencies": {

src/frontend/servers/collaboration/relay.ts

@@ -10,11 +10,18 @@
 
 import { WebSocketServer, WebSocket, RawData } from 'ws';
 import { createServer, IncomingMessage } from 'http';
+import { Mutex } from 'async-mutex';
 
 const PORT = parseInt(process.env.RELAY_PORT || '4100', 10);
 const DRIVE_API_URL = process.env.DRIVE_API_URL || 'http://app-dev:8000';
-const MAX_HISTORY = 500;
 const PING_INTERVAL_MS = 30_000;
+/**
+ * Grace period between a `save:committed` message and the actual purge of
+ * events whose timestamp is <= the committed epoch. The delay exists so a
+ * joiner who just fetched an older snapshot (epoch N-2) still finds the
+ * events that bring them up to epoch N-1 in the relay history.
+ */
+const HISTORY_PURGE_DELAY_MS = 10_000;
 // Upper bound on a single relayed message. The frontend caps raw image bytes
 // at 50 MB; allow ~75 MB on the wire to absorb base64 (+33%), the JSON change
 // envelope and the encryption header. A peer trying to push more is closed
@@ -85,9 +92,16 @@ async function authenticateConnection(
 
 // --- Room management ---
 
+interface HistoryEntry {
+  timestampMs: number;
+  data: RawData;
+}
+
 interface Room {
   peers: Map<WebSocket, PeerMeta>;
-  history: RawData[];
+  history: HistoryEntry[];
+  /** Serializes delayed history purges against each other. */
+  purgeMutex: Mutex;
 }
 
 interface PeerMeta {
@@ -104,7 +118,7 @@ const rooms = new Map<string, Room>();
 function getOrCreateRoom(roomId: string): Room {
   let room = rooms.get(roomId);
   if (!room) {
-    room = { peers: new Map(), history: [] };
+    room = { peers: new Map(), history: [], purgeMutex: new Mutex() };
     rooms.set(roomId, room);
   }
   return room;
@@ -131,7 +145,7 @@ function removeFromRoom(roomId: string, ws: WebSocket): void {
 
 // --- Message sending ---
 
-function sendRaw(ws: WebSocket, data: RawData): void {
+function sendRaw(ws: WebSocket, data: RawData | string): void {
   if (ws.readyState === WebSocket.OPEN) {
     ws.send(data, {}, err => {
       if (err) ws.close();
@@ -165,7 +179,8 @@ function broadcastSystem(room: Room, sender: WebSocket, msg: object): void {
 function handleConnection(
   ws: WebSocket,
   roomId: string,
-  auth: AuthResult
+  auth: AuthResult,
+  sinceTimestampMs: number
 ): void {
   ws.binaryType = 'arraybuffer';
 
@@ -200,29 +215,56 @@ function handleConnection(
     ws.close();
   });
 
-  ws.on('message', (data: RawData) => {
-    // Read-only users can't send patches
-    if (!meta.canEdit && data instanceof ArrayBuffer) {
+  // IMPORTANT: the ws library (v8+) delivers BOTH text and binary frames as
+  // Buffer. The `isBinary` second argument is the ONLY reliable way to tell
+  // them apart — `typeof data === 'string'` is never true here. Miss this
+  // and every client text frame (including `save:committed`) falls into the
+  // binary branch, gets broadcast to every peer, and pollutes history.
+  ws.on('message', (data: RawData, isBinary: boolean) => {
+    // Text = system message. Always consume the text path here — a text
+    // frame must never fall through to the encrypted-binary branch.
+    if (!isBinary) {
+      const text = Buffer.isBuffer(data)
+        ? data.toString('utf-8')
+        : typeof data === 'string'
+          ? (data as string)
+          : Buffer.from(data as ArrayBuffer).toString('utf-8');
+      try {
+        const msg = JSON.parse(text);
+        handleSystemMessage(room, ws, meta, msg);
+      } catch {
+        // Malformed text frame — drop silently.
+      }
       return;
     }
 
-    // Text = system message
-    if (typeof data === 'string') {
-      try {
-        const msg = JSON.parse(data);
-        handleSystemMessage(room, ws, meta, msg);
-        return;
-      } catch {
-        // Not JSON
-      }
+    // Read-only users can't send patches
+    if (!meta.canEdit) {
+      return;
     }
 
-    // Binary = encrypted OT patch — relay to all OTHER peers
-    broadcastRaw(room, ws, data);
-    room.history.push(data);
-    if (room.history.length > MAX_HISTORY) {
-      room.history.shift();
+    // Binary = encrypted OT patch — relay to all OTHER peers.
+    // ws delivers binary frames as Buffer, ArrayBuffer, or Buffer[]
+    // depending on version/fragmentation. Normalize to a Buffer so the
+    // history replay and broadcast paths have one concrete type.
+    let binary: Buffer;
+    if (Buffer.isBuffer(data)) {
+      binary = data;
+    } else if (Array.isArray(data)) {
+      binary = Buffer.concat(data);
+    } else if (data instanceof ArrayBuffer) {
+      binary = Buffer.from(data);
+    } else {
+      console.warn('[relay] dropped unknown binary frame type');
+      return;
     }
+    // Reject frames too small to be a valid vault envelope (nonce+tag+...).
+    if (binary.byteLength < 32) {
+      console.warn('[relay] dropped tiny binary frame size:', binary.byteLength);
+      return;
+    }
+    broadcastRaw(room, ws, binary);
+    room.history.push({ timestampMs: Date.now(), data: binary });
   });
 
   ws.on('close', () => {
@@ -246,27 +288,25 @@ function handleConnection(
       canEdit: m.canEdit,
     }));
 
+  // Filter history to events strictly newer than the snapshot epoch the
+  // joiner already has baked in (from S3 metadata). sinceTimestampMs = 0 means
+  // "no snapshot" — send everything we have.
+  const replay = room.history.filter(
+    e => e.timestampMs > sinceTimestampMs
+  );
+
   sendJSON(ws, {
     type: 'room:state',
     peers: peerList,
-    historyLength: room.history.length,
+    historyLength: replay.length,
+    sinceTimestampMs,
   });
 
-  // Send history to the new joiner (only if there are other peers who contributed)
-  if (peerList.length > 0) {
-    for (const historyMsg of room.history) {
-      sendRaw(ws, historyMsg);
-    }
+  for (const entry of replay) {
+    sendRaw(ws, entry.data);
   }
 }
 
-type SystemMessageType =
-  | 'lock:acquire'
-  | 'lock:release'
-  | 'cursor:update'
-  | 'save:lock'
-  | 'save:unlock';
-
 // Only non-encrypted system messages are parsed by the relay.
 // cursor:update and lock:acquire/release are now encrypted (sent as binary)
 // and relayed opaquely without parsing.
@@ -281,6 +321,12 @@ function handleSystemMessage(
   meta: PeerMeta,
   msg: { type: string; [key: string]: unknown }
 ): void {
+  if (msg.type === 'save:committed') {
+    const epochMs = Number(msg.epochMs);
+    if (!Number.isFinite(epochMs) || epochMs <= 0) return;
+    scheduleHistoryPurge(room, epochMs);
+    return;
+  }
   if (RELAYED_TYPES.has(msg.type)) {
     // Inject the authenticated userId (server-authoritative, not client-supplied)
     broadcastSystem(room, ws, { ...msg, userId: meta.userId });
@@ -288,6 +334,26 @@ function handleSystemMessage(
   // Unknown types are silently dropped
 }
 
+/**
+ * Delayed purge of history events older than the committed snapshot epoch.
+ * The grace window lets joiners mid-fetch still pick up events that bridge
+ * their older snapshot to the newly-committed one.
+ */
+function scheduleHistoryPurge(room: Room, epochMs: number): void {
+  setTimeout(() => {
+    room.purgeMutex.runExclusive(() => {
+      const before = room.history.length;
+      room.history = room.history.filter(e => e.timestampMs > epochMs);
+      const removed = before - room.history.length;
+      if (removed > 0) {
+        console.log(
+          `[relay] purged ${removed} history entries <= epoch ${epochMs}`
+        );
+      }
+    });
+  }, HISTORY_PURGE_DELAY_MS);
+}
+
 // --- Server setup ---
 
 const httpServer = createServer((_req, res) => {
@@ -317,12 +383,19 @@ const wss = new WebSocketServer({
 wss.on('connection', async (ws, req) => {
   const url = new URL(req.url || '/', 'ws://localhost');
   const roomId = url.searchParams.get('room');
+  const sinceRaw = url.searchParams.get('since');
+  const sinceTimestampMs = sinceRaw ? Number(sinceRaw) : 0;
 
   if (!roomId) {
     ws.close(1008, 'room parameter required');
     return;
   }
 
+  if (!Number.isFinite(sinceTimestampMs) || sinceTimestampMs < 0) {
+    ws.close(1008, 'invalid since parameter');
+    return;
+  }
+
   if (
     !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(
       roomId
@@ -353,7 +426,7 @@ wss.on('connection', async (ws, req) => {
     `[relay] ${auth.userName} (${auth.userId}) joined room ${roomId} (canEdit: ${auth.canEdit})`
   );
 
-  handleConnection(ws, roomId, auth);
+  handleConnection(ws, roomId, auth, sinceTimestampMs);
 });
 
 httpServer.listen(PORT, () => {