1import { assert, describe, it } from "@effect/vitest"
2import { Effect, Layer, PubSub, Queue, Ref, Schema } from "effect"
3import * as EventGroup from "effect/eventlog/EventGroup"
4import * as EventJournal from "effect/eventlog/EventJournal"
5import * as EventLog from "effect/eventlog/EventLog"
6import * as EventLogEncryption from "effect/eventlog/EventLogEncryption"
7import * as EventLogRemote from "effect/eventlog/EventLogRemote"
8import { TestClock } from "effect/testing"
9
10const UserPayload = Schema.Struct({
11 id: Schema.String
12})
13
14const UserGroup = EventGroup.empty.add({
15 tag: "UserCreated",
16 primaryKey: (payload) => payload.id,
17 payload: UserPayload
18})
19
20const schema = EventLog.schema(UserGroup)
21
22const handlerLayer = (handled: Ref.Ref<ReadonlyArray<string>>) =>
23 EventLog.group(
24 UserGroup,
25 (handlers) =>
26 handlers.handle("UserCreated", ({ payload }) => Ref.update(handled, (values) => [...values, payload.id]))
27 ).pipe(
28 Layer.provide(EventLog.layerRegistry)
29 )
30
31const logLayer = (
32 handled: Ref.Ref<ReadonlyArray<string>>,
33 journalLayer: Layer.Layer<EventJournal.EventJournal> = EventJournal.layerMemory
34) =>
35 EventLog.layer(schema, handlerLayer(handled)).pipe(
36 Layer.provide(journalLayer),
37 Layer.provide(
38 Layer.effect(EventLog.Identity, EventLog.makeIdentity).pipe(
39 Layer.provide(EventLogEncryption.layerSubtle)
40 )
41 )
42 )
43
44describe("EventLog", () => {
45 it.effect("writes a typed event, commits the entry, and runs its handler", () =>
46 Effect.gen(function*() {
47 const handled = yield* Ref.make<ReadonlyArray<string>>([])
48 return yield* Effect.gen(function*() {
49 const log = yield* EventLog.EventLog
50 yield* log.write({
51 schema,
52 event: "UserCreated",
53 payload: { id: "user-1" }
54 })
55 const entries = yield* log.entries
56 const seen = yield* Ref.get(handled)
57 assert.strictEqual(entries.length, 1)
58 assert.strictEqual(entries[0].event, "UserCreated")
59 assert.deepStrictEqual(seen, ["user-1"])
60 }).pipe(Effect.provide(logLayer(handled)))
61 }))
62
63 it.effect("retries a failed remote write and flushes changes made during the retry", () =>
64 Effect.gen(function*() {
65 const handled = yield* Ref.make<ReadonlyArray<string>>([])
66 const subscriptions = yield* Queue.unbounded<PubSub.Subscription<EventJournal.Entry>>()
67 const journalLayer = Layer.effect(
68 EventJournal.EventJournal,
69 EventJournal.makeMemory.pipe(
70 Effect.map((journal) =>
71 EventJournal.EventJournal.of({
72 ...journal,
73 changes: journal.changes.pipe(Effect.tap((subscription) => Queue.offer(subscriptions, subscription)))
74 })
75 )
76 )
77 )
78 return yield* Effect.gen(function*() {
79 const log = yield* EventLog.EventLog
80 const registry = yield* EventLog.Registry
81 const attempts = yield* Queue.unbounded<ReadonlyArray<EventJournal.Entry>>()
82 const release = yield* Queue.unbounded<void>()
83 let attempt = 0
84 let failedChangeWrite = false
85 const remote = EventLogRemote.EventLogRemote.of({
86 id: EventJournal.makeRemoteIdUnsafe(),
87 changes: () => Queue.unbounded<EventJournal.RemoteEntry, EventLogRemote.EventLogRemoteError>(),
88 write: ({ entries }) =>
89 Queue.offer(attempts, entries).pipe(
90 Effect.flatMap(() => {
91 attempt++
92 if (attempt === 1) {
93 return Effect.fail(new EventLogRemote.EventLogRemoteError({ method: "write", cause: "offline" }))
94 }
95 if (attempt === 2) return Queue.take(release)
96 if (!failedChangeWrite && entries.some((entry) => entry.primaryKey === "user-4")) {
97 failedChangeWrite = true
98 return Effect.fail(new EventLogRemote.EventLogRemoteError({ method: "write", cause: "offline" }))
99 }
100 return Effect.void
101 })
102 ),
103 whenAuthenticated: (effect) => effect
104 })
105
106 yield* log.write({ schema, event: "UserCreated", payload: { id: "user-1" } })
107 yield* registry.registerRemote(remote)
108 const subscription = yield* Queue.take(subscriptions)
109 const initial = yield* Queue.take(attempts)
110 yield* TestClock.adjust("200 millis")
111 const retry = yield* Queue.take(attempts)
112 yield* log.write({ schema, event: "UserCreated", payload: { id: "user-2" } })
113 yield* log.write({ schema, event: "UserCreated", payload: { id: "user-3" } })
114 yield* Effect.yieldNow
115 assert.strictEqual(yield* PubSub.remaining(subscription), 0)
116 yield* Queue.offer(release, undefined)
117 const afterChange = yield* Queue.take(attempts)
118 yield* Effect.yieldNow
119 assert.strictEqual(yield* Queue.size(attempts), 0)
120 yield* log.write({ schema, event: "UserCreated", payload: { id: "user-4" } })
121 const changeAttempt = yield* Queue.take(attempts)
122 yield* TestClock.adjust("200 millis")
123 const changeRetry = yield* Queue.take(attempts)
124
125 assert.deepStrictEqual(retry.map((entry) => entry.idString), initial.map((entry) => entry.idString))
126 assert.deepStrictEqual(afterChange.map((entry) => entry.primaryKey), ["user-2", "user-3"])
127 assert.deepStrictEqual(changeAttempt.map((entry) => entry.primaryKey), ["user-4"])
128 assert.deepStrictEqual(
129 changeRetry.map((entry) => entry.idString),
130 changeAttempt.map((entry) => entry.idString)
131 )
132 }).pipe(Effect.provide(logLayer(handled, journalLayer)))
133 }))
134
135 it.effect("encrypts and decrypts entries with a distinct IV per entry", () =>
136 Effect.gen(function*() {
137 const encryption = yield* EventLogEncryption.EventLogEncryption
138 const identity = yield* encryption.generateIdentity
139 const entries = ["user-1", "user-2"].map((primaryKey, index) =>
140 new EventJournal.Entry({
141 id: EventJournal.makeEntryIdUnsafe(),
142 event: "UserCreated",
143 primaryKey,
144 payload: new Uint8Array([index])
145 }, { disableChecks: true })
146 )
147 const encrypted = yield* encryption.encrypt(identity, entries)
148 assert.notDeepEqual(encrypted[0].iv, encrypted[1].iv)
149 const decrypted = yield* encryption.decrypt(
150 identity,
151 encrypted.map((entry, index) => ({ ...entry, sequence: index, entryId: entries[index].id }))
152 )
153 assert.deepStrictEqual(
154 decrypted.map((remote) => remote.entry.idString),
155 entries.map((entry) => entry.idString)
156 )
157 }).pipe(Effect.provide(EventLogEncryption.layerSubtle)))
158
159 it.effect("publishes local journal changes through a scoped subscription", () =>
160 Effect.gen(function*() {
161 const remoteId = EventJournal.makeRemoteIdUnsafe()
162 const journal = yield* EventJournal.EventJournal
163 const entry = new EventJournal.Entry({
164 id: EventJournal.makeEntryIdUnsafe(),
165 event: "UserCreated",
166 primaryKey: "user-1",
167 payload: new Uint8Array([1])
168 }, { disableChecks: true })
169
170 const first = yield* journal.writeFromRemote({
171 remoteId,
172 entries: [new EventJournal.RemoteEntry({ remoteSequence: 1, entry })],
173 effect: () => Effect.void
174 })
175 const second = yield* journal.writeFromRemote({
176 remoteId,
177 entries: [new EventJournal.RemoteEntry({ remoteSequence: 2, entry })],
178 effect: () => Effect.void
179 })
180
181 assert.deepStrictEqual(first.duplicateEntries, [])
182 assert.deepStrictEqual(second.duplicateEntries.map((_) => _.idString), [entry.idString])
183 }).pipe(Effect.provide(EventJournal.layerMemory)))
184})