1import { SqliteClient } from "@effect/sql-sqlite-node"
2import { assert, describe, it } from "@effect/vitest"
3import { Effect, Layer, Queue, Redacted, Stream } from "effect"
4import * as EventJournal from "effect/eventlog/EventJournal"
5import * as EventLog from "effect/eventlog/EventLog"
6import * as EventLogEncryption from "effect/eventlog/EventLogEncryption"
7import * as EventLogMessage from "effect/eventlog/EventLogMessage"
8import type { StoreId } from "effect/eventlog/EventLogMessage"
9import * as EventLogServer from "effect/eventlog/EventLogServerEncrypted"
10import * as EventLogSessionAuth from "effect/eventlog/EventLogSessionAuth"
11import { makeGetIdentityRootSecretMaterial } from "effect/eventlog/internal/identityRootSecretDerivation"
12import * as SqlEventLogServer from "effect/eventlog/SqlEventLogServerEncrypted"
13import { Reactivity } from "effect/reactivity"
14import * as RpcTest from "effect/rpc/RpcTest"
15import * as SqlClient from "effect/sql/SqlClient"
16
17const storeIdA = "store-a" as StoreId
18const storeIdB = "store-b" as StoreId
19
20const makeEntry = (value: number) =>
21 new EventJournal.Entry({
22 id: EventJournal.makeEntryIdUnsafe(),
23 event: "UserCreated",
24 primaryKey: `user-${value}`,
25 payload: new Uint8Array([value])
26 }, { disableChecks: true })
27
28const persistEntries = (
29 encryption: EventLogEncryption.EventLogEncryption["Service"],
30 identity: EventLog.Identity["Service"],
31 entries: ReadonlyArray<EventJournal.Entry>
32) =>
33 Effect.gen(function*() {
34 const encrypted = yield* encryption.encrypt(identity, entries)
35 return encrypted.map(({ encryptedEntry, iv }, index) =>
36 new EventLogServer.PersistedEntry({
37 entryId: entries[index].id,
38 iv,
39 encryptedEntry
40 })
41 )
42 })
43
44const encodeWrite = Effect.fnUntraced(function*(
45 encryption: EventLogEncryption.EventLogEncryption["Service"],
46 identity: EventLog.Identity["Service"],
47 entry: EventJournal.Entry
48) {
49 const encrypted = yield* encryption.encrypt(identity, [entry])
50 return yield* new EventLogMessage.WriteEntries({
51 publicKey: identity.publicKey,
52 storeId: storeIdA,
53 encryptedEntries: [{
54 entryId: entry.id,
55 iv: encrypted[0].iv,
56 encryptedEntry: encrypted[0].encryptedEntry
57 }]
58 }).encoded
59})
60
61const makePersistedEntry = (index: number, entryId = EventJournal.makeEntryIdUnsafe()) =>
62 new EventLogServer.PersistedEntry({
63 entryId,
64 iv: new Uint8Array(12),
65 encryptedEntry: Uint8Array.of(index)
66 })
67
68const getIdentityRootSecretMaterial = makeGetIdentityRootSecretMaterial(globalThis.crypto)
69
70const makeAuthenticateRequest = Effect.fnUntraced(function*(options: {
71 readonly identity: EventLog.Identity["Service"]
72 readonly challenge: Uint8Array
73 readonly remoteId: EventJournal.RemoteId
74}) {
75 const rootSecretMaterial = yield* getIdentityRootSecretMaterial(options.identity)
76 const signature = yield* EventLogSessionAuth.signSessionAuthPayload({
77 remoteId: options.remoteId,
78 challenge: options.challenge,
79 publicKey: options.identity.publicKey,
80 signingPublicKey: rootSecretMaterial.signingPublicKey,
81 signingPrivateKey: Redacted.value(rootSecretMaterial.signingPrivateKey)
82 })
83 return new EventLogMessage.Authenticate({
84 publicKey: options.identity.publicKey,
85 signingPublicKey: rootSecretMaterial.signingPublicKey,
86 signature,
87 algorithm: "Ed25519"
88 })
89})
90
91const makeAuthenticatedRpcClient = Effect.fnUntraced(function*(
92 storage: EventLogServer.Storage["Service"],
93 identities: ReadonlyArray<EventLog.Identity["Service"]>
94) {
95 const rpcClient = yield* RpcTest.makeClient(EventLogMessage.EventLogRemoteRpcs).pipe(
96 Effect.provide(
97 EventLogServer.layerRpcHandlers.pipe(
98 Layer.provide(Layer.succeed(EventLogServer.Storage, storage))
99 )
100 )
101 )
102 for (const identity of identities) {
103 const hello = yield* rpcClient["EventLog.Hello"]()
104 yield* rpcClient["EventLog.Authenticate"](
105 yield* makeAuthenticateRequest({
106 identity,
107 challenge: hello.challenge,
108 remoteId: hello.remoteId
109 })
110 )
111 }
112 return rpcClient
113})
114
115const assertForbidden = Effect.fnUntraced(function*<A>(
116 effect: Effect.Effect<A, EventLogMessage.EventLogProtocolError>
117) {
118 const error = yield* Effect.flip(effect)
119 assert.instanceOf(error, EventLogMessage.EventLogProtocolError)
120 assert.strictEqual(error.code, "Forbidden")
121})
122
123describe("SqlEventLogServer", () => {
124 it.effect("forbids reading another identity's changes", () =>
125 Effect.gen(function*() {
126 const sql = yield* SqliteClient.make({ filename: ":memory:" })
127 const storage = yield* SqlEventLogServer.makeStorage().pipe(
128 Effect.provideService(SqlClient.SqlClient, sql)
129 )
130 const encryption = yield* EventLogEncryption.EventLogEncryption
131 const identityA = yield* encryption.generateIdentity
132 const identityB = yield* encryption.generateIdentity
133 const rpcClient = yield* makeAuthenticatedRpcClient(storage, [identityA])
134
135 yield* storage.write(identityB.publicKey, storeIdA, [makePersistedEntry(1)])
136 const error = yield* rpcClient["EventLog.Changes"]({
137 publicKey: identityB.publicKey,
138 storeId: storeIdA,
139 startSequence: 0
140 }).pipe(
141 Stream.take(1),
142 Stream.runCollect,
143 Effect.flip
144 )
145
146 assert.instanceOf(error, EventLogMessage.EventLogProtocolError)
147 assert.strictEqual(error.code, "Forbidden")
148 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
149
150 it.effect("forbids writing entries for another identity", () =>
151 Effect.gen(function*() {
152 const sql = yield* SqliteClient.make({ filename: ":memory:" })
153 const storage = yield* SqlEventLogServer.makeStorage().pipe(
154 Effect.provideService(SqlClient.SqlClient, sql)
155 )
156 const encryption = yield* EventLogEncryption.EventLogEncryption
157 const identityA = yield* encryption.generateIdentity
158 const identityB = yield* encryption.generateIdentity
159 const rpcClient = yield* makeAuthenticatedRpcClient(storage, [identityA])
160 const entry = makeEntry(1)
161 const data = yield* encodeWrite(encryption, identityB, entry)
162
163 const error = yield* rpcClient["EventLog.WriteSingle"]({ data }).pipe(Effect.flip)
164
165 assert.instanceOf(error, EventLogMessage.EventLogProtocolError)
166 assert.strictEqual(error.code, "Forbidden")
167 const written = yield* storage.write(identityB.publicKey, storeIdA, [makePersistedEntry(2)])
168 assert.deepStrictEqual(written.map((entry) => entry.sequence), [1])
169 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
170
171 it.effect("forbids writing chunked entries for another identity", () =>
172 Effect.gen(function*() {
173 const sql = yield* SqliteClient.make({ filename: ":memory:" })
174 const storage = yield* SqlEventLogServer.makeStorage().pipe(
175 Effect.provideService(SqlClient.SqlClient, sql)
176 )
177 const encryption = yield* EventLogEncryption.EventLogEncryption
178 const identityA = yield* encryption.generateIdentity
179 const identityB = yield* encryption.generateIdentity
180 const rpcClient = yield* makeAuthenticatedRpcClient(storage, [identityA])
181 const entry = makeEntry(1)
182 const data = yield* encodeWrite(encryption, identityB, entry)
183 const midpoint = Math.ceil(data.byteLength / 2)
184 const parts = [
185 new EventLogMessage.ChunkedMessage({ id: 1, part: [0, 2], data: data.subarray(0, midpoint) }),
186 new EventLogMessage.ChunkedMessage({ id: 1, part: [1, 2], data: data.subarray(midpoint) })
187 ] as const
188
189 yield* rpcClient["EventLog.WriteChunked"](parts[0])
190 const error = yield* rpcClient["EventLog.WriteChunked"](parts[1]).pipe(Effect.flip)
191
192 assert.instanceOf(error, EventLogMessage.EventLogProtocolError)
193 assert.strictEqual(error.code, "Forbidden")
194 const written = yield* storage.write(identityB.publicKey, storeIdA, [makePersistedEntry(2)])
195 assert.deepStrictEqual(written.map((entry) => entry.sequence), [1])
196 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
197
198 it.effect("isolates authenticated identities between connections", () =>
199 Effect.gen(function*() {
200 const sql = yield* SqliteClient.make({ filename: ":memory:" })
201 const storage = yield* SqlEventLogServer.makeStorage().pipe(
202 Effect.provideService(SqlClient.SqlClient, sql)
203 )
204 const encryption = yield* EventLogEncryption.EventLogEncryption
205 const identityA = yield* encryption.generateIdentity
206 const identityB = yield* encryption.generateIdentity
207 const handlers = yield* Layer.build(
208 EventLogServer.layerRpcHandlers.pipe(
209 Layer.provide(Layer.succeed(EventLogServer.Storage, storage))
210 )
211 )
212 const rpcClientA = yield* RpcTest.makeClient(EventLogMessage.EventLogRemoteRpcs).pipe(
213 Effect.provide(handlers)
214 )
215 const rpcClientB = yield* RpcTest.makeClient(EventLogMessage.EventLogRemoteRpcs).pipe(
216 Effect.provide(handlers)
217 )
218 const helloA = yield* rpcClientA["EventLog.Hello"]()
219 yield* rpcClientA["EventLog.Authenticate"](
220 yield* makeAuthenticateRequest({
221 identity: identityA,
222 challenge: helloA.challenge,
223 remoteId: helloA.remoteId
224 })
225 )
226 const helloB = yield* rpcClientB["EventLog.Hello"]()
227 yield* rpcClientB["EventLog.Authenticate"](
228 yield* makeAuthenticateRequest({
229 identity: identityB,
230 challenge: helloB.challenge,
231 remoteId: helloB.remoteId
232 })
233 )
234 const data = yield* encodeWrite(encryption, identityB, makeEntry(1))
235 const midpoint = Math.ceil(data.byteLength / 2)
236 const parts = [
237 new EventLogMessage.ChunkedMessage({ id: 1, part: [0, 2], data: data.subarray(0, midpoint) }),
238 new EventLogMessage.ChunkedMessage({ id: 1, part: [1, 2], data: data.subarray(midpoint) })
239 ] as const
240
241 yield* storage.write(identityB.publicKey, storeIdA, [makePersistedEntry(1)])
242 yield* assertForbidden(
243 rpcClientA["EventLog.Changes"]({
244 publicKey: identityB.publicKey,
245 storeId: storeIdA,
246 startSequence: 0
247 }).pipe(Stream.take(1), Stream.runCollect)
248 )
249 yield* assertForbidden(rpcClientA["EventLog.WriteSingle"]({ data }))
250 yield* rpcClientA["EventLog.WriteChunked"](parts[0])
251 yield* assertForbidden(rpcClientA["EventLog.WriteChunked"](parts[1]))
252 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
253
254 it.effect("supports multiple authenticated identities on one connection", () =>
255 Effect.gen(function*() {
256 const sql = yield* SqliteClient.make({ filename: ":memory:" })
257 const storage = yield* SqlEventLogServer.makeStorage().pipe(
258 Effect.provideService(SqlClient.SqlClient, sql)
259 )
260 const encryption = yield* EventLogEncryption.EventLogEncryption
261 const identityA = yield* encryption.generateIdentity
262 const identityB = yield* encryption.generateIdentity
263 const rpcClient = yield* makeAuthenticatedRpcClient(storage, [identityA, identityB])
264 const entryA = makeEntry(1)
265 const entryB = makeEntry(2)
266 const dataA = yield* encodeWrite(encryption, identityA, entryA)
267 const dataB = yield* encodeWrite(encryption, identityB, entryB)
268
269 yield* rpcClient["EventLog.WriteSingle"]({ data: dataA })
270 yield* rpcClient["EventLog.WriteSingle"]({ data: dataB })
271 const changesA = yield* rpcClient["EventLog.Changes"]({
272 publicKey: identityA.publicKey,
273 storeId: storeIdA,
274 startSequence: 0
275 }).pipe(Stream.take(1), Stream.runCollect)
276 const changesB = yield* rpcClient["EventLog.Changes"]({
277 publicKey: identityB.publicKey,
278 storeId: storeIdA,
279 startSequence: 0
280 }).pipe(Stream.take(1), Stream.runCollect)
281
282 assert.strictEqual(changesA.length, 1)
283 assert.strictEqual(changesB.length, 1)
284 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
285
286 it.effect("persists remote id across storage instances", () =>
287 Effect.gen(function*() {
288 const sql = yield* SqliteClient.make({ filename: ":memory:" })
289 const storageA = yield* SqlEventLogServer.makeStorage().pipe(
290 Effect.provideService(SqlClient.SqlClient, sql)
291 )
292 const storageB = yield* SqlEventLogServer.makeStorage().pipe(
293 Effect.provideService(SqlClient.SqlClient, sql)
294 )
295 const idA = yield* storageA.getId
296 const idB = yield* storageB.getId
297 assert.deepStrictEqual(idA, idB)
298 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
299
300 it.effect("rejects session-auth rebinding for an existing publicKey", () =>
301 Effect.gen(function*() {
302 const sql = yield* SqliteClient.make({ filename: ":memory:" })
303 const storage = yield* SqlEventLogServer.makeStorage().pipe(
304 Effect.provideService(SqlClient.SqlClient, sql)
305 )
306 const rpcClient = yield* RpcTest.makeClient(EventLogMessage.EventLogRemoteRpcs).pipe(
307 Effect.provide(
308 EventLogServer.layerRpcHandlers.pipe(
309 Layer.provide(Layer.succeed(EventLogServer.Storage, storage))
310 )
311 )
312 )
313
314 const firstIdentity = yield* EventLog.makeIdentity
315 const secondIdentitySeed = yield* EventLog.makeIdentity
316 const secondIdentity: EventLog.Identity["Service"] = {
317 publicKey: firstIdentity.publicKey,
318 privateKey: secondIdentitySeed.privateKey
319 }
320
321 const firstMaterial = yield* getIdentityRootSecretMaterial(firstIdentity)
322 const secondMaterial = yield* getIdentityRootSecretMaterial(secondIdentity)
323 const sameSigningPublicKey =
324 firstMaterial.signingPublicKey.byteLength === secondMaterial.signingPublicKey.byteLength &&
325 firstMaterial.signingPublicKey.every((byte, index) => byte === secondMaterial.signingPublicKey[index])
326 assert.strictEqual(sameSigningPublicKey, false)
327
328 const firstHello = yield* rpcClient["EventLog.Hello"]()
329 yield* rpcClient["EventLog.Authenticate"](
330 yield* makeAuthenticateRequest({
331 identity: firstIdentity,
332 challenge: firstHello.challenge,
333 remoteId: firstHello.remoteId
334 })
335 )
336
337 const secondHello = yield* rpcClient["EventLog.Hello"]()
338 const error = yield* rpcClient["EventLog.Authenticate"](
339 yield* makeAuthenticateRequest({
340 identity: secondIdentity,
341 challenge: secondHello.challenge,
342 remoteId: secondHello.remoteId
343 })
344 ).pipe(Effect.flip)
345
346 assert.instanceOf(error, EventLogMessage.EventLogProtocolError)
347 assert.strictEqual(error.code, "Forbidden")
348 assert.strictEqual(error.message, "Session auth signature verification failed")
349 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
350
351 it.effect("writes entries and streams changes", () =>
352 Effect.gen(function*() {
353 const sql = yield* SqliteClient.make({ filename: ":memory:" })
354 const storage = yield* SqlEventLogServer.makeStorage().pipe(
355 Effect.provideService(SqlClient.SqlClient, sql)
356 )
357 const encryption = yield* EventLogEncryption.EventLogEncryption
358 const identity = yield* encryption.generateIdentity
359 const entries = [makeEntry(1), makeEntry(2)]
360 const persisted = yield* persistEntries(encryption, identity, entries)
361 const written = yield* storage.write(identity.publicKey, storeIdA, persisted)
362 assert.deepStrictEqual(written.map((entry) => entry.sequence), [1, 2])
363
364 const changes = yield* storage.changes(identity.publicKey, storeIdA, 0).pipe(
365 Stream.toQueue({ capacity: "unbounded" })
366 )
367 const taken = yield* Queue.takeAll(changes)
368 assert.deepStrictEqual(taken.map((entry) => entry.sequence), [1, 2])
369
370 const nextEntry = makeEntry(3)
371 const nextPersisted = yield* persistEntries(encryption, identity, [nextEntry])
372 const updated = yield* storage.write(identity.publicKey, storeIdA, nextPersisted)
373 assert.deepStrictEqual(updated.map((entry) => entry.sequence), [3])
374
375 const next = yield* Queue.take(changes)
376 assert.strictEqual(next.sequence, 3)
377 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
378
379 it.effect("isolates same publicKey across storeIds", () =>
380 Effect.gen(function*() {
381 const sql = yield* SqliteClient.make({ filename: ":memory:" })
382 const storage = yield* SqlEventLogServer.makeStorage().pipe(
383 Effect.provideService(SqlClient.SqlClient, sql)
384 )
385
386 yield* storage.write("client-1", storeIdA, [makePersistedEntry(1)])
387 yield* storage.write("client-1", storeIdB, [makePersistedEntry(2)])
388
389 const storeAEntries = yield* storage.changes("client-1", storeIdA, 0).pipe(
390 Stream.take(1),
391 Stream.runCollect
392 )
393 const storeBEntries = yield* storage.changes("client-1", storeIdB, 0).pipe(
394 Stream.take(1),
395 Stream.runCollect
396 )
397
398 assert.deepStrictEqual(storeAEntries.map((entry) => entry.sequence), [1])
399 assert.deepStrictEqual(storeBEntries.map((entry) => entry.sequence), [1])
400 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
401
402 it.effect("isolates same storeId across publicKeys", () =>
403 Effect.gen(function*() {
404 const sql = yield* SqliteClient.make({ filename: ":memory:" })
405 const storage = yield* SqlEventLogServer.makeStorage().pipe(
406 Effect.provideService(SqlClient.SqlClient, sql)
407 )
408
409 yield* storage.write("client-1", storeIdA, [makePersistedEntry(1)])
410 yield* storage.write("client-2", storeIdA, [makePersistedEntry(2)])
411
412 const clientOneEntries = yield* storage.changes("client-1", storeIdA, 0).pipe(
413 Stream.take(1),
414 Stream.runCollect
415 )
416 const clientTwoEntries = yield* storage.changes("client-2", storeIdA, 0).pipe(
417 Stream.take(1),
418 Stream.runCollect
419 )
420
421 assert.deepStrictEqual(clientOneEntries.map((entry) => entry.sequence), [1])
422 assert.deepStrictEqual(clientTwoEntries.map((entry) => entry.sequence), [1])
423 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
424
425 it.effect("keeps deduplication isolated per encrypted scope", () =>
426 Effect.gen(function*() {
427 const sql = yield* SqliteClient.make({ filename: ":memory:" })
428 const storage = yield* SqlEventLogServer.makeStorage().pipe(
429 Effect.provideService(SqlClient.SqlClient, sql)
430 )
431 const sharedEntryId = EventJournal.makeEntryIdUnsafe()
432
433 yield* storage.write("client-1", storeIdA, [makePersistedEntry(1, sharedEntryId)])
434 yield* storage.write("client-1", storeIdA, [makePersistedEntry(2, sharedEntryId)])
435 yield* storage.write("client-1", storeIdB, [makePersistedEntry(3, sharedEntryId)])
436 yield* storage.write("client-2", storeIdA, [makePersistedEntry(4, sharedEntryId)])
437
438 assert.deepStrictEqual(
439 (yield* storage.changes("client-1", storeIdA, 0).pipe(
440 Stream.take(1),
441 Stream.runCollect
442 )).map((entry) => entry.sequence),
443 [1]
444 )
445 assert.deepStrictEqual(
446 (yield* storage.changes("client-1", storeIdB, 0).pipe(
447 Stream.take(1),
448 Stream.runCollect
449 )).map((entry) => entry.sequence),
450 [1]
451 )
452 assert.deepStrictEqual(
453 (yield* storage.changes("client-2", storeIdA, 0).pipe(
454 Stream.take(1),
455 Stream.runCollect
456 )).map((entry) => entry.sequence),
457 [1]
458 )
459 }).pipe(Effect.provide([Reactivity.layer, EventLogEncryption.layerSubtle])))
460})