chore: sync changes

Author: IGassmann

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

@@ -4,11 +4,10 @@ import { Option, type Runtime, type Scope } from '@livestore/utils/effect'
 import { BucketQueue, Effect, FiberHandle, Queue, Schema, Stream, Subscribable } from '@livestore/utils/effect'
 import type * as otel from '@opentelemetry/api'
 
-import type { ClientSession, UnexpectedError } from '../adapter-types.js'
+import { type ClientSession, SyncError, type UnexpectedError } from '../adapter-types.js'
 import * as EventSequenceNumber from '../schema/EventSequenceNumber.js'
 import * as LiveStoreEvent from '../schema/LiveStoreEvent.js'
-import { getEventDef, type LiveStoreSchema, SystemTables } from '../schema/mod.js'
-import { sql } from '../util.js'
+import { getEventDef, type LiveStoreSchema } from '../schema/mod.js'
 import * as SyncState from './syncstate.js'
 
 /**
@@ -21,6 +20,10 @@ import * as SyncState from './syncstate.js'
  * - We might need to make the rebase behaviour configurable e.g. to let users manually trigger a rebase
  *
  * Longer term we should evalutate whether we can unify the ClientSessionSyncProcessor with the LeaderSyncProcessor.
+ *
+ * The session and leader sync processor are different in the following ways:
+ * - The leader sync processor pulls regular LiveStore events, while the session sync processor pulls SyncState.PayloadUpstream items
+ * - The session sync processor has no downstream nodes.
  */
 export const makeClientSessionSyncProcessor = ({
   schema,
@@ -37,7 +40,7 @@ export const makeClientSessionSyncProcessor = ({
   clientSession: ClientSession
   runtime: Runtime.Runtime<Scope.Scope>
   materializeEvent: (
-    eventDecoded: LiveStoreEvent.PartialAnyDecoded,
+    eventDecoded: LiveStoreEvent.AnyDecoded,
     options: { withChangeset: boolean; materializerHashLeader: Option.Option<number> },
   ) => Effect.Effect<{
     writeTables: Set<string>
@@ -82,7 +85,10 @@ export const makeClientSessionSyncProcessor = ({
     let baseEventSequenceNumber = syncStateRef.current.localHead
     const encodedEventDefs = batch.map(({ name, args }) => {
       const eventDef = getEventDef(schema, name)
-      const nextNumPair = EventSequenceNumber.nextPair(baseEventSequenceNumber, eventDef.eventDef.options.clientOnly)
+      const nextNumPair = EventSequenceNumber.nextPair({
+        seqNum: baseEventSequenceNumber,
+        isClient: eventDef.eventDef.options.clientOnly,
+      })
       baseEventSequenceNumber = nextNumPair.seqNum
       return new LiveStoreEvent.EncodedWithMeta(
         Schema.encodeUnknownSync(eventSchema)({
@@ -106,13 +112,13 @@ export const makeClientSessionSyncProcessor = ({
     )
 
     if (mergeResult._tag === 'unexpected-error') {
-      return yield* Effect.die(new Error(`Unexpected error in client-session-sync-processor: ${mergeResult.cause}`))
+      return yield* new SyncError({ cause: mergeResult.message })
     }
 
     if (TRACE_VERBOSE) yield* Effect.annotateCurrentSpan({ mergeResult: JSON.stringify(mergeResult) })
 
     if (mergeResult._tag !== 'advance') {
-      return yield* Effect.die(new Error(`Expected advance, got ${mergeResult._tag}`))
+      return yield* new SyncError({ cause: `Expected advance, got ${mergeResult._tag}` })
     }
 
     syncStateRef.current = mergeResult.newSyncState
@@ -183,18 +189,11 @@ export const makeClientSessionSyncProcessor = ({
 
     yield* FiberHandle.run(leaderPushingFiberHandle, backgroundLeaderPushing)
 
-    const getMergeCounter = () =>
-      clientSession.sqliteDb.select<{ mergeCounter: number }>(
-        sql`SELECT mergeCounter FROM ${SystemTables.LEADER_MERGE_COUNTER_TABLE} WHERE id = 0`,
-      )[0]?.mergeCounter ?? 0
-
     // NOTE We need to lazily call `.pull` as we want the cursor to be updated
     yield* Stream.suspend(() =>
-      clientSession.leaderThread.events.pull({
-        cursor: { mergeCounter: getMergeCounter(), eventNum: syncStateRef.current.localHead },
-      }),
+      clientSession.leaderThread.events.pull({ cursor: syncStateRef.current.upstreamHead }),
     ).pipe(
-      Stream.tap(({ payload, mergeCounter: leaderMergeCounter }) =>
+      Stream.tap(({ payload }) =>
         Effect.gen(function* () {
           // yield* Effect.logDebug('ClientSessionSyncProcessor:pull', payload)
 
@@ -210,13 +209,13 @@ export const makeClientSessionSyncProcessor = ({
           })
 
           if (mergeResult._tag === 'unexpected-error') {
-            return yield* Effect.fail(mergeResult.cause)
+            return yield* new SyncError({ cause: mergeResult.message })
           } else if (mergeResult._tag === 'reject') {
             return shouldNeverHappen('Unexpected reject in client-session-sync-processor', mergeResult)
           }
 
           syncStateRef.current = mergeResult.newSyncState
-          syncStateUpdateQueue.offer(mergeResult.newSyncState).pipe(Effect.runSync)
+          yield* syncStateUpdateQueue.offer(mergeResult.newSyncState)
 
           if (mergeResult._tag === 'rebase') {
             span.addEvent('merge:pull:rebase', {
@@ -225,7 +224,7 @@ export const makeClientSessionSyncProcessor = ({
               newEventsCount: mergeResult.newEvents.length,
               rollbackCount: mergeResult.rollbackEvents.length,
               res: TRACE_VERBOSE ? JSON.stringify(mergeResult) : undefined,
-              leaderMergeCounter,
+              rebaseGeneration: mergeResult.newSyncState.localHead.rebaseGeneration,
             })
 
             debugInfo.rebaseCount++
@@ -242,7 +241,6 @@ export const makeClientSessionSyncProcessor = ({
                 'merge:pull:rebase: rollback',
                 mergeResult.rollbackEvents.length,
                 ...mergeResult.rollbackEvents.slice(0, 10).map((_) => _.toJSON()),
-                { leaderMergeCounter },
               ).pipe(Effect.provide(runtime), Effect.runSync)
             }
 
@@ -262,7 +260,6 @@ export const makeClientSessionSyncProcessor = ({
               payload: TRACE_VERBOSE ? JSON.stringify(payload) : undefined,
               newEventsCount: mergeResult.newEvents.length,
               res: TRACE_VERBOSE ? JSON.stringify(mergeResult) : undefined,
-              leaderMergeCounter,
             })
 
             debugInfo.advanceCount++
@@ -339,7 +336,7 @@ export interface ClientSessionSyncProcessor {
     {
       writeTables: Set<string>
     },
-    never
+    SyncError
   >
   boot: Effect.Effect<void, UnexpectedError, Scope.Scope>
   /**

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

@@ -1,12 +1,7 @@
 import {
+  type Bindable,
   type ClientSession,
   type ClientSessionSyncProcessor,
-  type ParamsObject,
-  type PreparedBindValues,
-  type QueryBuilder,
-  UnexpectedError,
-} from '@livestore/common'
-import {
   Devtools,
   getDurationMsFromSpan,
   getExecStatementsFromMaterializer,
@@ -16,9 +11,12 @@ import {
   isQueryBuilder,
   liveStoreVersion,
   makeClientSessionSyncProcessor,
+  type PreparedBindValues,
   prepareBindValues,
+  type QueryBuilder,
   QueryBuilderAstSymbol,
   replaceSessionIdSymbol,
+  UnexpectedError,
 } from '@livestore/common'
 import type { LiveStoreSchema } from '@livestore/common/schema'
 import { getEventDef, LiveStoreEvent, SystemTables } from '@livestore/common/schema'
@@ -178,6 +176,8 @@ export class Store<TSchema extends LiveStoreSchema = LiveStoreSchema.Any, TConte
                   for (const table of writeTables) {
                     writeTablesForEvent.add(table)
                   }
+
+                  this.sqliteDbWrapper.debug.head = eventDecoded.seqNum
                 }
               }
 
@@ -441,15 +441,19 @@ export class Store<TSchema extends LiveStoreSchema = LiveStoreSchema.Any, TConte
       | LiveQuery<TResult>
       | LiveQueryDef<TResult>
       | SignalDef<TResult>
-      | { query: string; bindValues: ParamsObject },
+      | { query: string; bindValues: Bindable; schema?: Schema.Schema<TResult> },
     options?: { otelContext?: otel.Context; debugRefreshReason?: RefreshReason },
   ): TResult => {
     this.checkShutdown('query')
 
     if (typeof query === 'object' && 'query' in query && 'bindValues' in query) {
-      return this.sqliteDbWrapper.cachedSelect(query.query, prepareBindValues(query.bindValues, query.query), {
+      const res = this.sqliteDbWrapper.cachedSelect(query.query, prepareBindValues(query.bindValues, query.query), {
         otelContext: options?.otelContext,
       }) as any
+      if (query.schema) {
+        return Schema.decodeSync(query.schema)(res)
+      }
+      return res
     } else if (isQueryBuilder(query)) {
       const ast = query[QueryBuilderAstSymbol]
       if (ast._tag === 'RowQuery') {
@@ -626,14 +630,14 @@ export class Store<TSchema extends LiveStoreSchema = LiveStoreSchema.Any, TConte
         const otelContext = otel.trace.setSpan(otel.context.active(), span)
 
         try {
+          // Materialize events to state
           const { writeTables } = (() => {
             try {
               const materializeEvents = () => {
                 return Runtime.runSync(this.effectContext.runtime, this.syncProcessor.push(events))
               }
 
               if (events.length > 1) {
-                // TODO: what to do about leader transaction here?
                 return this.sqliteDbWrapper.txn(materializeEvents)
               } else {
                 return materializeEvents()