1/**
2 * Stores event-log entries and replication state.
3 *
4 * `EventJournal` records committed entries, exposes them for replay, publishes
5 * local changes, and tracks the remote metadata needed to exchange entries with
6 * other journals. Higher-level event-log schemas and handlers use this service
7 * to rebuild projections, sync offline clients, import remote changes, and
8 * coordinate writes per store. This module also defines journal errors, entry
9 * and remote identifiers, schemas, and in-memory or IndexedDB-backed journal
10 * layers.
11 *
12 * @stability unstable
13 * @since 4.0.0
14 */
15import * as Arr from "../Array.ts"
16import type { Brand } from "../Brand.ts"
17import * as Context from "../Context.ts"
18import * as Data from "../Data.ts"
19import * as DateTime from "../DateTime.ts"
20import * as Effect from "../Effect.ts"
21import * as SchemaBinary from "../encoding/SchemaBinary.ts"
22import * as Uuid from "../internal/uuid.ts"
23import * as Layer from "../Layer.ts"
24import type { Option } from "../Option.ts"
25import * as Order from "../Order.ts"
26import * as PubSub from "../PubSub.ts"
27import * as Schema from "../Schema.ts"
28import type { Scope } from "../Scope.ts"
29import * as Semaphore from "../Semaphore.ts"
30import type { StoreId } from "./EventLogMessage.ts"
31
32/**
33 * Context service for storing and replaying event journal entries.
34 *
35 * **Details**
36 *
37 * The service writes local entries, imports entries from remote journals, exposes
38 * a stream of local changes, and provides per-store locking.
39 *
40 * @stability unstable
41 * @category services
42 * @since 4.0.0
43 */
44export class EventJournal extends Context.Service<EventJournal, {
45 /**
46 * Read all the entries in the journal.
47 */
48 readonly entries: Effect.Effect<ReadonlyArray<Entry>, EventJournalError>
49
50 /**
51 * Write an event to the journal, performing an effect before committing the
52 * event.
53 */
54 readonly write: <A, E, R>(options: {
55 readonly event: string
56 readonly primaryKey: string
57 readonly payload: Uint8Array
58 readonly effect: (entry: Entry) => Effect.Effect<A, E, R>
59 }) => Effect.Effect<A, EventJournalError | E, R>
60
61 /**
62 * Write events from a remote source to the journal.
63 *
64 * **Details**
65 *
66 * Effects run sequentially in compaction bracket order.
67 */
68 readonly writeFromRemote: (
69 options: {
70 readonly remoteId: RemoteId
71 readonly entries: ReadonlyArray<RemoteEntry>
72 readonly compact?:
73 | ((uncommitted: ReadonlyArray<RemoteEntry>) => Effect.Effect<ReadonlyArray<Entry>, EventJournalError>)
74 | undefined
75 readonly effect: (options: {
76 readonly entry: Entry
77 readonly conflicts: ReadonlyArray<Entry>
78 }) => Effect.Effect<void, EventJournalError>
79 }
80 ) => Effect.Effect<{
81 readonly duplicateEntries: ReadonlyArray<Entry>
82 }, EventJournalError>
83
84 /**
85 * Run an effect with the uncommitted entries for a remote source.
86 *
87 * The effect is not run when there are no uncommitted entries, in which case
88 * `Option.none()` is returned. Otherwise, its result is wrapped in
89 * `Option.some()`.
90 */
91 readonly withRemoteUncommited: <A, E, R>(
92 remoteId: RemoteId,
93 f: (entries: Arr.NonEmptyReadonlyArray<Entry>) => Effect.Effect<A, E, R>
94 ) => Effect.Effect<Option<A>, EventJournalError | E, R>
95
96 /**
97 * Retrieve the first unused sequence number for a remote source.
98 */
99 readonly nextRemoteSequence: (remoteId: RemoteId) => Effect.Effect<number, EventJournalError>
100
101 /**
102 * The entries added to the local journal.
103 */
104 readonly changes: Effect.Effect<PubSub.Subscription<Entry>, never, Scope>
105
106 /**
107 * Remove all data
108 */
109 readonly destroy: Effect.Effect<void, EventJournalError>
110
111 /**
112 * Run an effect with a lock on the journal.
113 */
114 readonly withLock: (storeId: StoreId) => <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
115}>()("effect/eventlog/EventJournal") {}
116
117const TypeId = "effect/eventlog/EventJournal/EventJournalError" as const
118
119/**
120 * Error raised by event journal operations.
121 *
122 * **Details**
123 *
124 * The error records the journal method that failed and the underlying cause.
125 *
126 * @stability unstable
127 * @category errors
128 * @since 4.0.0
129 */
130export class EventJournalError extends Data.TaggedError("EventJournalError")<{
131 readonly method: string
132 readonly cause: unknown
133}> {
134 /**
135 * Marks this value as an event journal error for runtime guards.
136 *
137 * @stability unstable
138 * @since 4.0.0
139 */
140 readonly [TypeId] = TypeId
141}
142
143/**
144 * Brand identifier used for `RemoteId` values.
145 *
146 * @stability unstable
147 * @category type IDs
148 * @since 4.0.0
149 */
150export type RemoteIdTypeId = "effect/eventlog/EventJournal/RemoteId"
151
152/**
153 * Runtime brand identifier used for `RemoteId` values.
154 *
155 * @stability unstable
156 * @category type IDs
157 * @since 4.0.0
158 */
159export const RemoteIdTypeId: RemoteIdTypeId = "effect/eventlog/EventJournal/RemoteId"
160
161/**
162 * Branded byte identifier for a remote event journal source.
163 *
164 * @stability unstable
165 * @category models
166 * @since 4.0.0
167 */
168export type RemoteId = Uint8Array & Brand<RemoteIdTypeId>
169
170/**
171 * Schema for branded remote event journal identifiers.
172 *
173 * @stability unstable
174 * @category schemas
175 * @since 4.0.0
176 */
177export const RemoteId = Schema.Uint8Array.pipe(Schema.brand(RemoteIdTypeId))
178
179/**
180 * Generates a new random `RemoteId`.
181 *
182 * **When to use**
183 *
184 * Use when generating a fresh event-log remote id internally and the UUID bytes
185 * are trusted to satisfy the brand.
186 *
187 * **Gotchas**
188 *
189 * This is unsafe because the generated UUID bytes are cast to the brand without
190 * schema validation.
191 *
192 * @stability unstable
193 * @category unsafe
194 * @since 4.0.0
195 */
196export const makeRemoteIdUnsafe = (): RemoteId => Uuid.v4Bytes() as RemoteId
197
198/**
199 * Runtime brand identifier used for `EntryId` values.
200 *
201 * @stability unstable
202 * @category type IDs
203 * @since 4.0.0
204 */
205export const EntryIdTypeId: EntryIdTypeId = "effect/eventlog/EventJournal/EntryId"
206
207/**
208 * Brand identifier used for `EntryId` values.
209 *
210 * @stability unstable
211 * @category type IDs
212 * @since 4.0.0
213 */
214export type EntryIdTypeId = "effect/eventlog/EventJournal/EntryId"
215
216/**
217 * Branded byte identifier for an event journal entry.
218 *
219 * @stability unstable
220 * @category models
221 * @since 4.0.0
222 */
223export type EntryId = Uint8Array<ArrayBuffer> & Brand<EntryIdTypeId>
224
225/**
226 * Schema for branded event journal entry identifiers.
227 *
228 * @stability unstable
229 * @category schemas
230 * @since 4.0.0
231 */
232export const EntryId = (Schema.Uint8Array as Schema.instanceOf<Uint8Array<ArrayBuffer>>).pipe(
233 Schema.brand(EntryIdTypeId)
234)
235
236/**
237 * Provides an Ordering instance for entry identifiers based on their raw UUID bytes.
238 *
239 * @stability unstable
240 * @category ordering
241 * @since 4.0.0
242 */
243export const EntryIdOrder = Order.make<EntryId>((a, b) => {
244 for (let i = 0; i < 16; i++) {
245 if (a[i] !== b[i]) {
246 return (a[i] - b[i]) < 0 ? -1 : 1
247 }
248 }
249 return 0
250})
251
252/**
253 * Generates a UUID v7 `EntryId`, optionally using the supplied millisecond
254 * timestamp.
255 *
256 * **When to use**
257 *
258 * Use when generating an event-log entry id internally and the UUID v7 bytes
259 * are trusted to satisfy the brand.
260 *
261 * **Gotchas**
262 *
263 * This is unsafe because the generated UUID bytes are cast to the brand without
264 * schema validation.
265 *
266 * @stability unstable
267 * @category unsafe
268 * @since 4.0.0
269 */
270export const makeEntryIdUnsafe = (options: { msecs?: number } = {}): EntryId =>
271 Uuid.v7Bytes(options.msecs ?? DateTime.nowUnsafe().epochMilliseconds) as EntryId
272
273/**
274 * Extracts the millisecond timestamp encoded in a UUID v7 `EntryId`.
275 *
276 * @stability unstable
277 * @category getters
278 * @since 4.0.0
279 */
280export const entryIdMillis = (entryId: EntryId): number => {
281 const bytes = new Uint8Array(8)
282 bytes.set(entryId.subarray(0, 6), 2)
283 return Number(new DataView(bytes.buffer).getBigUint64(0))
284}
285
286/**
287 * Schema for a committed event journal entry.
288 *
289 * **Details**
290 *
291 * An entry records its ID, event tag, primary key, and SchemaBinary-encoded
292 * payload, with helpers for array SchemaBinary encoding and creation timestamps.
293 *
294 * @stability unstable
295 * @category schemas
296 * @since 4.0.0
297 */
298export class Entry extends Schema.Class<Entry>("effect/eventlog/EventJournal/Entry")({
299 id: EntryId,
300 event: Schema.String,
301 primaryKey: Schema.String,
302 payload: Schema.Uint8Array
303}) {
304 /**
305 * SchemaBinary codec for arrays of committed event journal entries.
306 *
307 * @stability unstable
308 * @since 4.0.0
309 */
310 static arraySchemaBinary = Schema.Array(SchemaBinary.toCodec(Entry))
311
312 /**
313 * Encodes arrays of committed entries with the SchemaBinary entry codec.
314 *
315 * @stability unstable
316 * @since 4.0.0
317 */
318 static encodeArray = Schema.encodeUnknownEffect(Entry.arraySchemaBinary)
319
320 /**
321 * Decodes arrays of committed entries with the SchemaBinary entry codec.
322 *
323 * @stability unstable
324 * @since 4.0.0
325 */
326 static decodeArray = Schema.decodeUnknownEffect(Entry.arraySchemaBinary)
327
328 /**
329 * Ordering for committed entries by their event journal entry id.
330 *
331 * @stability unstable
332 * @since 4.0.0
333 */
334 static Order = Order.make<Entry>((a, b) => EntryIdOrder(a.id, b.id))
335
336 /**
337 * String representation of the entry id.
338 *
339 * @stability unstable
340 * @since 4.0.0
341 */
342 get idString(): string {
343 return Uuid.stringify(this.id)
344 }
345
346 /**
347 * Creation timestamp encoded in the UUID v7 entry id, in epoch milliseconds.
348 *
349 * @stability unstable
350 * @since 4.0.0
351 */
352 get createdAtMillis(): number {
353 return entryIdMillis(this.id)
354 }
355
356 /**
357 * Creation timestamp encoded in the UUID v7 entry id, as a UTC date-time.
358 *
359 * @stability unstable
360 * @since 4.0.0
361 */
362 get createdAt(): DateTime.Utc {
363 return DateTime.makeUnsafe(this.createdAtMillis)
364 }
365}
366
367/**
368 * Schema for an event journal entry received from a remote source.
369 *
370 * **Details**
371 *
372 * It pairs the remote sequence number with the journal entry payload.
373 *
374 * @stability unstable
375 * @category schemas
376 * @since 4.0.0
377 */
378export class RemoteEntry extends Schema.Class<RemoteEntry>("effect/eventlog/EventJournal/RemoteEntry")({
379 remoteSequence: Schema.Natural,
380 entry: Entry
381}) {}
382
383/**
384 * Creates an in-memory `EventJournal` service.
385 *
386 * **Gotchas**
387 *
388 * Entries, remote tracking state, and locks live only in the current process and
389 * are lost when the service is discarded.
390 *
391 * @stability unstable
392 * @category constructors
393 * @since 4.0.0
394 */
395export const makeMemory: Effect.Effect<EventJournal["Service"]> = Effect.gen(function*() {
396 const journal: Array<Entry> = []
397 const byId = new Map<string, Entry>()
398 const remotes = new Map<string, { sequence: number; missing: Array<Entry> }>()
399 const pubsub = yield* PubSub.unbounded<Entry>()
400 const storeSemaphores = new Map<StoreId, Semaphore.Semaphore>()
401 const withLock = (storeId: StoreId) => {
402 let semaphore = storeSemaphores.get(storeId)
403 if (!semaphore) {
404 semaphore = Semaphore.makeUnsafe(1)
405 storeSemaphores.set(storeId, semaphore)
406 }
407 return semaphore.withPermit
408 }
409
410 const ensureRemote = (remoteId: RemoteId) => {
411 const remoteIdString = Uuid.stringify(remoteId)
412 const remote = remotes.get(remoteIdString)
413 if (remote) return remote
414 const created = { sequence: 0, missing: journal.slice() }
415 remotes.set(remoteIdString, created)
416 return created
417 }
418
419 return EventJournal.of({
420 entries: Effect.sync(() => journal.slice()),
421 write({ effect, event, payload, primaryKey }) {
422 return Effect.acquireUseRelease(
423 Effect.sync(() =>
424 new Entry({
425 id: makeEntryIdUnsafe(),
426 event,
427 primaryKey,
428 payload
429 }, { disableChecks: true })
430 ),
431 effect,
432 (entry, exit) =>
433 Effect.suspend(() => {
434 if (exit._tag === "Failure" || byId.has(entry.idString)) return Effect.void
435 journal.push(entry)
436 byId.set(entry.idString, entry)
437 remotes.forEach((remote) => {
438 remote.missing.push(entry)
439 })
440 return PubSub.publish(pubsub, entry)
441 })
442 )
443 },
444 writeFromRemote: Effect.fnUntraced(function*(options) {
445 const remote = ensureRemote(options.remoteId)
446 const uncommittedRemotes: Array<RemoteEntry> = []
447 const uncommitted: Array<Entry> = []
448 const duplicateEntries: Array<Entry> = []
449 for (const remoteEntry of options.entries) {
450 if (byId.has(remoteEntry.entry.idString)) {
451 duplicateEntries.push(remoteEntry.entry)
452 if (remoteEntry.remoteSequence >= remote.sequence) {
453 remote.sequence = remoteEntry.remoteSequence + 1
454 }
455 continue
456 }
457 uncommittedRemotes.push(remoteEntry)
458 uncommitted.push(remoteEntry.entry)
459 }
460
461 const compacted = options.compact
462 ? yield* options.compact(uncommittedRemotes)
463 : uncommitted
464
465 for (const originEntry of compacted) {
466 const entryMillis = entryIdMillis(originEntry.id)
467 const conflicts: Array<Entry> = []
468 for (let i = journal.length - 1; i >= -1; i--) {
469 const entry = journal[i]
470 if (entry !== undefined && entry.createdAtMillis > entryMillis) {
471 continue
472 }
473 for (let j = i + 1; j < journal.length; j++) {
474 const scannedEntry = journal[j]!
475 if (scannedEntry.event === originEntry.event && scannedEntry.primaryKey === originEntry.primaryKey) {
476 conflicts.push(scannedEntry)
477 }
478 }
479 yield* options.effect({ entry: originEntry, conflicts })
480 break
481 }
482 }
483 for (const remoteEntry of uncommittedRemotes) {
484 journal.push(remoteEntry.entry)
485 byId.set(remoteEntry.entry.idString, remoteEntry.entry)
486 remotes.forEach((target) => {
487 if (target !== remote) {
488 target.missing.push(remoteEntry.entry)
489 }
490 })
491 if (remoteEntry.remoteSequence >= remote.sequence) {
492 remote.sequence = remoteEntry.remoteSequence + 1
493 }
494 }
495 journal.sort((a, b) => a.createdAtMillis - b.createdAtMillis)
496 remotes.forEach((remote) => {
497 remote.missing.sort((a, b) => a.createdAtMillis - b.createdAtMillis)
498 })
499 return {
500 duplicateEntries
501 }
502 }),
503 withRemoteUncommited: (remoteId, f) =>
504 Effect.acquireUseRelease(
505 Effect.sync(() => ensureRemote(remoteId).missing.slice()),
506 (entries) => Arr.isReadonlyArrayNonEmpty(entries) ? Effect.asSome(f(entries)) : Effect.succeedNone,
507 (entries, exit) =>
508 Effect.sync(() => {
509 if (exit._tag === "Failure") return
510 const last = entries[entries.length - 1]
511 if (!last) return
512 const remote = ensureRemote(remoteId)
513 for (let i = remote.missing.length - 1; i >= 0; i--) {
514 if (remote.missing[i].id === last.id) {
515 remote.missing = remote.missing.slice(i + 1)
516 break
517 }
518 }
519 })
520 ),
521 nextRemoteSequence: (remoteId) => Effect.sync(() => ensureRemote(remoteId).sequence),
522 changes: PubSub.subscribe(pubsub),
523 destroy: Effect.sync(() => {
524 journal.length = 0
525 byId.clear()
526 remotes.clear()
527 }),
528 withLock
529 })
530})
531
532/**
533 * Layer that provides an in-memory `EventJournal`.
534 *
535 * **Gotchas**
536 *
537 * All journal data is stored in process memory and is not persisted across layer
538 * lifetimes.
539 *
540 * @stability unstable
541 * @category layers
542 * @since 4.0.0
543 */
544export const layerMemory: Layer.Layer<EventJournal> = Layer.effect(EventJournal, makeMemory)
545
546/**
547 * Creates an `EventJournal` backed by IndexedDB.
548 *
549 * **Details**
550 *
551 * The journal stores entries and remote replication metadata in the configured
552 * browser database, publishes local changes, and requires `Scope` so the database
553 * connection can be closed when the scope ends.
554 *
555 * @stability unstable
556 * @category constructors
557 * @since 4.0.0
558 */
559export const makeIndexedDb = (options?: {
560 readonly database?: string
561}): Effect.Effect<EventJournal["Service"], EventJournalError, Scope> =>
562 Effect.gen(function*() {
563 const database = options?.database ?? "effect_event_journal"
564 const openRequest = indexedDB.open(database, 1)
565 openRequest.onupgradeneeded = () => {
566 const db = openRequest.result
567
568 const entries = db.createObjectStore("entries", { keyPath: "id" })
569 entries.createIndex("id", "id", { unique: true })
570 entries.createIndex("event", "event")
571
572 const remotes = db.createObjectStore("remotes", { keyPath: ["remoteId", "entryId"] })
573 remotes.createIndex("id", ["remoteId", "entryId"], { unique: true })
574 remotes.createIndex("sequence", ["remoteId", "sequence"], { unique: true })
575
576 const remoteEntryId = db.createObjectStore("remoteEntryId", { keyPath: ["remoteId"] })
577 remoteEntryId.createIndex("id", "remoteId", { unique: true })
578 }
579
580 const db = yield* Effect.acquireRelease(
581 idbReq("open", () => openRequest),
582 (db) => Effect.sync(() => db.close())
583 )
584
585 const pubsub = yield* PubSub.unbounded<Entry>()
586
587 return EventJournal.of({
588 entries: idbReq("entries", () =>
589 db.transaction("entries", "readonly")
590 .objectStore("entries")
591 .index("id")
592 .getAll()).pipe(
593 Effect.flatMap((_) =>
594 decodeEntryIdbArray(_).pipe(
595 Effect.mapError((cause) => new EventJournalError({ method: "entries", cause }))
596 )
597 )
598 ),
599 write: ({ effect, event, payload, primaryKey }) =>
600 Effect.uninterruptibleMask((restore) => {
601 const entry = new Entry({
602 id: makeEntryIdUnsafe(),
603 event,
604 primaryKey,
605 payload
606 }, { disableChecks: true })
607 return restore(effect(entry)).pipe(
608 Effect.tap(
609 idbReq("write", () =>
610 db.transaction("entries", "readwrite")
611 .objectStore("entries")
612 .put(encodeEntryIdb(entry)))
613 ),
614 Effect.tap(PubSub.publish(pubsub, entry))
615 )
616 }),
617 writeFromRemote: Effect.fnUntraced(function*(options) {
618 const uncommitted: Array<Entry> = []
619 const uncommittedRemotes: Array<RemoteEntry> = []
620 const duplicateEntries: Array<Entry> = []
621
622 yield* Effect.callback<void, EventJournalError>((resume) => {
623 const tx = db.transaction(["entries", "remotes"], "readwrite")
624 const entries = tx.objectStore("entries")
625 const remotes = tx.objectStore("remotes")
626 const iterator = options.entries[Symbol.iterator]()
627 const handleNext = (state: IteratorResult<RemoteEntry, void>) => {
628 if (state.done) return
629 const remoteEntry = state.value
630 const entry = remoteEntry.entry
631 const entryIdKey = entry.id as IDBValidKey
632 entries.get(entryIdKey).onsuccess = (event) => {
633 if (event.target && "result" in event.target && event.target.result) {
634 duplicateEntries.push(entry)
635 remotes.put({
636 remoteId: options.remoteId,
637 entryId: entry.id,
638 sequence: remoteEntry.remoteSequence
639 })
640 handleNext(iterator.next())
641 return
642 }
643 uncommitted.push(entry)
644 uncommittedRemotes.push(remoteEntry)
645 handleNext(iterator.next())
646 }
647 }
648 handleNext(iterator.next())
649 tx.oncomplete = () => resume(Effect.void)
650 tx.onerror = () => resume(Effect.fail(new EventJournalError({ method: "writeFromRemote", cause: tx.error })))
651 return Effect.sync(() => tx.abort())
652 })
653
654 const compacted = options.compact
655 ? yield* options.compact(uncommittedRemotes)
656 : uncommitted
657
658 for (const originEntry of compacted) {
659 const conflicts: Array<Entry> = []
660 yield* Effect.callback<void, EventJournalError>((resume) => {
661 const tx = db.transaction("entries", "readonly")
662 const entries = tx.objectStore("entries")
663 const cursorRequest = entries.index("id").openCursor(
664 IDBKeyRange.lowerBound(originEntry.id as IDBValidKey, true),
665 "next"
666 )
667 cursorRequest.onsuccess = () => {
668 const cursor = cursorRequest.result
669 if (!cursor) return
670 const decodedEntry = decodeEntryIdb(cursor.value)
671 if (
672 decodedEntry.event === originEntry.event &&
673 decodedEntry.primaryKey === originEntry.primaryKey
674 ) {
675 conflicts.push(decodedEntry)
676 }
677 cursor.continue()
678 }
679 tx.oncomplete = () => resume(Effect.void)
680 tx.onerror = () =>
681 resume(Effect.fail(new EventJournalError({ method: "writeFromRemote", cause: tx.error })))
682 return Effect.sync(() => tx.abort())
683 })
684
685 yield* options.effect({ entry: originEntry, conflicts })
686 }
687
688 yield* Effect.callback<void, EventJournalError>((resume) => {
689 const tx = db.transaction(["entries", "remotes"], "readwrite")
690 const entries = tx.objectStore("entries")
691 const remotes = tx.objectStore("remotes")
692 for (const remoteEntry of uncommittedRemotes) {
693 entries.add(encodeEntryIdb(remoteEntry.entry))
694 remotes.put({
695 remoteId: options.remoteId,
696 entryId: remoteEntry.entry.id,
697 sequence: remoteEntry.remoteSequence
698 })
699 }
700 tx.oncomplete = () => resume(Effect.void)
701 tx.onerror = () => resume(Effect.fail(new EventJournalError({ method: "writeFromRemote", cause: tx.error })))
702 return Effect.sync(() => tx.abort())
703 })
704 return {
705 duplicateEntries
706 }
707 }),
708 withRemoteUncommited: (remoteId, f) =>
709 Effect.callback<ReadonlyArray<Entry>, EventJournalError>((resume) => {
710 const entries: Array<Entry> = []
711 const tx = db.transaction(["entries", "remotes", "remoteEntryId"], "readwrite")
712
713 const entriesStore = tx.objectStore("entries")
714 const remotesStore = tx.objectStore("remotes")
715 const remoteEntryIdStore = tx.objectStore("remoteEntryId")
716
717 const remoteIdKey = remoteId as IDBValidKey
718 const request = remoteEntryIdStore.get(remoteIdKey) as IDBRequest<{ entryId: IDBValidKey } | undefined>
719 request.onsuccess = () => {
720 const startEntryId = request.result?.entryId
721 const entryCursor = entriesStore.index("id").openCursor(
722 startEntryId ? IDBKeyRange.lowerBound(startEntryId, true) : null,
723 "next"
724 )
725 entryCursor.onsuccess = () => {
726 const cursor = entryCursor.result
727 if (!cursor) return
728 const entry = decodeEntryIdb(cursor.value)
729 remotesStore.get([remoteIdKey, entry.id as IDBValidKey]).onsuccess = (event) => {
730 if (!(event.target && "result" in event.target && event.target.result)) entries.push(entry)
731 cursor.continue()
732 }
733 }
734 }
735
736 tx.oncomplete = () => resume(Effect.succeed(entries))
737 tx.onerror = () =>
738 resume(Effect.fail(new EventJournalError({ method: "withRemoteUncommited", cause: tx.error })))
739 return Effect.sync(() => tx.abort())
740 }).pipe(
741 Effect.flatMap((entries) => {
742 if (!Arr.isReadonlyArrayNonEmpty(entries)) return Effect.succeedNone
743 const entryId = entries[entries.length - 1].id
744 return Effect.uninterruptibleMask((restore) =>
745 restore(f(entries)).pipe(
746 Effect.tap(
747 idbReq("withRemoteUncommited", () =>
748 db.transaction("remoteEntryId", "readwrite").objectStore("remoteEntryId").put({
749 remoteId,
750 entryId
751 }))
752 ),
753 Effect.asSome
754 )
755 )
756 })
757 ),
758 nextRemoteSequence: (remoteId) =>
759 Effect.callback<number, EventJournalError>((resume) => {
760 const tx = db.transaction("remotes", "readonly")
761 let sequence = 0
762 const remoteIdKey = remoteId as IDBValidKey
763 const cursorRequest = tx.objectStore("remotes").index("sequence").openCursor(
764 IDBKeyRange.bound([remoteIdKey, 0], [remoteIdKey, Infinity]),
765 "prev"
766 )
767 cursorRequest.onsuccess = () => {
768 const cursor = cursorRequest.result
769 if (!cursor) return
770 sequence = cursor.value.sequence + 1
771 }
772 tx.oncomplete = () => resume(Effect.succeed(sequence))
773 tx.onerror = () =>
774 resume(Effect.fail(new EventJournalError({ method: "nextRemoteSequence", cause: tx.error })))
775 return Effect.sync(() => tx.abort())
776 }),
777 changes: PubSub.subscribe(pubsub),
778 destroy: Effect.sync(() => {
779 indexedDB.deleteDatabase(database)
780 }),
781 withLock: yield* makeBrowserWithLock(database)
782 })
783 })
784
785const makeBrowserWithLock = Effect.fnUntraced(function*(key: string) {
786 if (typeof navigator !== "undefined" && "locks" in navigator) {
787 return (storeId: StoreId) => <A, E, R>(self: Effect.Effect<A, E, R>) =>
788 Effect.callback<A, E, R>((resume, signal) => {
789 navigator.locks.request(`${key}/${storeId}`, { signal }, () =>
790 new Promise<void>((resolve) => {
791 resume(Effect.onExit(self, () => {
792 resolve()
793 return Effect.void
794 }))
795 })).catch((defect) => resume(Effect.die(defect)))
796 })
797 }
798 const semaphores = new Map<StoreId, Semaphore.Semaphore>()
799 return (storeId: StoreId) => {
800 let semaphore = semaphores.get(storeId)
801 if (!semaphore) {
802 semaphore = Semaphore.makeUnsafe(1)
803 semaphores.set(storeId, semaphore)
804 }
805 return semaphore.withPermit
806 }
807})
808
809const decodeEntryIdb = Schema.decodeSync(Entry)
810const encodeEntryIdb = Schema.encodeSync(Entry)
811const EntryIdbArray = Schema.Array(Entry)
812const decodeEntryIdbArray = Schema.decodeUnknownEffect(EntryIdbArray)
813
814/**
815 * Provides `EventJournal` using the IndexedDB-backed implementation created by
816 * `makeIndexedDb`.
817 *
818 * @stability unstable
819 * @category layers
820 * @since 4.0.0
821 */
822export const layerIndexedDb = (options?: {
823 readonly database?: string
824}): Layer.Layer<EventJournal, EventJournalError> =>
825 Layer.effect(
826 EventJournal,
827 makeIndexedDb(options)
828 )
829
830const idbReq = <T>(method: string, evaluate: () => IDBRequest<T>) =>
831 Effect.callback<T, EventJournalError>((resume) => {
832 const request = evaluate()
833 if (request.readyState === "done") {
834 resume(Effect.succeed(request.result))
835 return
836 }
837 request.onsuccess = () => resume(Effect.succeed(request.result))
838 request.onerror = () => resume(Effect.fail(new EventJournalError({ method, cause: request.error })))
839 })