refactor: migrate ClientSessionSyncProcessor.push to Effect

Author: IGassmann

packages/@livestore/common/src/sync/ClientSessionSyncProcessor.ts

@@ -76,7 +76,10 @@ export const makeClientSessionSyncProcessor = ({
   /** We're queuing push requests to reduce the number of messages sent to the leader by batching them */
   const leaderPushQueue = BucketQueue.make<LiveStoreEvent.EncodedWithMeta>().pipe(Effect.runSync)
 
-  const push: ClientSessionSyncProcessor['push'] = (batch, { otelContext }) => {
+  const push: ClientSessionSyncProcessor['push'] = Effect.fn('client-session-sync-processor:push')(function* (
+    batch,
+    { otelContext },
+  ) {
     // TODO validate batch
 
     let baseEventSequenceNumber = syncStateRef.current.localHead
@@ -94,29 +97,29 @@ export const makeClientSessionSyncProcessor = ({
         }),
       )
     })
-
-    const mergeResult = SyncState.merge({
-      syncState: syncStateRef.current,
-      payload: { _tag: 'local-push', newEvents: encodedEventDefs },
-      isClientEvent,
-      isEqualEvent: LiveStoreEvent.isEqualEncoded,
-    })
+    yield* Effect.annotateCurrentSpan({ batchSize: encodedEventDefs.length })
+
+    const mergeResult = yield* Effect.sync(() =>
+      SyncState.merge({
+        syncState: syncStateRef.current,
+        payload: { _tag: 'local-push', newEvents: encodedEventDefs },
+        isClientEvent,
+        isEqualEvent: LiveStoreEvent.isEqualEncoded,
+      }),
+    )
 
     if (mergeResult._tag === 'unexpected-error') {
-      return shouldNeverHappen('Unexpected error in client-session-sync-processor', mergeResult.cause)
+      return yield* Effect.die(new Error(`Unexpected error in client-session-sync-processor: ${mergeResult.cause}`))
     }
 
-    span.addEvent('local-push', {
-      batchSize: encodedEventDefs.length,
-      mergeResult: TRACE_VERBOSE ? JSON.stringify(mergeResult) : undefined,
-    })
+    if (TRACE_VERBOSE) yield* Effect.annotateCurrentSpan({ mergeResult: JSON.stringify(mergeResult) })
 
     if (mergeResult._tag !== 'advance') {
-      return shouldNeverHappen(`Expected advance, got ${mergeResult._tag}`)
+      return yield* Effect.die(new Error(`Expected advance, got ${mergeResult._tag}`))
     }
 
     syncStateRef.current = mergeResult.newSyncState
-    syncStateUpdateQueue.offer(mergeResult.newSyncState).pipe(Effect.runSync)
+    yield* syncStateUpdateQueue.offer(mergeResult.newSyncState)
 
     // Materialize events to state
     const writeTables = new Set<string>()
@@ -135,16 +138,18 @@ export const makeClientSessionSyncProcessor = ({
       for (const table of newWriteTables) {
         writeTables.add(table)
       }
-      event.meta.sessionChangeset = sessionChangeset
-      event.meta.materializerHashSession = materializerHash
+      yield* Effect.sync(() => {
+        event.meta.sessionChangeset = sessionChangeset
+        event.meta.materializerHashSession = materializerHash
+      })
     }
 
     // Trigger push to leader
     // console.debug('pushToLeader', encodedEventDefs.length, ...encodedEventDefs.map((_) => _.toJSON()))
-    BucketQueue.offerAll(leaderPushQueue, encodedEventDefs).pipe(Effect.runSync)
+    yield* BucketQueue.offerAll(leaderPushQueue, encodedEventDefs)
 
     return { writeTables }
-  }
+  })
 
   const debugInfo = {
     rebaseCount: 0,
@@ -340,9 +345,12 @@ export interface ClientSessionSyncProcessor {
   push: (
     batch: ReadonlyArray<LiveStoreEvent.PartialAnyDecoded>,
     options: { otelContext: otel.Context },
-  ) => {
-    writeTables: Set<string>
-  }
+  ) => Effect.Effect<
+    {
+      writeTables: Set<string>
+    },
+    never
+  >
   boot: Effect.Effect<void, UnexpectedError, Scope.Scope>
   /**
    * Only used for debugging / observability.

packages/@livestore/livestore/src/live-queries/__snapshots__/db-query.test.ts.snap

@@ -409,6 +409,12 @@ exports[`otel > otel 3`] = `
     ",
       },
     },
+    {
+      "_name": "client-session-sync-processor:push",
+      "attributes": {
+        "batchSize": 1,
+      },
+    },
     {
       "_name": "@livestore/common:LeaderSyncProcessor:push",
       "attributes": {
@@ -684,6 +690,12 @@ exports[`otel > with thunks 7`] = `
     ",
       },
     },
+    {
+      "_name": "client-session-sync-processor:push",
+      "attributes": {
+        "batchSize": 1,
+      },
+    },
     {
       "_name": "@livestore/common:LeaderSyncProcessor:push",
       "attributes": {
@@ -793,6 +805,12 @@ exports[`otel > with thunks with query builder and without labels 3`] = `
     ",
       },
     },
+    {
+      "_name": "client-session-sync-processor:push",
+      "attributes": {
+        "batchSize": 1,
+      },
+    },
     {
       "_name": "@livestore/common:LeaderSyncProcessor:push",
       "attributes": {

packages/@livestore/livestore/src/store/store.ts

@@ -589,7 +589,9 @@ export class Store<TSchema extends LiveStoreSchema = LiveStoreSchema, TContext =
           // Materialize events to state
           const { writeTables } = (() => {
             try {
-              const materializeEvents = () => this.syncProcessor.push(events, { otelContext })
+              const materializeEvents = () => {
+                return Runtime.runSync(this.effectContext.runtime, this.syncProcessor.push(events, { otelContext }))
+              }
 
               if (events.length > 1) {
                 return this.sqliteDbWrapper.txn(materializeEvents)

packages/@livestore/react/src/__snapshots__/useClientDocument.test.tsx.snap

@@ -20,6 +20,18 @@ exports[`useClientDocument > otel > should update the data based on component ke
     ",
       },
     },
+    {
+      "_name": "client-session-sync-processor:push",
+      "attributes": {
+        "batchSize": 1,
+      },
+    },
+    {
+      "_name": "client-session-sync-processor:push",
+      "attributes": {
+        "batchSize": 1,
+      },
+    },
     {
       "_name": "@livestore/common:LeaderSyncProcessor:push",
       "attributes": {
@@ -246,6 +258,18 @@ exports[`useClientDocument > otel > should update the data based on component ke
     ",
       },
     },
+    {
+      "_name": "client-session-sync-processor:push",
+      "attributes": {
+        "batchSize": 1,
+      },
+    },
+    {
+      "_name": "client-session-sync-processor:push",
+      "attributes": {
+        "batchSize": 1,
+      },
+    },
     {
       "_name": "@livestore/common:LeaderSyncProcessor:push",
       "attributes": {