Merge pull request #753 from slashv/fix-513-event-streaming

Author: schickling

docs/src/content/_assets/code/reference/events/stream.ts

@@ -0,0 +1,7 @@
+import type { Store } from '@livestore/livestore'
+
+declare const store: Store
+
+for await (const event of store.events()) {
+  console.log('event from leader', event)
+}

docs/src/content/_assets/code/reference/store/stream-events.ts

@@ -0,0 +1,20 @@
+import type { Store } from '@livestore/livestore'
+
+declare const store: Store
+
+// Run once
+for await (const event of store.events()) {
+  console.log('event from leader', event)
+}
+
+// Continuos stream
+const iterator = store.events()[Symbol.asyncIterator]()
+try {
+  while (true) {
+    const { value, done } = await iterator.next()
+    if (done) break
+    console.log('event from stream:', value)
+  }
+} finally {
+  await iterator.return?.()
+}

docs/src/content/docs/building-with-livestore/events.mdx

@@ -11,6 +11,7 @@ import { EventsVisualizer } from '../../../components/EventsVisualizer'
 import EventNodesDiagram from '../../_assets/diagrams/event-nodes.tldr?tldraw'
 import UnknownEventHandlingSnippet from '../../_assets/code/reference/events/unknown-event-handling.ts?snippet'
 import EventTypeNamespacesSnippet from '../../_assets/code/reference/events/event-type-namespaces.ts?snippet'
+import StreamEventsSnippet from '../../_assets/code/reference/events/stream.ts?snippet'
 
 ## Event definitions
 
@@ -31,6 +32,12 @@ Events will be synced across clients and materialized into state (i.e. SQLite ta
 
 <CommitEventSnippet />
 
+### Streaming events
+
+Currently only events confirmed by the sync backend are supported.
+
+<StreamEventsSnippet />
+
 ### Best practices
 
 - It's strongly recommended to use past-tense event names (e.g. `todoCreated`/`createdTodo` instead of `todoCreate`/`createTodo`) to indicate something already occurred.

docs/src/content/docs/building-with-livestore/store.mdx

@@ -9,6 +9,7 @@ import CreateStoreSnippet from '../../_assets/code/reference/store/create-store.
 import QueryDataSnippet from '../../_assets/code/reference/store/query-data.ts?snippet'
 import ShutdownSnippet from '../../_assets/code/reference/store/shutdown.ts?snippet'
 import SubscribeSnippet from '../../_assets/code/reference/store/subscribe.ts?snippet'
+import StreamEventsSnippet from '../../_assets/code/reference/store/stream-events.ts?snippet'
 
 The `Store` is the most common way to interact with LiveStore from your application code. It provides a way to query data, commit events, and subscribe to data changes.
 
@@ -32,6 +33,12 @@ For how to create a store in React, see the [React integration docs](/framework-
 
 <CommitEventSnippet />
 
+### Streaming events
+
+Currently only events confirmed by the sync backend are supported.
+
+<StreamEventsSnippet />
+
 ### Shutting down a store
 
 LiveStore provides two APIs for shutting down a store:
@@ -58,4 +65,4 @@ __debugLiveStore.default._dev.hardReset()
 
 // See the current sync state
 __debugLiveStore.default._dev.syncStates()
-```   
+```

packages/@livestore/adapter-cloudflare/src/make-adapter.ts

@@ -7,7 +7,13 @@ import {
   type SyncOptions,
   UnknownError,
 } from '@livestore/common'
-import { type DevtoolsOptions, Eventlog, LeaderThreadCtx, makeLeaderThreadLayer } from '@livestore/common/leader-thread'
+import {
+  type DevtoolsOptions,
+  Eventlog,
+  LeaderThreadCtx,
+  makeLeaderThreadLayer,
+  streamEventsWithSyncState,
+} from '@livestore/common/leader-thread'
 import type { CfTypes } from '@livestore/common-cf'
 import { LiveStoreEvent } from '@livestore/livestore'
 import { sqliteDbFactory } from '@livestore/sqlite-wasm/cf'
@@ -112,6 +118,12 @@ export const makeAdapter =
                   batch.map((item) => new LiveStoreEvent.Client.EncodedWithMeta(item)),
                   { waitForProcessing: true },
                 ),
+              stream: (options) =>
+                streamEventsWithSyncState({
+                  dbEventlog,
+                  syncState: syncProcessor.syncState,
+                  options,
+                }),
             },
             initialState: { leaderHead: initialLeaderHead, migrationsReport: initialState.migrationsReport },
             export: Effect.sync(() => dbState.export()),

packages/@livestore/adapter-expo/src/index.ts

@@ -13,7 +13,12 @@ import {
   UnknownError,
 } from '@livestore/common'
 import type { DevtoolsOptions, LeaderSqliteDb } from '@livestore/common/leader-thread'
-import { Eventlog, LeaderThreadCtx, makeLeaderThreadLayer } from '@livestore/common/leader-thread'
+import {
+  Eventlog,
+  LeaderThreadCtx,
+  makeLeaderThreadLayer,
+  streamEventsWithSyncState,
+} from '@livestore/common/leader-thread'
 import type { LiveStoreSchema } from '@livestore/common/schema'
 import { LiveStoreEvent } from '@livestore/common/schema'
 import { shouldNeverHappen } from '@livestore/utils'
@@ -300,6 +305,12 @@ const makeLeaderThread = ({
                 { waitForProcessing: true },
               )
               .pipe(Effect.provide(layer), Effect.scoped),
+          stream: (options) =>
+            streamEventsWithSyncState({
+              dbEventlog,
+              syncState: syncProcessor.syncState,
+              options,
+            }),
         },
         initialState: { leaderHead: initialLeaderHead, migrationsReport: initialState.migrationsReport },
         export: Effect.sync(() => db.export()),

packages/@livestore/adapter-node/src/client-session/adapter.ts

@@ -13,7 +13,7 @@ import {
   type SyncOptions,
   UnknownError,
 } from '@livestore/common'
-import { Eventlog, LeaderThreadCtx } from '@livestore/common/leader-thread'
+import { Eventlog, LeaderThreadCtx, streamEventsWithSyncState } from '@livestore/common/leader-thread'
 import type { LiveStoreSchema } from '@livestore/common/schema'
 import { LiveStoreEvent } from '@livestore/common/schema'
 import { loadSqlite3Wasm } from '@livestore/sqlite-wasm/load-wasm'
@@ -406,6 +406,12 @@ const makeLocalLeaderThread = ({
                 batch.map((item) => new LiveStoreEvent.Client.EncodedWithMeta(item)),
                 { waitForProcessing: true },
               ),
+            stream: (options) =>
+              streamEventsWithSyncState({
+                dbEventlog,
+                syncState: syncProcessor.syncState,
+                options,
+              }),
           },
           initialState: { leaderHead: initialLeaderHead, migrationsReport: initialState.migrationsReport },
           export: Effect.sync(() => dbState.export()),
@@ -546,6 +552,11 @@ const makeWorkerLeaderThread = ({
                 attributes: { batchSize: batch.length },
               }),
             ),
+          stream: (options) =>
+            runInWorkerStream(new WorkerSchema.LeaderWorkerInnerStreamEvents(options)).pipe(
+              Stream.withSpan('@livestore/adapter-node:client-session:streamEvents'),
+              Stream.orDie,
+            ),
         },
         initialState: {
           leaderHead: initialLeaderHead,

packages/@livestore/adapter-node/src/make-leader-worker.ts

@@ -9,7 +9,8 @@ if (process.execArgv.includes('--inspect')) {
 
 import type { SyncOptions } from '@livestore/common'
 import { LogConfig, UnknownError } from '@livestore/common'
-import { Eventlog, LeaderThreadCtx } from '@livestore/common/leader-thread'
+import type { StreamEventsOptions } from '@livestore/common/leader-thread'
+import { Eventlog, LeaderThreadCtx, streamEventsWithSyncState } from '@livestore/common/leader-thread'
 import type { LiveStoreSchema } from '@livestore/common/schema'
 import { LiveStoreEvent } from '@livestore/common/schema'
 import { loadSqlite3Wasm } from '@livestore/sqlite-wasm/load-wasm'
@@ -86,6 +87,20 @@ export const makeWorkerEffect = (options: WorkerOptions) => {
         const { syncProcessor } = yield* LeaderThreadCtx
         return syncProcessor.pull({ cursor })
       }).pipe(Stream.unwrapScoped),
+    StreamEvents: (options: WorkerSchema.LeaderWorkerInnerStreamEvents) =>
+      LeaderThreadCtx.pipe(
+        Effect.map(({ dbEventlog, syncProcessor }) => {
+          const { _tag: _ignored, ...payload } = options
+          const streamOptions = payload as StreamEventsOptions
+          return streamEventsWithSyncState({
+            dbEventlog,
+            syncState: syncProcessor.syncState,
+            options: streamOptions,
+          })
+        }),
+        Stream.unwrapScoped,
+        Stream.withSpan('@livestore/adapter-node:worker:StreamEvents'),
+      ),
     Export: () =>
       Effect.andThen(LeaderThreadCtx, (_) => _.dbState.export()).pipe(
         UnknownError.mapToUnknownError,

packages/@livestore/adapter-node/src/worker-schema.ts

@@ -7,6 +7,7 @@ import {
   SyncState,
   UnknownError,
 } from '@livestore/common'
+import { StreamEventsOptionsFields } from '@livestore/common/leader-thread'
 import { EventSequenceNumber, LiveStoreEvent } from '@livestore/common/schema'
 import { Schema, Transferable } from '@livestore/utils/effect'
 
@@ -111,6 +112,15 @@ export class LeaderWorkerInnerPullStream extends Schema.TaggedRequest<LeaderWork
   failure: UnknownError,
 }) {}
 
+export class LeaderWorkerInnerStreamEvents extends Schema.TaggedRequest<LeaderWorkerInnerStreamEvents>()(
+  'StreamEvents',
+  {
+    payload: StreamEventsOptionsFields,
+    success: LiveStoreEvent.Client.Encoded,
+    failure: UnknownError,
+  },
+) {}
+
 export class LeaderWorkerInnerPushToLeader extends Schema.TaggedRequest<LeaderWorkerInnerPushToLeader>()(
   'PushToLeader',
   {
@@ -215,6 +225,7 @@ export const LeaderWorkerInnerRequest = Schema.Union(
   LeaderWorkerInnerInitialMessage,
   LeaderWorkerInnerBootStatusStream,
   LeaderWorkerInnerPullStream,
+  LeaderWorkerInnerStreamEvents,
   LeaderWorkerInnerPushToLeader,
   LeaderWorkerInnerExport,
   LeaderWorkerInnerGetRecreateSnapshot,

packages/@livestore/adapter-web/src/in-memory/in-memory-adapter.ts

@@ -9,7 +9,13 @@ import {
   UnknownError,
 } from '@livestore/common'
 import type { DevtoolsOptions, LeaderSqliteDb } from '@livestore/common/leader-thread'
-import { configureConnection, Eventlog, LeaderThreadCtx, makeLeaderThreadLayer } from '@livestore/common/leader-thread'
+import {
+  configureConnection,
+  Eventlog,
+  LeaderThreadCtx,
+  makeLeaderThreadLayer,
+  streamEventsWithSyncState,
+} from '@livestore/common/leader-thread'
 import type { LiveStoreSchema } from '@livestore/common/schema'
 import { LiveStoreEvent } from '@livestore/common/schema'
 import * as DevtoolsWeb from '@livestore/devtools-web-common/web-channel'
@@ -265,6 +271,12 @@ const makeLeaderThread = ({
               batch.map((item) => new LiveStoreEvent.Client.EncodedWithMeta(item)),
               { waitForProcessing: true },
             ),
+          stream: (options) =>
+            streamEventsWithSyncState({
+              dbEventlog,
+              syncState: syncProcessor.syncState,
+              options,
+            }),
         },
         initialState: { leaderHead: initialLeaderHead, migrationsReport: initialState.migrationsReport },
         export: Effect.sync(() => dbState.export()),