1import { SqliteClient } from "@effect/sql-sqlite-node"
2import { assert, describe, it } from "@effect/vitest"
3import { Effect, Option } from "effect"
4import * as EventJournal from "effect/eventlog/EventJournal"
5import * as SqlEventJournal from "effect/eventlog/SqlEventJournal"
6import { Reactivity } from "effect/reactivity"
7import * as SqlClient from "effect/sql/SqlClient"
8
9const makeJournal = Effect.gen(function*() {
10 const sql = yield* SqliteClient.make({ filename: ":memory:" })
11 return yield* SqlEventJournal.make().pipe(
12 Effect.provideService(SqlClient.SqlClient, sql)
13 )
14}).pipe(Effect.provide(Reactivity.layer))
15
16describe("SqlEventJournal", () => {
17 it.effect("preserves write callback error identity", () =>
18 Effect.gen(function*() {
19 const journal = yield* makeJournal
20 const error = new Error("callback failed")
21 const actual = yield* Effect.flip(journal.write({
22 event: "Repro",
23 primaryKey: "key",
24 payload: new Uint8Array([1]),
25 effect: () => Effect.fail(error)
26 }))
27 assert.strictEqual(actual, error)
28 }))
29
30 it.effect("preserves remote callback error identity", () =>
31 Effect.gen(function*() {
32 const journal = yield* makeJournal
33 yield* journal.write({
34 event: "Repro",
35 primaryKey: "key",
36 payload: new Uint8Array([1]),
37 effect: () => Effect.void
38 })
39 const error = new Error("callback failed")
40 const actual = yield* Effect.flip(
41 journal.withRemoteUncommited(EventJournal.makeRemoteIdUnsafe(), () => Effect.fail(error))
42 )
43 assert.strictEqual(actual, error)
44 }))
45
46 it.effect("commits only after the write callback succeeds", () =>
47 Effect.gen(function*() {
48 const sql = yield* SqliteClient.make({ filename: ":memory:" })
49 const journal = yield* SqlEventJournal.make().pipe(Effect.provideService(SqlClient.SqlClient, sql))
50 yield* Effect.exit(journal.write({
51 event: "Repro",
52 primaryKey: "key",
53 payload: new Uint8Array([1]),
54 effect: () => Effect.fail("callback failed")
55 }))
56 assert.deepStrictEqual(yield* journal.entries, [])
57 }).pipe(Effect.provide(Reactivity.layer)))
58
59 it.effect("writes and reads entries", () =>
60 Effect.gen(function*() {
61 const journal = yield* makeJournal
62 let createdAt = 0
63 yield* journal.write({
64 event: "UserCreated",
65 primaryKey: "user-1",
66 payload: new Uint8Array([1]),
67 effect: (entry) =>
68 Effect.sync(() => {
69 createdAt = entry.createdAtMillis
70 })
71 })
72 const entries = yield* journal.entries
73 assert.strictEqual(entries.length, 1)
74 assert.strictEqual(entries[0].event, "UserCreated")
75 assert.strictEqual(entries[0].createdAtMillis, createdAt)
76 }))
77
78 it.effect("writes remote entries and sequences", () =>
79 Effect.gen(function*() {
80 const journal = yield* makeJournal
81 const remoteId = EventJournal.makeRemoteIdUnsafe()
82 const initial = yield* journal.nextRemoteSequence(remoteId)
83 assert.strictEqual(initial, 0)
84
85 const entryA = new EventJournal.Entry({
86 id: EventJournal.makeEntryIdUnsafe(),
87 event: "UserCreated",
88 primaryKey: "user-2",
89 payload: new Uint8Array([2])
90 }, { disableChecks: true })
91 const entryB = new EventJournal.Entry({
92 id: EventJournal.makeEntryIdUnsafe(),
93 event: "UserCreated",
94 primaryKey: "user-3",
95 payload: new Uint8Array([3])
96 }, { disableChecks: true })
97 const remoteEntries = [
98 new EventJournal.RemoteEntry({ remoteSequence: 0, entry: entryA }),
99 new EventJournal.RemoteEntry({ remoteSequence: 1, entry: entryB })
100 ]
101 const seenConflicts: Array<ReadonlyArray<EventJournal.Entry>> = []
102 yield* journal.writeFromRemote({
103 remoteId,
104 entries: remoteEntries,
105 effect: ({ conflicts }) =>
106 Effect.sync(() => {
107 seenConflicts.push(conflicts)
108 })
109 })
110 const next = yield* journal.nextRemoteSequence(remoteId)
111 assert.strictEqual(next, 2)
112
113 let emptyCalls = 0
114 const emptyResult = yield* journal.withRemoteUncommited(remoteId, () =>
115 Effect.sync(() => {
116 emptyCalls++
117 return "called"
118 }))
119 assert.isTrue(Option.isNone(emptyResult))
120 assert.strictEqual(emptyCalls, 0)
121
122 yield* journal.write({
123 event: "LocalCreated",
124 primaryKey: "local-1",
125 payload: new Uint8Array([4]),
126 effect: () => Effect.void
127 })
128 let uncommittedCalls = 0
129 const uncommitted = yield* journal.withRemoteUncommited(remoteId, (entries) =>
130 Effect.sync(() => {
131 uncommittedCalls++
132 return entries
133 }))
134 if (Option.isNone(uncommitted)) assert.fail("Expected the callback result")
135 assert.strictEqual(uncommitted.value.length, 1)
136 assert.strictEqual(uncommitted.value[0].event, "LocalCreated")
137 assert.strictEqual(uncommittedCalls, 1)
138 assert.strictEqual(seenConflicts.length, 2)
139 assert.strictEqual(seenConflicts[0][0]?.idString, entryA.idString)
140 assert.strictEqual(seenConflicts[1][0]?.idString, entryB.idString)
141 }))
142})