Skip to content

Commit e63a429

Browse files
IGassmannIGassmann
committed
refactor: migrate ClientSessionSyncProcessor.materializeEvent to Effect
1 parent a2bc64d commit e63a429

File tree

4 files changed

+95
-61
lines changed

4 files changed

+95
-61
lines changed

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

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -39,11 +39,11 @@ export const makeClientSessionSyncProcessor = ({
3939
materializeEvent: (
4040
eventDecoded: LiveStoreEvent.PartialAnyDecoded,
4141
options: { otelContext: otel.Context; withChangeset: boolean; materializerHashLeader: Option.Option<number> },
42-
) => {
42+
) => Effect.Effect<{
4343
writeTables: Set<string>
4444
sessionChangeset: { _tag: 'sessionChangeset'; data: Uint8Array; debug: any } | { _tag: 'no-op' } | { _tag: 'unset' }
4545
materializerHash: Option.Option<number>
46-
}
46+
}>
4747
rollback: (changeset: Uint8Array) => void
4848
refreshTables: (tables: Set<string>) => void
4949
span: otel.Span
@@ -130,7 +130,7 @@ export const makeClientSessionSyncProcessor = ({
130130
writeTables: newWriteTables,
131131
sessionChangeset,
132132
materializerHash,
133-
} = materializeEvent(decodedEventDef, {
133+
} = yield* materializeEvent(decodedEventDef, {
134134
otelContext,
135135
withChangeset: true,
136136
materializerHashLeader: Option.none(),
@@ -284,7 +284,7 @@ export const makeClientSessionSyncProcessor = ({
284284
writeTables: newWriteTables,
285285
sessionChangeset,
286286
materializerHash,
287-
} = materializeEvent(decodedEventDef, {
287+
} = yield* materializeEvent(decodedEventDef, {
288288
otelContext,
289289
withChangeset: true,
290290
materializerHashLeader: event.meta.materializerHashLeader,

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

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,11 @@ exports[`otel > otel 3`] = `
2525
"attributes": {
2626
"batchSize": 1,
2727
},
28+
"children": [
29+
{
30+
"_name": "client-session-sync-processor:materialize-event",
31+
},
32+
],
2833
},
2934
{
3035
"_name": "@livestore/common:LeaderSyncProcessor:push",
@@ -306,6 +311,11 @@ exports[`otel > with thunks 7`] = `
306311
"attributes": {
307312
"batchSize": 1,
308313
},
314+
"children": [
315+
{
316+
"_name": "client-session-sync-processor:materialize-event",
317+
},
318+
],
309319
},
310320
{
311321
"_name": "@livestore/common:LeaderSyncProcessor:push",
@@ -421,6 +431,11 @@ exports[`otel > with thunks with query builder and without labels 3`] = `
421431
"attributes": {
422432
"batchSize": 1,
423433
},
434+
"children": [
435+
{
436+
"_name": "client-session-sync-processor:materialize-event",
437+
},
438+
],
424439
},
425440
{
426441
"_name": "@livestore/common:LeaderSyncProcessor:push",

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

Lines changed: 56 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -115,70 +115,69 @@ export class Store<TSchema extends LiveStoreSchema = LiveStoreSchema, TContext =
115115
schema,
116116
clientSession,
117117
runtime: effectContext.runtime,
118-
materializeEvent: (eventDecoded, { otelContext, withChangeset, materializerHashLeader }) => {
119-
const { eventDef, materializer } = getEventDef(schema, eventDecoded.name)
120-
121-
const execArgsArr = getExecStatementsFromMaterializer({
122-
eventDef,
123-
materializer,
124-
dbState: this.sqliteDbWrapper,
125-
event: { decoded: eventDecoded, encoded: undefined },
126-
})
127-
128-
const materializerHash = isDevEnv() ? Option.some(hashMaterializerResults(execArgsArr)) : Option.none()
129-
130-
if (
131-
materializerHashLeader._tag === 'Some' &&
132-
materializerHash._tag === 'Some' &&
133-
materializerHashLeader.value !== materializerHash.value
134-
) {
135-
void this.shutdown(
136-
Cause.fail(
137-
UnexpectedError.make({
118+
materializeEvent: Effect.fn('client-session-sync-processor:materialize-event')(
119+
(eventDecoded, { otelContext, withChangeset, materializerHashLeader }) =>
120+
Effect.gen(this, function* () {
121+
const { eventDef, materializer } = getEventDef(schema, eventDecoded.name)
122+
123+
const execArgsArr = getExecStatementsFromMaterializer({
124+
eventDef,
125+
materializer,
126+
dbState: this.sqliteDbWrapper,
127+
event: { decoded: eventDecoded, encoded: undefined },
128+
})
129+
130+
const materializerHash = isDevEnv() ? Option.some(hashMaterializerResults(execArgsArr)) : Option.none()
131+
132+
if (
133+
materializerHashLeader._tag === 'Some' &&
134+
materializerHash._tag === 'Some' &&
135+
materializerHashLeader.value !== materializerHash.value
136+
) {
137+
const error = UnexpectedError.make({
138138
cause: `Materializer hash mismatch detected for event "${eventDecoded.name}".`,
139139
note: `Please make sure your event materializer is a pure function without side effects.`,
140-
}),
141-
),
142-
)
143-
}
140+
})
144141

145-
const writeTablesForEvent = new Set<string>()
142+
// Fork the shutdown effect to run in the background as a daemon,
143+
// ensuring it's not interrupted.
144+
yield* Effect.forkDaemon(this.clientSession.shutdown(Cause.fail(error)))
146145

147-
const exec = () => {
148-
for (const {
149-
statementSql,
150-
bindValues,
151-
writeTables = this.sqliteDbWrapper.getTablesUsed(statementSql),
152-
} of execArgsArr) {
153-
try {
154-
this.sqliteDbWrapper.cachedExecute(statementSql, bindValues, { otelContext, writeTables })
155-
} catch (cause) {
156-
throw UnexpectedError.make({
157-
cause,
158-
note: `Error executing materializer for event "${eventDecoded.name}".\nStatement: ${statementSql}\nBind values: ${JSON.stringify(bindValues)}`,
159-
})
146+
// TODO: we should probably handle this more gracefully using Effect’s error channel
160147
}
161148

162-
// durationMsTotal += durationMs
163-
for (const table of writeTables) {
164-
writeTablesForEvent.add(table)
165-
}
166-
}
167-
}
168-
169-
let sessionChangeset:
170-
| { _tag: 'sessionChangeset'; data: Uint8Array; debug: any }
171-
| { _tag: 'no-op' }
172-
| { _tag: 'unset' } = { _tag: 'unset' }
149+
return yield* Effect.sync(() => {
150+
const writeTablesForEvent = new Set<string>()
151+
152+
const exec = () => {
153+
for (const {
154+
statementSql,
155+
bindValues,
156+
writeTables = this.sqliteDbWrapper.getTablesUsed(statementSql),
157+
} of execArgsArr) {
158+
this.sqliteDbWrapper.cachedExecute(statementSql, bindValues, { otelContext, writeTables })
159+
160+
// durationMsTotal += durationMs
161+
for (const table of writeTables) {
162+
writeTablesForEvent.add(table)
163+
}
164+
}
165+
}
173166

174-
if (withChangeset === true) {
175-
sessionChangeset = this.sqliteDbWrapper.withChangeset(exec).changeset
176-
} else {
177-
exec()
178-
}
167+
let sessionChangeset:
168+
| { _tag: 'sessionChangeset'; data: Uint8Array; debug: any }
169+
| { _tag: 'no-op' }
170+
| { _tag: 'unset' } = { _tag: 'unset' }
171+
if (withChangeset === true) {
172+
sessionChangeset = this.sqliteDbWrapper.withChangeset(exec).changeset
173+
} else {
174+
exec()
175+
}
179176

180-
return { writeTables: writeTablesForEvent, sessionChangeset, materializerHash }
181-
},
177+
return { writeTables: writeTablesForEvent, sessionChangeset, materializerHash }
178+
})
179+
}),
180+
),
182181
rollback: (changeset) => {
183182
this.sqliteDbWrapper.rollback(changeset)
184183
},
@@ -584,14 +583,14 @@ export class Store<TSchema extends LiveStoreSchema = LiveStoreSchema, TContext =
584583
const otelContext = otel.trace.setSpan(otel.context.active(), span)
585584

586585
try {
587-
// Materialize events to state
588586
const { writeTables } = (() => {
589587
try {
590588
const materializeEvents = () => {
591589
return Runtime.runSync(this.effectContext.runtime, this.syncProcessor.push(events, { otelContext }))
592590
}
593591

594592
if (events.length > 1) {
593+
// TODO: what to do about leader transaction here?
595594
return this.sqliteDbWrapper.txn(materializeEvents)
596595
} else {
597596
return materializeEvents()

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

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,12 +25,22 @@ exports[`useClientDocument > otel > should update the data based on component ke
2525
"attributes": {
2626
"batchSize": 1,
2727
},
28+
"children": [
29+
{
30+
"_name": "client-session-sync-processor:materialize-event",
31+
},
32+
],
2833
},
2934
{
3035
"_name": "client-session-sync-processor:push",
3136
"attributes": {
3237
"batchSize": 1,
3338
},
39+
"children": [
40+
{
41+
"_name": "client-session-sync-processor:materialize-event",
42+
},
43+
],
3444
},
3545
{
3646
"_name": "@livestore/common:LeaderSyncProcessor:push",
@@ -263,12 +273,22 @@ exports[`useClientDocument > otel > should update the data based on component ke
263273
"attributes": {
264274
"batchSize": 1,
265275
},
276+
"children": [
277+
{
278+
"_name": "client-session-sync-processor:materialize-event",
279+
},
280+
],
266281
},
267282
{
268283
"_name": "client-session-sync-processor:push",
269284
"attributes": {
270285
"batchSize": 1,
271286
},
287+
"children": [
288+
{
289+
"_name": "client-session-sync-processor:materialize-event",
290+
},
291+
],
272292
},
273293
{
274294
"_name": "@livestore/common:LeaderSyncProcessor:push",

0 commit comments

Comments
 (0)