Livestore sync engine migration

[juliusmarminge]

This pull request contains changes generated by a Cursor Cloud Agent

  

[cursor]

Cursor Agent can help with this pull request. Just @cursor in comments and I'll start working on changes in this branch.
_{Learn more about Cursor Agents}

[coderabbitai]

Important

Review skipped

Draft detected.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch cursor/livestore-sync-engine-migration-7f06

Tip

Issue Planner is now in beta. Read the docs and try it out! Share your feedback on Discord.

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[macroscopeapp]

Migrate server state handling to apps/server/src/wsServer.ts using StateSyncEngine and expose syncEngineMode for LiveStore rollout

Introduce SyncEngineMode resolution and enforcement, refactor WebSocket server to delegate state routes to StateSyncEngine, add LegacyStateSyncEngine, ShadowStateSyncEngine, and LiveStoreReadPilotStateSyncEngine, wire startup to select engine via env, and return syncEngineMode in server config.

📍Where to Start

Start with createServer in wsServer.ts and follow calls into inferSyncEngineMode, StateSyncEngine implementations, and index.ts bootstrap.

---

📊 Macroscope summarized c3975c1. 17 files reviewed, 48 issues evaluated, 1 issue filtered, 5 comments posted. View details

[macroscopeapp]

🟡 Medium

livestore/liveStoreEngine.ts:286 Messages are stored in threadMessagesByThreadId without verifying the thread exists in threadsById. This creates orphaned entries that debugReadSnapshot never exposes. If this handles out-of-order events, consider documenting that rationale; otherwise, consider adding a guard to skip messages for unknown threads.

🚀 Want me to fix this? Reply ex: "fix it for me".

🤖 Prompt for AI

In file apps/server/src/livestore/liveStoreEngine.ts around line 286:

Messages are stored in `threadMessagesByThreadId` without verifying the thread exists in `threadsById`. This creates orphaned entries that `debugReadSnapshot` never exposes. If this handles out-of-order events, consider documenting that rationale; otherwise, consider adding a guard to skip messages for unknown threads.

Evidence trail:
apps/server/src/livestore/liveStoreEngine.ts lines 280-289 (commit c3975c1b224b9a1a1b631dfb18b9258c3dd4a76b): `message.upsert` handler stores messages in `threadMessagesByThreadId` without checking `threadsById.has(threadId)`.

apps/server/src/livestore/liveStoreEngine.ts lines 177-198 (commit c3975c1b224b9a1a1b631dfb18b9258c3dd4a76b): `debugReadSnapshot` iterates only over `threadsById.values()` and fetches messages for those threads, meaning messages stored for non-existent threads are never exposed.

[macroscopeapp]

🟠 High

livestore/liveStoreEngine.ts:69 Concurrent calls to this async method can interleave between the seq check (line 73) and state updates (lines 92-93), causing later events to be overwritten by earlier ones that resume after await. Consider serializing calls with a queue or mutex.

🚀 Want me to fix this? Reply ex: "fix it for me".

🤖 Prompt for AI

In file apps/server/src/livestore/liveStoreEngine.ts around line 69:

Concurrent calls to this async method can interleave between the `seq` check (line 73) and state updates (lines 92-93), causing later events to be overwritten by earlier ones that resume after `await`. Consider serializing calls with a queue or mutex.

Evidence trail:
apps/server/src/livestore/liveStoreEngine.ts lines 69-103 at commit c3975c1b224b9a1a1b631dfb18b9258c3dd4a76b: async method `mirrorStateEvent` at line 69, seq check at line 73, await at line 77, state updates at lines 92-93.

[macroscopeapp]

🟡 Medium

src/stateSyncEngineShadow.ts:294 Suggestion: In close(), handle rejections from mirror.dispose() (e.g., add .catch(...)) to avoid unhandled promise rejections.

Suggested change

	void this.mirror.dispose();
	void Promise.resolve(this.mirror.dispose()).catch((error) => this.logger.warn("mirror dispose failed", { error }));

🚀 Want me to fix this? Reply ex: "fix it for me".

🤖 Prompt for AI

In file apps/server/src/stateSyncEngineShadow.ts around line 294:

Suggestion: In `close()`, handle rejections from `mirror.dispose()` (e.g., add `.catch(...)`) to avoid unhandled promise rejections.

Evidence trail:
apps/server/src/stateSyncEngineShadow.ts line 294: `void this.mirror.dispose();`
apps/server/src/stateSyncEngineShadow.ts lines 34-40: `StateEventMirror` interface defines `dispose(): Promise<void> | void;`
apps/server/src/livestore/liveStoreEngine.ts line 105: `async dispose(): Promise<void>` implementation showing it returns a Promise

[macroscopeapp]

🟡 Medium

livestore/liveStoreEngine.ts:208 Suggestion: mirroredEventsBySeq grows without bound and debugCatchUp copies it; add a retention policy (max size or time window) and evict oldest entries to avoid OOM on long-running instances.

🚀 Want me to fix this? Reply ex: "fix it for me".

🤖 Prompt for AI

In file apps/server/src/livestore/liveStoreEngine.ts around line 208:

Suggestion: `mirroredEventsBySeq` grows without bound and `debugCatchUp` copies it; add a retention policy (max size or time window) and evict oldest entries to avoid OOM on long-running instances.

Evidence trail:
apps/server/src/livestore/liveStoreEngine.ts lines 62, 207-208, 233-234, 118 at commit c3975c1b224b9a1a1b631dfb18b9258c3dd4a76b. git_grep for 'mirroredEventsBySeq\.delete|mirroredEventsBySeq\.clear' returns only line 118 (dispose method).

[macroscopeapp]

🟡 Medium

src/stateSyncEngineLiveStoreReadPilot.ts:192 Reads use mirror debug helpers that materialize/copy whole state (debugReadSnapshot, debugCatchUp). This is heavy for hot paths and can OOM as history grows. Suggest adding scoped, zero‑copy read APIs to LiveStoreStateMirror (per‑thread messages, event range/iterator) and pruning/capping retained events, or document why unbounded retention is acceptable in the pilot.

🚀 Want me to fix this? Reply ex: "fix it for me".

🤖 Prompt for AI

In file apps/server/src/stateSyncEngineLiveStoreReadPilot.ts around line 192:

Reads use mirror debug helpers that materialize/copy whole state (`debugReadSnapshot`, `debugCatchUp`). This is heavy for hot paths and can OOM as history grows. Suggest adding scoped, zero‑copy read APIs to `LiveStoreStateMirror` (per‑thread messages, event range/iterator) and pruning/capping retained events, or document why unbounded retention is acceptable in the pilot.

Evidence trail:
apps/server/src/stateSyncEngineLiveStoreReadPilot.ts:192 (uses `this.mirror.debugListMessages(raw)`), apps/server/src/stateSyncEngineLiveStoreReadPilot.ts:122 (uses `this.mirror.debugReadSnapshot()`), apps/server/src/stateSyncEngineLiveStoreReadPilot.ts:266 (uses `this.mirror.debugCatchUp(afterSeq)`), apps/server/src/livestore/liveStoreEngine.ts:177-231 (implementations show `debugReadSnapshot` uses `Array.from(this.projectsById.values())` and `Array.from(this.threadsById.values())` materializing all data, `debugCatchUp` uses `Array.from(this.mirroredEventsBySeq.entries())`, and `debugListMessages` calls `this.debugReadSnapshot()` on line 219 before filtering)