1import { assert, it } from "@effect/vitest"
2import { Effect, Fiber, Layer, Option, Queue, Stream } from "effect"
3import * as EventJournal from "effect/eventlog/EventJournal"
4import type { StoreId } from "effect/eventlog/EventLogMessage"
5import * as SqlEventLogServerUnencrypted from "effect/eventlog/SqlEventLogServerUnencrypted"
6import { Reactivity } from "effect/reactivity"
7import type * as SqlClient from "effect/sql/SqlClient"
8
9let nextNamespace = 0
10
11const uniqueNamespace = (prefix: string) => `${prefix}_${++nextNamespace}`
12
13const makeOptions = (prefix: string) => {
14 const namespace = uniqueNamespace(prefix)
15 return {
16 entryTablePrefix: `effect_events_${namespace}`,
17 remoteIdTable: `effect_remote_id_${namespace}`,
18 insertBatchSize: 2
19 }
20}
21
22const makeStoreId = (prefix: string) => `${uniqueNamespace(prefix)}_store` as StoreId
23
24const makeEntry = (
25 name: string,
26 options: {
27 readonly id?: EventJournal.EntryId | undefined
28 readonly primaryKey?: string | undefined
29 } = {}
30) =>
31 new EventJournal.Entry({
32 id: options.id ?? EventJournal.makeEntryIdUnsafe(),
33 event: "UserNameSet",
34 primaryKey: options.primaryKey ?? "user-1",
35 payload: new TextEncoder().encode(name)
36 }, { disableChecks: true })
37
38const makeStorage = (options: {
39 readonly entryTablePrefix?: string
40 readonly remoteIdTable?: string
41 readonly insertBatchSize?: number
42}) =>
43 SqlEventLogServerUnencrypted.makeStorage(options).pipe(
44 Effect.orDie
45 )
46
47export const suite = (name: string, layer: Layer.Layer<SqlClient.SqlClient, unknown>) =>
48 it.layer(
49 Layer.mergeAll(Reactivity.layer, layer),
50 { timeout: "30 seconds" }
51 )(`SqlEventLogServerUnencrypted (${name})`, (it) => {
52 it.effect("persists remote id across storage instances", () =>
53 Effect.gen(function*() {
54 const options = makeOptions("remote_id")
55 const storageA = yield* makeStorage(options)
56 const storageB = yield* makeStorage(options)
57
58 const idA = yield* storageA.getId
59 const idB = yield* storageB.getId
60
61 assert.deepStrictEqual(idA, idB)
62 }))
63
64 it.effect("replays backlog and then streams live changes without startup duplication", () =>
65 Effect.gen(function*() {
66 const storage = yield* makeStorage(makeOptions("changes_backlog_then_live"))
67 const storeId = makeStoreId("changes")
68 const entryA = makeEntry("Ada")
69 const entryB = makeEntry("Grace")
70 const entryC = makeEntry("Margaret")
71
72 yield* storage.write(storeId, [entryA, entryB])
73
74 const changes = yield* storage.changes({
75 storeId,
76 startSequence: 0,
77 compactors: new Map()
78 }).pipe(
79 Stream.toQueue({ capacity: "unbounded" })
80 )
81 const replayed = yield* Queue.takeAll(changes)
82
83 assert.deepStrictEqual(replayed.map((entry) => entry.remoteSequence), [1, 2])
84 assert.deepStrictEqual(replayed.map((entry) => entry.entry.idString), [entryA.idString, entryB.idString])
85
86 yield* storage.write(storeId, [entryC])
87
88 const next = yield* Queue.take(changes)
89 assert.strictEqual(next.remoteSequence, 3)
90 assert.strictEqual(next.entry.idString, entryC.idString)
91
92 yield* Effect.yieldNow
93 assert.strictEqual(Option.isNone(yield* Queue.poll(changes)), true)
94 }))
95
96 it.effect("handles the changes startup race without losing or duplicating rows", () =>
97 Effect.gen(function*() {
98 const storage = yield* makeStorage(makeOptions("changes_startup_race"))
99
100 for (let iteration = 0; iteration < 5; iteration++) {
101 const storeId = makeStoreId(`startup_race_${iteration}`)
102 const backlogEntry = makeEntry(`Ada_${iteration}`)
103 const racedEntry = makeEntry(`Grace_${iteration}`)
104
105 yield* storage.write(storeId, [backlogEntry])
106
107 const changesFiber = yield* storage.changes({
108 storeId,
109 startSequence: 0,
110 compactors: new Map()
111 }).pipe(
112 Stream.toQueue({ capacity: "unbounded" }),
113 Effect.forkChild
114 )
115 yield* storage.write(storeId, [racedEntry])
116 const changes = yield* Fiber.join(changesFiber)
117
118 const first = yield* Queue.take(changes)
119 const second = yield* Queue.take(changes)
120
121 assert.deepStrictEqual(
122 [first.remoteSequence, second.remoteSequence],
123 [1, 2],
124 `iteration ${iteration} should deliver exactly the backlog row and the raced row`
125 )
126 assert.deepStrictEqual(
127 [first.entry.idString, second.entry.idString],
128 [backlogEntry.idString, racedEntry.idString]
129 )
130
131 yield* Effect.yieldNow
132 assert.strictEqual(Option.isNone(yield* Queue.poll(changes)), true)
133 }
134 }))
135
136 it.effect("isolates reads and streams between stores", () =>
137 Effect.gen(function*() {
138 const storage = yield* makeStorage(makeOptions("store_isolation"))
139 const storeA = makeStoreId("isolation_a")
140 const storeB = makeStoreId("isolation_b")
141 const entryA1 = makeEntry("Ada")
142 const entryB1 = makeEntry("Grace")
143 const entryA2 = makeEntry("Margaret")
144 const entryB2 = makeEntry("Linus")
145
146 yield* storage.write(storeA, [entryA1])
147 yield* storage.write(storeB, [entryB1])
148
149 const changesA = yield* storage.changes({
150 storeId: storeA,
151 startSequence: 0,
152 compactors: new Map()
153 }).pipe(
154 Stream.toQueue({ capacity: "unbounded" })
155 )
156 const backlogA = yield* Queue.takeAll(changesA)
157 assert.deepStrictEqual(backlogA.map((entry) => entry.entry.idString), [entryA1.idString])
158
159 yield* storage.write(storeB, [entryB2])
160 yield* Effect.yieldNow
161 assert.strictEqual(Option.isNone(yield* Queue.poll(changesA)), true)
162
163 yield* storage.write(storeA, [entryA2])
164 const nextA = yield* Queue.take(changesA)
165
166 assert.strictEqual(nextA.remoteSequence, 2)
167 assert.strictEqual(nextA.entry.idString, entryA2.idString)
168 }))
169 })