feat(collaboration): implement lazy CRDT initialization with enhanced logging

Changes the CollaborationActor to initialize the Automerge document only after joining a room, enabling offline-first usage before peers connect. This improves the actor lifecycle and makes peer synchronization more predictable.

Key changes:
- Defer Automerge document initialization until roomJoined event
- First peer in room initializes from current state, joining peers start empty
- Add comprehensive logging throughout sync workflow for debugging
- Fix race condition where sync messages could arrive before doc initialization
- Make setState() return Promise to allow awaiting state commits
- Update PeerMessagingActor to emit roomJoined event and handle existing peers correctly
- Prevent duplicate WebRTC initiation when joining existing peers

This addresses sync timing issues and provides better visibility into the collaboration flow.

Author: d-buckner

packages/collaboration/src/CollaborationActor.ts

@@ -1,7 +1,7 @@
 import * as Automerge from '@automerge/automerge';
-import { Actor, effect, type IActorClient, type StateShape } from '@d-buckner/ensemble-core';
+import { Actor, effect, type IActorClient } from '@d-buckner/ensemble-core';
 import type { PeerMessagingActor } from './PeerMessagingActor';
-import type { CollaborationEvents, AutomergeDoc, SyncState, MessagePayload } from './types';
+import type { CollaborationEvents, AutomergeDoc, SyncState, MessagePayload, RoomJoinedPayload } from './types';
 import type { Draft } from 'mutative';
 
 /**
@@ -44,49 +44,40 @@ export interface CollaborationDeps {
  */
 export class CollaborationActor<TDoc extends Record<string, any> = Record<string, any>> extends Actor<TDoc, CollaborationEvents> {
   // Automerge internals (private, NOT in state)
-  private automergeDoc: AutomergeDoc<TDoc>;
+  // Null until roomJoined - allows offline-first usage before peers connect
+  private automergeDoc: AutomergeDoc<TDoc> | null = null;
   private syncStates = new Map<string, SyncState>();
 
   protected declare deps: CollaborationDeps;
 
   /**
-   * Create a new CollaborationActor with an initial document.
-   * The document becomes the actor's state directly.
-   *
-   * @param initialDocument - Initial CRDT document (will be wrapped by Automerge)
-   */
-  constructor(initialDocument: StateShape<TDoc>) {
-    super(initialDocument);
-    this.automergeDoc = Automerge.from(initialDocument as TDoc);
-  }
-
-  /**
-   * Override setState to route through Automerge CRDT.
-   *
    * This enables transparent conflict resolution - users call setState()
    * just like any other actor, but changes are automatically synced
    * with peers and conflicts are resolved via Automerge.
    *
+   * Before peers connect (automergeDoc is null), acts as normal Actor with local state only.
+   *
    * @param updater - Function that mutates a draft of the current state
+   * @returns Promise that resolves when the state update has been committed
    */
-  protected setState(updater: (draft: Draft<TDoc>) => void): void {
+  protected setState(updater: (draft: Draft<TDoc>) => void): Promise<void> {
+    // If no Automerge doc yet, just update local state (offline-first)
+    if (!this.automergeDoc) {
+      return super.setState(updater);
+    }
+
     // 1. Apply via Automerge (conflict resolution)
     const newDoc = Automerge.change(this.automergeDoc, updater as any);
     const changes = Automerge.getChanges(this.automergeDoc, newDoc);
     this.automergeDoc = newDoc;
 
     // 2. Update actor state via parent (triggers state events)
     const jsDoc = Automerge.toJS(newDoc);
-    super.setState(draft => {
-      // Directly assign each property to ensure reactivity
-      (Object.keys(jsDoc) as Array<keyof TDoc>).forEach(key => {
-        (draft as TDoc)[key] = jsDoc[key];
-      });
-    });
 
     // 3. Generate and send sync messages for peers
     if (changes.length > 0) {
       const peers = this.deps.connection.state.connectedPeers;
+      console.log(`[CollaborationActor] 📝 Broadcasting ${changes.length} change(s) to ${peers.length} peer(s)`);
       for (const peerId of peers) {
         const syncMsg = this.generateSyncMessageForPeer(peerId);
         if (syncMsg) {
@@ -95,18 +86,51 @@ export class CollaborationActor<TDoc extends Record<string, any> = Record<string
         }
       }
     }
+
+    return super.setState(draft => {
+      // Directly assign each property to ensure reactivity
+      (Object.keys(jsDoc) as Array<keyof TDoc>).forEach(key => {
+        (draft as TDoc)[key] = jsDoc[key];
+      });
+    });
   }
 
   // ========================================
   // Effects: React to connection events
   // ========================================
 
+  /**
+   * Handle room joined event to initialize Automerge document.
+   * First peer in room creates doc from current state, others start empty.
+   */
+  @effect('connection.roomJoined')
+  private handleRoomJoined(payload: RoomJoinedPayload): void {
+    // Joining peer - start empty, receive via sync
+    if (payload.peerIds.length > 0) {
+      this.automergeDoc = Automerge.init();
+      console.log(`[CollaborationActor] 👋 Joining peer - initialized empty Automerge doc (${payload.peerIds.length} existing peer(s))`);
+      return;
+    }
+
+    // First peer - initialize from current state
+    this.automergeDoc = Automerge.from(this.state as TDoc);
+    console.log('[CollaborationActor] 🎉 First peer - initialized Automerge doc from current state');
+  }
+
   /**
    * Handle incoming CRDT sync messages from peers.
    * Applies remote changes and generates response if needed.
    */
   @effect('connection.messageReceived')
   private handleIncomingMessage({ peerId, message }: MessagePayload): void {
+    console.log(`[CollaborationActor] 📨 Received sync message from ${peerId} (${message.length} bytes)`);
+
+    // Guard: Automerge doc must be initialized before receiving sync messages
+    if (!this.automergeDoc) {
+      console.warn('[CollaborationActor] ⚠️  Received sync message before Automerge doc initialized - dropping message');
+      return;
+    }
+
     const syncState = this.syncStates.get(peerId) || Automerge.initSyncState();
 
     const [newDoc, newSyncState] = Automerge.receiveSyncMessage(
@@ -119,6 +143,7 @@ export class CollaborationActor<TDoc extends Record<string, any> = Record<string
 
     // Document changed - update via parent setState (skip sync broadcast)
     if (newDoc !== this.automergeDoc) {
+      console.log(`[CollaborationActor] ✅ Document updated from ${peerId}`);
       this.automergeDoc = newDoc;
       const jsDoc = Automerge.toJS(newDoc);
       super.setState(draft => {
@@ -127,6 +152,8 @@ export class CollaborationActor<TDoc extends Record<string, any> = Record<string
           (draft as TDoc)[key] = jsDoc[key];
         });
       });
+    } else {
+      console.log(`[CollaborationActor] ℹ️  No document changes from ${peerId}`);
     }
 
     // Generate response if needed
@@ -136,9 +163,12 @@ export class CollaborationActor<TDoc extends Record<string, any> = Record<string
     );
 
     if (responseMsg) {
+      console.log(`[CollaborationActor] 📤 Sending sync response to ${peerId} (${responseMsg.length} bytes)`);
       this.syncStates.set(peerId, nextSyncState);
       // PeerMessagingActor handles routing
       this.deps.connection.actions.sendTo(peerId, responseMsg);
+    } else {
+      console.log(`[CollaborationActor] ℹ️  No sync response needed for ${peerId}`);
     }
   }
 
@@ -148,17 +178,29 @@ export class CollaborationActor<TDoc extends Record<string, any> = Record<string
    */
   @effect('connection.peerConnected')
   private initSyncWithPeer(peerId: string): void {
+    console.log(`[CollaborationActor] 🔄 Peer connected, initiating sync with: ${peerId}`);
+
+    // Guard: Automerge doc must be initialized before syncing with peers
+    if (!this.automergeDoc) {
+      console.warn('[CollaborationActor] ⚠️  Peer connected before Automerge doc initialized - skipping sync');
+      return;
+    }
+
     const syncState = Automerge.initSyncState();
     const [newSyncState, message] = Automerge.generateSyncMessage(
       this.automergeDoc,
       syncState
     );
 
-    if (message) {
-      this.syncStates.set(peerId, newSyncState);
-      // PeerMessagingActor handles routing
-      this.deps.connection.actions.sendTo(peerId, message);
+    if (!message) {
+      console.log(`[CollaborationActor] ⚠️  No sync message generated for ${peerId}`);
+      return;
     }
+
+    console.log(`[CollaborationActor] 📤 Sending initial sync message to ${peerId} (${message.length} bytes)`);
+    this.syncStates.set(peerId, newSyncState);
+    // PeerMessagingActor handles routing
+    this.deps.connection.actions.sendTo(peerId, message);
   }
 
   /**
@@ -181,6 +223,11 @@ export class CollaborationActor<TDoc extends Record<string, any> = Record<string
    * @returns Sync message or null if no sync needed
    */
   private generateSyncMessageForPeer(peerId: string): Uint8Array | null {
+    // Guard: Automerge doc must be initialized
+    if (!this.automergeDoc) {
+      return null;
+    }
+
     const syncState = this.syncStates.get(peerId) || Automerge.initSyncState();
     const [newSyncState, message] = Automerge.generateSyncMessage(
       this.automergeDoc,

packages/collaboration/src/PeerMessagingActor.ts

@@ -4,6 +4,7 @@ import type {
   PeerMessagingEvents,
   MessagePayload,
   SignalingPayload,
+  RoomJoinedPayload,
 } from './types';
 import type { WebRTCActor } from './WebRTCActor';
 import type { WebSocketActor } from './WebSocketActor';
@@ -58,22 +59,29 @@ export class PeerMessagingActor extends Actor<PeerMessagingState, PeerMessagingE
   sendTo(peerId: string, message: Uint8Array): void {
     const transport = this.state.peerTransports[peerId];
 
+    console.log(`[PeerMessagingActor] 📨 sendTo called for peer: ${peerId}, transport: ${transport}, message: ${message.length} bytes`);
+
     if (!transport) {
+      console.warn(`[PeerMessagingActor] ⚠️  No transport found for peer: ${peerId} - message dropped!`);
       return;
     }
 
     // Check WebRTC connection state before attempting to use it
     if (transport === 'webrtc') {
       const webrtcState = this.deps.webrtc.state.peerConnectionStates[peerId];
+      console.log(`[PeerMessagingActor] 🔀 Transport is WebRTC, state: ${webrtcState}`);
       if (webrtcState === 'connected') {
+        console.log(`[PeerMessagingActor] ⚡ Sending via WebRTC to ${peerId}`);
         this.deps.webrtc.actions.sendTo(peerId, message);
         return;
       }
       // WebRTC not connected, fallback to WebSocket
+      console.log(`[PeerMessagingActor] 🔄 WebRTC not ready, falling back to WebSocket for ${peerId}`);
       this.deps.websocket.actions.sendTo(peerId, message);
       return;
     }
 
+    console.log(`[PeerMessagingActor] 🌐 Sending via WebSocket to ${peerId}`);
     this.deps.websocket.actions.sendTo(peerId, message);
   }
 
@@ -93,25 +101,55 @@ export class PeerMessagingActor extends Actor<PeerMessagingState, PeerMessagingE
   // Effects: WebSocket events
   // ========================================
 
+  /**
+   * Handle room joined via WebSocket.
+   * Emits roomJoined event, then adds existing peers to state WITHOUT initiating WebRTC.
+   * The existing peers will initiate the connection when they receive peer-joined.
+   */
+  @effect('websocket.roomJoined')
+  private async handleWebSocketRoomJoined(payload: RoomJoinedPayload): Promise<void> {
+    console.log(`[PeerMessagingActor] 🏠 Joined room with ${payload.peerIds.length} existing peer(s) - NOT initiating WebRTC`);
+
+    // Emit roomJoined event for dependent actors
+    this.emit('roomJoined', payload);
+
+    // Add all existing peers to state
+    for (const peerId of payload.peerIds) {
+      if (!this.state.connectedPeers.includes(peerId)) {
+        await this.setState(draft => {
+          draft.connectedPeers.push(peerId);
+          draft.peerTransports[peerId] = 'websocket';
+        });
+
+        // Emit after state is committed
+        this.emit('peerConnected', peerId);
+      }
+    }
+    // Do NOT initiate WebRTC - let the existing peers initiate to us
+  }
+
   /**
    * Handle peer joined via WebSocket.
    * Adds peer to state with WebSocket transport and emits peerConnected.
    * Also initiates WebRTC connection as the initiator.
    */
   @effect('websocket.peerJoined')
-  private handleWebSocketPeerJoined(peerId: string): void {
+  private async handleWebSocketPeerJoined(peerId: string): Promise<void> {
     if (this.state.connectedPeers.includes(peerId)) {
       return;
     }
 
-    this.setState(draft => {
+    console.log(`[PeerMessagingActor] 👋 New peer joined: ${peerId} - initiating WebRTC as existing peer`);
+
+    await this.setState(draft => {
       draft.connectedPeers.push(peerId);
       draft.peerTransports[peerId] = 'websocket';
     });
 
+    // Emit after state is committed
     this.emit('peerConnected', peerId);
 
-    // Initiate WebRTC connection
+    // Initiate WebRTC connection as the initiator (we're the existing peer)
     this.deps.webrtc.actions.connectToPeer(peerId);
   }
 
@@ -143,6 +181,7 @@ export class PeerMessagingActor extends Actor<PeerMessagingState, PeerMessagingE
    */
   @effect('websocket.signalingMessage')
   private handleWebSocketSignaling(payload: SignalingPayload): void {
+    console.log(`[PeerMessagingActor] 🔀 Forwarding WebSocket signaling to WebRTC for peer: ${payload.peerId}`);
     this.deps.webrtc.actions.handleSignaling(payload);
   }
 
@@ -182,6 +221,8 @@ export class PeerMessagingActor extends Actor<PeerMessagingState, PeerMessagingE
 
     const previousTransport = this.state.peerTransports[peerId];
 
+    console.log(`[PeerMessagingActor] 🚀 Upgrading transport to WebRTC for peer: ${peerId} (was: ${previousTransport})`);
+
     this.setState(draft => {
       draft.peerTransports[peerId] = 'webrtc';
     });
@@ -219,6 +260,7 @@ export class PeerMessagingActor extends Actor<PeerMessagingState, PeerMessagingE
    */
   @effect('webrtc.signalingData')
   private handleWebRTCSignaling(payload: SignalingPayload): void {
+    console.log(`[PeerMessagingActor] 🔀 Forwarding WebRTC signaling to WebSocket for peer: ${payload.peerId}`);
     this.deps.websocket.actions.sendSignal(payload.peerId, payload.data);
   }
 

packages/collaboration/src/WebRTCActor.ts

@@ -164,11 +164,13 @@ export class WebRTCActor extends Actor<WebRTCState, WebRTCEvents> {
 
     // Connection closed
     peer.on('close', () => {
+      console.log(`[WebRTCActor] 🔌 WebRTC connection closed for peer: ${peerId}`);
       this.handlePeerConnectionClosed(peerId);
     });
 
     // Connection error
-    peer.on('error', (_error: Error) => {
+    peer.on('error', (error: Error) => {
+      console.error(`[WebRTCActor] ❌ WebRTC error for peer ${peerId}:`, error);
       this.setState(draft => {
         draft.peerConnectionStates[peerId] = 'failed';
       });

packages/collaboration/src/WebSocketActor.ts

@@ -135,9 +135,12 @@ export class WebSocketActor extends Actor<WebSocketState, WebSocketEvents> {
   @action
   sendTo(peerId: string, message: Uint8Array): void {
     if (!this.socket?.connected) {
+      console.warn('[WebSocketActor] ⚠️  Cannot send CRDT message - socket not connected');
       return;
     }
 
+    console.log(`[WebSocketActor] 📤 Sending CRDT sync message to peer: ${peerId} (${message.length} bytes)`);
+
     this.socket.emit('sync-message', {
       to: peerId,
       message: Array.from(message), // Convert Uint8Array to regular array for JSON serialization
@@ -190,11 +193,6 @@ export class WebSocketActor extends Actor<WebSocketState, WebSocketEvents> {
         peerId: data.peerId,
         peerIds: data.peers,
       });
-
-      // Emit peerJoined for each existing peer
-      for (const peerId of data.peers) {
-        this.emit('peerJoined', peerId);
-      }
     });
 
     this.socket.on('peer-joined', (peerId: string) => {
@@ -217,6 +215,7 @@ export class WebSocketActor extends Actor<WebSocketState, WebSocketEvents> {
     this.socket.on('sync-message', (data: { from: string; message: number[] }) => {
       // Convert array back to Uint8Array
       const message = new Uint8Array(data.message);
+      console.log(`[WebSocketActor] 📥 Received CRDT sync message from peer: ${data.from} (${message.length} bytes)`);
       this.emit('messageReceived', {
         peerId: data.from,
         message,

packages/collaboration/src/types.ts

@@ -31,6 +31,7 @@ export interface TransportChangedPayload {
 }
 
 export interface PeerMessagingEvents {
+  roomJoined: RoomJoinedPayload;
   peerConnected: string;
   peerDisconnected: string;
   transportChanged: TransportChangedPayload;