1import { assert, describe, it } from "@effect/vitest"
2import { Effect, Option } from "effect"
3import * as EventJournal from "effect/eventlog/EventJournal"
4
5const entry = (msecs: number) =>
6 new EventJournal.Entry({
7 id: EventJournal.makeEntryIdUnsafe({ msecs }),
8 event: "Repro",
9 primaryKey: "key",
10 payload: new Uint8Array()
11 }, { disableChecks: true })
12
13describe("EventJournal", () => {
14 it.effect("relays an imported entry to another remote", () =>
15 Effect.gen(function*() {
16 const journal = yield* EventJournal.makeMemory
17 const source = EventJournal.makeRemoteIdUnsafe()
18 const target = EventJournal.makeRemoteIdUnsafe()
19 yield* journal.nextRemoteSequence(target)
20 const entry = new EventJournal.Entry({
21 id: EventJournal.makeEntryIdUnsafe(),
22 event: "Repro",
23 primaryKey: "key",
24 payload: new Uint8Array()
25 }, { disableChecks: true })
26 yield* journal.writeFromRemote({
27 remoteId: source,
28 entries: [new EventJournal.RemoteEntry({ remoteSequence: 0, entry })],
29 effect: () => Effect.void
30 })
31 let targetCalls = 0
32 const missing = yield* journal.withRemoteUncommited(target, (entries) =>
33 Effect.sync(() => {
34 targetCalls++
35 return entries
36 }))
37 if (Option.isNone(missing)) assert.fail("Expected the callback result")
38 assert.deepStrictEqual(missing.value.map((item) => item.idString), [entry.idString])
39 assert.strictEqual(targetCalls, 1)
40
41 let sourceCalls = 0
42 const sourceMissing = yield* journal.withRemoteUncommited(source, () =>
43 Effect.sync(() => {
44 sourceCalls++
45 return "called"
46 }))
47 assert.isTrue(Option.isNone(sourceMissing))
48 assert.strictEqual(sourceCalls, 0)
49 }))
50
51 it.effect("returns the next unused remote sequence", () =>
52 Effect.gen(function*() {
53 const journal = yield* EventJournal.makeMemory
54 const remoteId = EventJournal.makeRemoteIdUnsafe()
55 const entry = new EventJournal.Entry({
56 id: EventJournal.makeEntryIdUnsafe(),
57 event: "Repro",
58 primaryKey: "key",
59 payload: new Uint8Array()
60 }, { disableChecks: true })
61
62 assert.strictEqual(yield* journal.nextRemoteSequence(remoteId), 0)
63 yield* journal.writeFromRemote({
64 remoteId,
65 entries: [new EventJournal.RemoteEntry({ remoteSequence: 0, entry })],
66 effect: () => Effect.void
67 })
68 assert.strictEqual(yield* journal.nextRemoteSequence(remoteId), 1)
69 yield* journal.writeFromRemote({
70 remoteId,
71 entries: [new EventJournal.RemoteEntry({ remoteSequence: 5, entry })],
72 effect: () => Effect.void
73 })
74 assert.strictEqual(yield* journal.nextRemoteSequence(remoteId), 6)
75 }))
76
77 it.effect("reports the first newer conflicting entry", () =>
78 Effect.gen(function*() {
79 const journal = yield* EventJournal.makeMemory
80 const newer = entry(2_000)
81 yield* journal.writeFromRemote({
82 remoteId: EventJournal.makeRemoteIdUnsafe(),
83 entries: [new EventJournal.RemoteEntry({ remoteSequence: 0, entry: newer })],
84 effect: () => Effect.void
85 })
86 let conflicts: ReadonlyArray<EventJournal.Entry> = []
87 yield* journal.writeFromRemote({
88 remoteId: EventJournal.makeRemoteIdUnsafe(),
89 entries: [new EventJournal.RemoteEntry({ remoteSequence: 0, entry: entry(1_000) })],
90 effect: (options) => Effect.sync(() => conflicts = options.conflicts)
91 })
92 assert.deepStrictEqual(conflicts.map((item) => item.idString), [newer.idString])
93 }))
94
95 it.effect("records entries in memory and publishes local changes", () =>
96 Effect.gen(function*() {
97 const journal = yield* EventJournal.EventJournal
98 let created = 0
99 yield* journal.write({
100 event: "test",
101 primaryKey: "pk-1",
102 payload: new Uint8Array([1, 2, 3]),
103 effect: (entry) =>
104 Effect.sync(() => {
105 created = entry.createdAtMillis
106 })
107 })
108 const entries = yield* journal.entries
109 assert.strictEqual(entries.length, 1)
110 assert.strictEqual(entries[0].event, "test")
111 assert.strictEqual(entries[0].primaryKey, "pk-1")
112 assert.strictEqual(entries[0].createdAtMillis, created)
113 }).pipe(Effect.provide(EventJournal.layerMemory)))
114})