1/**
2 * Runtime for writing typed events to an event journal.
3 *
4 * `EventLog` combines event groups, handlers, a journal, local identity,
5 * optional remote replicas, and reactivity hooks. Writers send typed payloads
6 * through a client; the matching handler runs first, and the journal entry is
7 * committed only after the handler succeeds. This module also contains the
8 * layers and helpers needed to assemble that runtime.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Context from "../Context.ts"
14import * as Effect from "../Effect.ts"
15import * as FiberMap from "../FiberMap.ts"
16import { constant, identity } from "../Function.ts"
17import * as InternalRecord from "../internal/record.ts"
18import * as Layer from "../Layer.ts"
19import type { Pipeable } from "../Pipeable.ts"
20import { pipeArguments } from "../Pipeable.ts"
21import * as Predicate from "../Predicate.ts"
22import * as PubSub from "../PubSub.ts"
23import * as Queue from "../Queue.ts"
24import { Reactivity } from "../reactivity/Reactivity.ts"
25import * as ReactivityLayer from "../reactivity/Reactivity.ts"
26import type * as Record from "../Record.ts"
27import * as Redacted from "../Redacted.ts"
28import * as Schedule from "../Schedule.ts"
29import * as Schema from "../Schema.ts"
30import * as SchemaGetter from "../SchemaGetter.ts"
31import type * as Scope from "../Scope.ts"
32import type { Covariant } from "../Types.ts"
33import type * as Event from "./Event.ts"
34import type * as EventGroup from "./EventGroup.ts"
35import { Entry, EventJournal, type EventJournalError, makeEntryIdUnsafe, type RemoteId } from "./EventJournal.ts"
36import * as EventLogEncryption from "./EventLogEncryption.ts"
37import { StoreId } from "./EventLogMessage.ts"
38import type { EventLogRemote } from "./EventLogRemote.ts"
39
40/**
41 * Service for writing typed event-log events through registered handlers.
42 *
43 * **Details**
44 *
45 * `write` encodes the event payload, runs the matching handler, commits the entry
46 * only when the handler succeeds, and exposes access to the underlying journal
47 * entries and destroy operation.
48 *
49 * @stability unstable
50 * @category services
51 * @since 4.0.0
52 */
53export class EventLog extends Context.Service<EventLog, {
54 readonly write: <Groups extends EventGroup.Any, Tag extends Event.Tag<EventGroup.Events<Groups>>>(options: {
55 readonly schema: EventLogSchema<Groups>
56 readonly event: Tag
57 readonly payload: Event.PayloadWithTag<EventGroup.Events<Groups>, Tag>
58 }) => Effect.Effect<
59 Event.SuccessWithTag<EventGroup.Events<Groups>, Tag>,
60 Event.ErrorWithTag<EventGroup.Events<Groups>, Tag> | EventJournalError
61 >
62 readonly entries: Effect.Effect<ReadonlyArray<Entry>, EventJournalError>
63 readonly destroy: Effect.Effect<void, EventJournalError>
64}>()("effect/eventlog/EventLog") {}
65
66/**
67 * Service that collects event handlers, compaction handlers, remote replicas,
68 * and reactivity invalidation keys.
69 *
70 * @stability unstable
71 * @category services
72 * @since 4.0.0
73 */
74export class Registry extends Context.Service<Registry, {
75 readonly registerHandlerUnsafe: (options: {
76 readonly event: string
77 readonly handler: Handlers.Item<any>
78 }) => void
79
80 readonly handlers: ReadonlyMap<string, Handlers.Item<any>>
81
82 readonly registerCompaction: (options: {
83 readonly events: ReadonlyArray<string>
84 readonly effect: (options: {
85 readonly entries: ReadonlyArray<Entry>
86 readonly write: (entry: Entry) => Effect.Effect<void>
87 }) => Effect.Effect<void>
88 }) => Effect.Effect<void, never, Scope.Scope>
89
90 readonly compactors: ReadonlyMap<string, {
91 readonly events: ReadonlySet<string>
92 readonly effect: (options: {
93 readonly entries: ReadonlyArray<Entry>
94 readonly write: (entry: Entry) => Effect.Effect<void>
95 }) => Effect.Effect<void>
96 }>
97
98 readonly registerRemote: (remote: EventLogRemote["Service"]) => Effect.Effect<void, never, Scope.Scope>
99 readonly handleRemote: (handler: (remote: EventLogRemote["Service"]) => Effect.Effect<void>) => Effect.Effect<void>
100
101 readonly registerReactivity: (keys: Record<string, ReadonlyArray<string>>) => Effect.Effect<void, never, Scope.Scope>
102
103 readonly reactivityKeys: Record<string, ReadonlyArray<string>>
104}>()("effect/eventlog/EventLog/Registry") {}
105
106/**
107 * Provides an in-memory `Registry` for event handlers, compactors, remote
108 * replicas, and reactivity keys.
109 *
110 * @stability unstable
111 * @category layers
112 * @since 4.0.0
113 */
114export const layerRegistry = Layer.effect(
115 Registry,
116 Effect.gen(function*() {
117 const handlers = new Map<string, Handlers.Item<any>>()
118 const compactors = new Map<string, {
119 readonly events: ReadonlySet<string>
120 readonly effect: (options: {
121 readonly entries: ReadonlyArray<Entry>
122 readonly write: (entry: Entry) => Effect.Effect<void>
123 }) => Effect.Effect<void>
124 }>()
125
126 const remoteFiberMap = yield* FiberMap.make<RemoteId>()
127 const remotes = new Map<RemoteId, EventLogRemote["Service"]>()
128 let remoteHandler: (remote: EventLogRemote["Service"]) => Effect.Effect<void> = (_) => Effect.void
129
130 const reactivityKeys: Record<string, ReadonlyArray<string>> = {}
131 return Registry.of({
132 registerHandlerUnsafe(options) {
133 handlers.set(options.event, options.handler)
134 },
135 handlers,
136 registerCompaction: (options) =>
137 Effect.sync(() => {
138 const events = new Set(options.events)
139 const compactor = {
140 events,
141 effect: options.effect
142 }
143 for (const event of options.events) {
144 compactors.set(event, compactor)
145 }
146 }),
147 compactors,
148 registerRemote: (remote) =>
149 Effect.acquireRelease(
150 Effect.suspend(() => {
151 remotes.set(remote.id, remote)
152 return Effect.asVoid(FiberMap.run(remoteFiberMap, remote.id, remoteHandler(remote)))
153 }),
154 () => {
155 remotes.delete(remote.id)
156 return FiberMap.remove(remoteFiberMap, remote.id)
157 }
158 ),
159 handleRemote(handler) {
160 remoteHandler = handler
161 return Effect.forEach(remotes, ([id, remote]) => FiberMap.run(remoteFiberMap, id, handler(remote)), {
162 discard: true
163 })
164 },
165 registerReactivity: (keys) =>
166 Effect.sync(() => {
167 for (const [key, value] of Object.entries(keys)) {
168 InternalRecord.assignProperty(reactivityKeys, key, value)
169 }
170 }),
171 reactivityKeys
172 })
173 })
174)
175
176/**
177 * Context service for an event-log identity containing a public key and redacted
178 * private key material.
179 *
180 * **Details**
181 *
182 * The identity is used by remote replication for authentication and by the
183 * encryption service to derive signing and encryption keys.
184 *
185 * @stability unstable
186 * @category services
187 * @since 4.0.0
188 */
189export class Identity extends Context.Service<Identity, {
190 readonly publicKey: string
191 readonly privateKey: Redacted.Redacted<Uint8Array<ArrayBuffer>>
192}>()("effect/eventlog/EventLog/Identity") {}
193
194/**
195 * Type-level identifier used to brand `EventLogSchema` values.
196 *
197 * @stability unstable
198 * @category type IDs
199 * @since 4.0.0
200 */
201export type SchemaTypeId = "~effect/eventlog/EventLog/Schema"
202
203/**
204 * Runtime property key used to identify `EventLogSchema` values.
205 *
206 * @stability unstable
207 * @category type IDs
208 * @since 4.0.0
209 */
210export const SchemaTypeId: SchemaTypeId = "~effect/eventlog/EventLog/Schema"
211
212/**
213 * Returns `true` when a value carries the `EventLogSchema` marker.
214 *
215 * @stability unstable
216 * @category guards
217 * @since 4.0.0
218 */
219export const isEventLogSchema = (u: unknown): u is EventLogSchema<EventGroup.Any> =>
220 Predicate.hasProperty(u, SchemaTypeId)
221
222/**
223 * Schema describing the event groups that can be written through an `EventLog`.
224 *
225 * @stability unstable
226 * @category schemas
227 * @since 4.0.0
228 */
229export interface EventLogSchema<Groups extends EventGroup.Any> {
230 readonly [SchemaTypeId]: SchemaTypeId
231 readonly groups: ReadonlyArray<Groups>
232}
233
234/**
235 * Creates an `EventLogSchema` from one or more event groups.
236 *
237 * @stability unstable
238 * @category schemas
239 * @since 4.0.0
240 */
241export const schema = <Groups extends ReadonlyArray<EventGroup.Any>>(
242 ...groups: Groups
243): EventLogSchema<Groups[number]> => {
244 const EventLog = Object.assign(function EventLog() {}, {
245 [SchemaTypeId]: SchemaTypeId,
246 groups
247 }) satisfies EventLogSchema<Groups[number]>
248 return EventLog
249}
250
251/**
252 * Type-level identifier used to brand `Handlers` values.
253 *
254 * @stability unstable
255 * @category type IDs
256 * @since 4.0.0
257 */
258export type HandlersTypeId = "~effect/eventlog/EventLog/Handlers"
259
260/**
261 * Runtime property key used to identify `Handlers` values.
262 *
263 * @stability unstable
264 * @category type IDs
265 * @since 4.0.0
266 */
267export const HandlersTypeId: HandlersTypeId = "~effect/eventlog/EventLog/Handlers"
268
269/**
270 * Builder for the handlers associated with an `EventGroup`.
271 *
272 * **Details**
273 *
274 * The `Events` type parameter tracks the event tags that still need handlers, and
275 * each call to `handle` records a handler while accumulating any required
276 * services.
277 *
278 * @stability unstable
279 * @category handlers
280 * @since 4.0.0
281 */
282export interface Handlers<
283 R,
284 Events extends Event.Any = never
285> extends Pipeable {
286 readonly [HandlersTypeId]: {
287 _Events: Covariant<Events>
288 }
289 readonly group: EventGroup.AnyWithProps
290 readonly handlers: Record.ReadonlyRecord<string, Handlers.Item<R>>
291 readonly context: Context.Context<R>
292
293 /**
294 * Add the implementation for an `Event` to a `Handlers` group.
295 */
296 handle<Tag extends Event.Tag<Events>, R1>(
297 name: Tag,
298 handler: (options: {
299 readonly storeId: StoreId
300 readonly payload: Event.PayloadWithTag<Events, Tag>
301 readonly entry: Entry
302 readonly conflicts: ReadonlyArray<{
303 readonly entry: Entry
304 readonly payload: Event.PayloadWithTag<Events, Tag>
305 }>
306 }) => Effect.Effect<Event.SuccessWithTag<Events, Tag>, Event.ErrorWithTag<Events, Tag>, R1>
307 ): Handlers<
308 R | R1,
309 Event.ExcludeTag<Events, Tag>
310 >
311}
312
313/**
314 * Namespace containing helper types for `Handlers` values and handler-producing
315 * layers.
316 *
317 * @stability unstable
318 * @since 4.0.0
319 */
320export declare namespace Handlers {
321 /**
322 * Type that matches any `Handlers` value regardless of its services or remaining
323 * events.
324 *
325 * @stability unstable
326 * @category handlers
327 * @since 4.0.0
328 */
329 export interface Any {
330 readonly [HandlersTypeId]: unknown
331 }
332
333 /**
334 * Runtime representation of one registered event handler, including its event
335 * metadata, captured context, and handler function.
336 *
337 * @stability unstable
338 * @category handlers
339 * @since 4.0.0
340 */
341 export type Item<R> = {
342 readonly event: Event.AnyWithProps
343 readonly context: Context.Context<R>
344 readonly handler: (options: {
345 readonly storeId: StoreId
346 readonly payload: unknown
347 readonly entry: Entry
348 readonly conflicts: ReadonlyArray<{
349 readonly entry: Entry
350 readonly payload: unknown
351 }>
352 }) => Effect.Effect<unknown, unknown, R>
353 }
354
355 /**
356 * Validates that a handler builder returned all required handlers.
357 *
358 * **Details**
359 *
360 * If any event tag remains unhandled, the type evaluates to an explanatory
361 * compile-time error string.
362 *
363 * @stability unstable
364 * @category handlers
365 * @since 4.0.0
366 */
367 export type ValidateReturn<A> = A extends (
368 | Handlers<
369 infer _R,
370 infer _Events
371 >
372 | Effect.Effect<
373 Handlers<
374 infer _R,
375 infer _Events
376 >,
377 infer _EX,
378 infer _RX
379 >
380 ) ? [_Events] extends [never] ? A
381 : `Event not handled: ${Event.Tag<_Events>}` :
382 `Must return the implemented handlers`
383
384 /**
385 * Extracts the error type from an effect that produces `Handlers`.
386 *
387 * @stability unstable
388 * @category handlers
389 * @since 4.0.0
390 */
391 export type Error<A> = A extends Effect.Effect<
392 Handlers<
393 infer _R,
394 infer _Events
395 >,
396 infer _EX,
397 infer _RX
398 > ? _EX :
399 never
400
401 /**
402 * Computes the services required by a `Handlers` value or by an effect that
403 * produces one, including event schema services.
404 *
405 * @stability unstable
406 * @category handlers
407 * @since 4.0.0
408 */
409 export type Services<A> = A extends Handlers<
410 infer _R,
411 infer _Events
412 > ? _R | Event.Services<_Events> :
413 A extends Effect.Effect<
414 Handlers<
415 infer _R,
416 infer _Events
417 >,
418 infer _EX,
419 infer _RX
420 > ? _R | _RX | Event.Services<_Events> :
421 never
422}
423
424/**
425 * Context reference for the store id used by event-log writes and remote
426 * replication.
427 *
428 * **Details**
429 *
430 * Defaults to the branded store id `"default"`.
431 *
432 * @stability unstable
433 * @category services
434 * @since 4.0.0
435 */
436export class CurrentStoreId extends Context.Reference<StoreId>("effect/eventlog/EventLog/CurrentStoreId", {
437 defaultValue: constant(StoreId.make("default"))
438}) {}
439
440const RedactedUint8Array = Schema.Uint8ArrayFromBase64.pipe(
441 Schema.decodeTo(Schema.Redacted(Schema.Uint8Array), {
442 decode: SchemaGetter.transform((value) => Redacted.make(value)),
443 encode: SchemaGetter.transform((value) => Redacted.value(value))
444 })
445)
446
447/**
448 * Schema for an event-log identity with a string public key and redacted
449 * base64-encoded private key bytes.
450 *
451 * @stability unstable
452 * @category schemas
453 * @since 4.0.0
454 */
455export const IdentitySchema = Schema.Struct({
456 publicKey: Schema.String,
457 privateKey: RedactedUint8Array
458})
459
460const IdentityEncodedSchema = Schema.Struct({
461 publicKey: Schema.String,
462 privateKey: Schema.Uint8ArrayFromBase64
463})
464
465const IdentityStringSchema = Schema.StringFromBase64Url.pipe(
466 Schema.decodeTo(Schema.fromJsonString(IdentityEncodedSchema))
467)
468
469/**
470 * Decodes a base64url identity string produced by `encodeIdentityString`.
471 *
472 * **Gotchas**
473 *
474 * Invalid input throws a schema decoding error.
475 *
476 * @stability unstable
477 * @category constructors
478 * @since 4.0.0
479 */
480export const decodeIdentityString = (value: string): Identity["Service"] => {
481 const decoded = Schema.decodeUnknownSync(IdentityStringSchema)(value)
482 return {
483 publicKey: decoded.publicKey,
484 privateKey: Redacted.make(decoded.privateKey as Uint8Array<ArrayBuffer>)
485 }
486}
487
488/**
489 * Encodes an event-log identity as a base64url string containing the public key
490 * and private key bytes.
491 *
492 * @stability unstable
493 * @category constructors
494 * @since 4.0.0
495 */
496export const encodeIdentityString = (identity: Identity["Service"]): string =>
497 Schema.encodeSync(IdentityStringSchema)({
498 publicKey: identity.publicKey,
499 privateKey: Redacted.value(identity.privateKey)
500 })
501
502/**
503 * Generates a new event-log identity using the configured
504 * `EventLogEncryption` service.
505 *
506 * @stability unstable
507 * @category constructors
508 * @since 4.0.0
509 */
510export const makeIdentity: Effect.Effect<Identity["Service"], never, EventLogEncryption.EventLogEncryption> =
511 EventLogEncryption.EventLogEncryption.use((_) => _.generateIdentity)
512
513const handlersProto = {
514 [HandlersTypeId]: {
515 _Events: identity
516 },
517 handle<Tag extends string, R1>(
518 this: Handlers<any, any>,
519 tag: Tag,
520 handler: (payload: unknown) => Effect.Effect<unknown, unknown, R1>
521 ): Handlers<any, any> {
522 return makeHandlers({
523 group: this.group,
524 context: this.context,
525 handlers: {
526 ...this.handlers,
527 [tag]: {
528 event: Object.hasOwn(this.group.events, tag) ? this.group.events[tag] : undefined!,
529 context: this.context,
530 handler
531 }
532 }
533 })
534 },
535 pipe() {
536 return pipeArguments(this, arguments)
537 }
538}
539
540const makeHandlers = (options: {
541 readonly group: EventGroup.AnyWithProps
542 readonly handlers: Record.ReadonlyRecord<string, Handlers.Item<any>>
543 readonly context: Context.Context<any>
544}): Handlers<any, any> => Object.assign(Object.create(handlersProto), options)
545
546/**
547 * Creates a layer that registers handlers for every event in an event group.
548 *
549 * **Details**
550 *
551 * The callback receives a `Handlers` builder; its return type is checked so every
552 * event in the group is handled.
553 *
554 * @stability unstable
555 * @category handlers
556 * @since 4.0.0
557 */
558export const group = <Events extends Event.Any, Return>(
559 group: EventGroup.EventGroup<Events>,
560 f: (handlers: Handlers<never, Events>) => Handlers.ValidateReturn<Return>
561): Layer.Layer<
562 Event.ToService<Events>,
563 Handlers.Error<Return>,
564 Exclude<Handlers.Services<Return>, Scope.Scope | Identity> | Registry
565> =>
566 Layer.effectDiscard(
567 Effect.gen(function*() {
568 const registry = yield* Registry
569 const context = yield* Effect.context<Exclude<Handlers.Services<Return>, Scope.Scope | Identity>>()
570 const result = f(makeHandlers({
571 group: group as EventGroup.AnyWithProps,
572 handlers: {},
573 context
574 }) as unknown as Handlers<never, Events>)
575 const handlers = Effect.isEffect(result)
576 ? (yield* (result as unknown as Effect.Effect<Handlers<any>>))
577 : (result as unknown as Handlers<any>)
578 for (const tag of Object.keys(handlers.handlers)) {
579 registry.registerHandlerUnsafe({ event: tag, handler: handlers.handlers[tag] })
580 }
581 })
582 ) as any
583
584/**
585 * Registers a compaction handler for an event group.
586 *
587 * **Details**
588 *
589 * During remote replay, matching entries are decoded, grouped by primary key, and
590 * passed to the compaction effect, which may write replacement entries.
591 *
592 * @stability unstable
593 * @category compaction
594 * @since 4.0.0
595 */
596export const groupCompaction = <Events extends Event.Any, R>(
597 group: EventGroup.EventGroup<Events>,
598 effect: (options: {
599 readonly primaryKey: string
600 readonly entries: ReadonlyArray<Entry>
601 readonly events: ReadonlyArray<Event.TaggedPayload<Events>>
602 readonly write: <Tag extends Event.Tag<Events>>(
603 tag: Tag,
604 payload: Event.PayloadWithTag<Events, Tag>
605 ) => Effect.Effect<void, never, Event.PayloadSchemaWithTag<Events, Tag>["EncodingServices"]>
606 }) => Effect.Effect<void, never, R>
607): Layer.Layer<never, never, R | Event.PayloadSchema<Events>["DecodingServices"] | Registry> =>
608 Layer.effectDiscard(
609 Effect.gen(function*() {
610 const registry = yield* Registry
611 const services = yield* Effect.context<R | Event.PayloadSchema<Events>["DecodingServices"]>()
612
613 yield* registry.registerCompaction({
614 events: Object.keys(group.events),
615 effect: Effect.fnUntraced(function*({ entries, write }): Effect.fn.Return<void> {
616 const isEventTag = (tag: string): tag is Event.Tag<Events> => Object.hasOwn(group.events, tag)
617 const decodePayload = <Tag extends Event.Tag<Events>>(tag: Tag, payload: Uint8Array) =>
618 Schema.decodeUnknownEffect(group.events[tag].payloadSchemaBinary)(payload).pipe(
619 Effect.updateContext((input) => Context.merge(services, input)),
620 Effect.orDie
621 ) as unknown as Effect.Effect<Event.PayloadWithTag<Events, Tag>>
622 const writePayload = Effect.fnUntraced(function*<Tag extends Event.Tag<Events>>(
623 timestamp: number,
624 tag: Tag,
625 payload: Event.PayloadWithTag<Events, Tag>
626 ): Effect.fn.Return<void, never, Event.PayloadSchemaWithTag<Events, Tag>["EncodingServices"]> {
627 const event = Object.hasOwn(group.events, tag) ? group.events[tag] : undefined!
628 const entry = new Entry({
629 id: makeEntryIdUnsafe({ msecs: timestamp }),
630 event: tag,
631 payload: yield* Schema.encodeUnknownEffect(event.payloadSchemaBinary)(payload).pipe(
632 Effect.orDie
633 ) as any,
634 primaryKey: event.primaryKey(payload)
635 }, { disableChecks: true })
636 yield* write(entry)
637 })
638
639 const byPrimaryKey = new Map<
640 string,
641 {
642 readonly entries: Array<Entry>
643 readonly taggedPayloads: Array<Event.TaggedPayload<Events>>
644 }
645 >()
646 for (const entry of entries) {
647 if (!isEventTag(entry.event)) {
648 continue
649 }
650 const payload = yield* decodePayload(entry.event, entry.payload)
651 const record = byPrimaryKey.get(entry.primaryKey)
652 const taggedPayload = { _tag: entry.event, payload } as unknown as Event.TaggedPayload<Events>
653 if (record) {
654 record.entries.push(entry)
655 record.taggedPayloads.push(taggedPayload)
656 } else {
657 byPrimaryKey.set(entry.primaryKey, {
658 entries: [entry],
659 taggedPayloads: [taggedPayload]
660 })
661 }
662 }
663
664 for (const [primaryKey, { entries, taggedPayloads }] of byPrimaryKey) {
665 yield* Effect.orDie(
666 effect({
667 primaryKey,
668 entries,
669 events: taggedPayloads,
670 write(tag, payload) {
671 return Effect.orDie(writePayload(entries[0].createdAtMillis, tag, payload))
672 }
673 }).pipe(
674 Effect.updateContext((input) => Context.merge(services, input))
675 )
676 ) as any
677 }
678 })
679 })
680 })
681 )
682
683/**
684 * Registers reactivity keys to invalidate when events from a group are written or
685 * replayed.
686 *
687 * **Details**
688 *
689 * Pass a single key list for all events or a mapping from event tag to key list.
690 *
691 * @stability unstable
692 * @category reactivity
693 * @since 4.0.0
694 */
695export const groupReactivity = <Events extends Event.Any>(
696 group: EventGroup.EventGroup<Events>,
697 keys:
698 | { readonly [Tag in Event.Tag<Events>]?: ReadonlyArray<string> }
699 | ReadonlyArray<string>
700): Layer.Layer<never, never, Registry> =>
701 Effect.gen(function*() {
702 const registry = yield* Registry
703 if (!Array.isArray(keys)) {
704 yield* registry.registerReactivity(keys as Record.ReadonlyRecord<string, ReadonlyArray<string>>)
705 return
706 }
707 const obj: Record<string, ReadonlyArray<string>> = Object.create(null)
708 for (const tag of Object.keys(group.events)) {
709 obj[tag] = keys
710 }
711 yield* registry.registerReactivity(obj)
712 }).pipe(
713 Layer.effectDiscard
714 )
715
716/**
717 * Builds the effect used to replay entries received from a remote event log.
718 *
719 * **Details**
720 *
721 * The returned handler decodes the entry and conflicts with the registered event
722 * schema, runs the matching handler with the supplied identity and store id, logs
723 * failures, and invalidates configured reactivity keys.
724 *
725 * @stability unstable
726 * @category handlers
727 * @since 4.0.0
728 */
729export const makeReplayFromRemote = (options: {
730 readonly handlers: ReadonlyMap<string, Handlers.Item<any>>
731 readonly storeId: StoreId
732 readonly identity: Identity["Service"]
733 readonly reactivity: Reactivity
734 readonly reactivityKeys: Record<string, ReadonlyArray<string>>
735 readonly logAnnotations: {
736 readonly service: string
737 readonly effect: string
738 }
739}) =>
740 Effect.fnUntraced(
741 function*({ conflicts, entry }: {
742 readonly entry: Entry
743 readonly conflicts: ReadonlyArray<Entry>
744 }): Effect.fn.Return<void, Schema.SchemaError> {
745 const handler = options.handlers.get(entry.event) as Handlers.Item<any> | undefined
746 if (!handler) {
747 return yield* Effect.logDebug(`Event handler not found for: "${entry.event}"`)
748 }
749
750 const decodePayload = Schema.decodeUnknownEffect(handler.event.payloadSchemaBinary)
751 const decodedConflicts: Array<{ entry: Entry; payload: unknown }> = new Array(conflicts.length)
752 for (let i = 0; i < conflicts.length; i++) {
753 decodedConflicts[i] = {
754 entry: conflicts[i],
755 payload: yield* decodePayload(conflicts[i].payload).pipe(
756 Effect.updateContext((input) => Context.merge(handler.context, input))
757 ) as any
758 }
759 }
760
761 yield* decodePayload(entry.payload).pipe(
762 Effect.flatMap((payload) =>
763 handler.handler({
764 storeId: options.storeId,
765 payload,
766 entry,
767 conflicts: decodedConflicts
768 })
769 ),
770 Effect.provideService(Identity, options.identity),
771 Effect.updateContext((input) => Context.merge(handler.context, input)),
772 Effect.asVoid
773 ) as any
774
775 const keys = Object.hasOwn(options.reactivityKeys, entry.event)
776 ? options.reactivityKeys[entry.event]
777 : undefined
778 if (keys) {
779 for (const key of keys) {
780 options.reactivity.invalidateUnsafe({
781 [key]: [entry.primaryKey]
782 })
783 }
784 }
785 },
786 Effect.catchCause(Effect.logError),
787 (effect, { entry }) =>
788 Effect.annotateLogs(effect, {
789 ...options.logAnnotations,
790 entryId: entry.idString
791 })
792 )
793
794const remoteRetrySchedule = Schedule.min([
795 Schedule.exponential(200, 1.5),
796 Schedule.spaced({ seconds: 10 })
797])
798
799const make = Effect.gen(function*() {
800 const storeId = yield* CurrentStoreId
801 const identity = yield* Identity
802 const journal = yield* EventJournal
803 const registry = yield* Registry
804
805 const reactivity = yield* Reactivity
806 const replayFromRemote = makeReplayFromRemote({
807 handlers: registry.handlers,
808 storeId,
809 identity,
810 reactivity,
811 reactivityKeys: registry.reactivityKeys,
812 logAnnotations: {
813 service: "EventLog",
814 effect: "writeFromRemote"
815 }
816 })
817
818 const invalidateReactivityEntries = (entries: ReadonlyArray<Entry>) =>
819 Effect.sync(() => {
820 for (const entry of entries) {
821 const keys = Object.hasOwn(registry.reactivityKeys, entry.event)
822 ? registry.reactivityKeys[entry.event]
823 : undefined
824 if (!keys) {
825 continue
826 }
827 for (const key of keys) {
828 reactivity.invalidateUnsafe({
829 [key]: [entry.primaryKey]
830 })
831 }
832 }
833 })
834
835 const runRemote = Effect.fnUntraced(
836 function*(remote: EventLogRemote["Service"]) {
837 const startSequence = yield* journal.nextRemoteSequence(remote.id)
838
839 yield* Effect.gen(function*() {
840 const changes = yield* remote.changes({ identity, startSequence, storeId })
841 while (true) {
842 const entries = yield* Queue.takeAll(changes)
843 yield* journal.writeFromRemote({
844 remoteId: remote.id,
845 entries: entries.flat(),
846 compact: registry.compactors.size > 0
847 ? Effect.fnUntraced(function*(remoteEntries) {
848 const finalEntries: Array<Entry> = []
849 const compactable = new Map<
850 (options: {
851 readonly entries: ReadonlyArray<Entry>
852 readonly write: (entry: Entry) => Effect.Effect<void>
853 }) => Effect.Effect<void>,
854 Array<Entry>
855 >()
856
857 for (let i = 0; i < remoteEntries.length; i++) {
858 const remoteEntry = remoteEntries[i]
859 const entry = remoteEntry.entry
860 const compactor = registry.compactors.get(entry.event)
861 if (!compactor) {
862 finalEntries.push(entry)
863 continue
864 }
865 let arr = compactable.get(compactor.effect)
866 if (!arr) {
867 arr = []
868 compactable.set(compactor.effect, arr)
869 }
870 arr.push(entry)
871 }
872
873 for (const [compact, entries] of compactable) {
874 yield* compact({
875 entries,
876 write(entry) {
877 return Effect.sync(() => {
878 finalEntries.push(entry)
879 })
880 }
881 })
882 }
883
884 return finalEntries.sort(Entry.Order)
885 })
886 : undefined,
887 effect: replayFromRemote
888 }).pipe(
889 Effect.tap(({ duplicateEntries }) => invalidateReactivityEntries(duplicateEntries)),
890 journal.withLock(storeId)
891 )
892 }
893 }).pipe(
894 Effect.scoped,
895 Effect.catchCause(Effect.logError),
896 Effect.repeat(remoteRetrySchedule),
897 Effect.annotateLogs({
898 service: "EventLog",
899 effect: "runRemote consume"
900 }),
901 Effect.forkScoped
902 )
903
904 const write = journal.withRemoteUncommited(remote.id, (entries) => remote.write({ identity, entries, storeId }))
905 const writeUntilSuccess = write.pipe(
906 Effect.tapCause(Effect.logDebug),
907 Effect.retry(remoteRetrySchedule)
908 )
909 yield* Effect.addFinalizer(() => Effect.ignore(write))
910 const changesSub = yield* journal.changes
911 const changes = yield* Queue.dropping<void>(1)
912 yield* PubSub.takeAll(changesSub).pipe(
913 Effect.andThen(Queue.offer(changes, undefined)),
914 Effect.forever,
915 Effect.forkScoped
916 )
917 yield* writeUntilSuccess
918 return yield* Queue.take(changes).pipe(
919 Effect.andThen(writeUntilSuccess),
920 Effect.catchCause(Effect.logError),
921 Effect.forever
922 )
923 },
924 Effect.scoped,
925 Effect.provideService(Identity, identity),
926 Effect.orDie
927 )
928
929 const writeHandler = Effect.fnUntraced(function*(handler: Handlers.Item<any>, options: {
930 readonly schema: EventLogSchema<any>
931 readonly event: string
932 readonly payload: unknown
933 }) {
934 const payload = yield* Schema.encodeUnknownEffect(handler.event.payloadSchemaBinary)(options.payload).pipe(
935 Effect.orDie
936 )
937 return yield* journal.withLock(storeId)(journal.write({
938 event: options.event,
939 primaryKey: handler.event.primaryKey(options.payload as never),
940 payload,
941 effect: (entry) =>
942 handler.handler({
943 storeId,
944 payload: options.payload,
945 entry,
946 conflicts: []
947 }).pipe(
948 Effect.updateContext((input) => Context.merge(handler.context, input)),
949 Effect.provideService(Identity, identity),
950 Effect.tap(() => {
951 const keys = Object.hasOwn(registry.reactivityKeys, entry.event)
952 ? registry.reactivityKeys[entry.event]
953 : undefined
954 if (keys) {
955 for (const key of keys) {
956 reactivity.invalidateUnsafe({
957 [key]: [entry.primaryKey]
958 })
959 }
960 }
961 return Effect.void
962 })
963 )
964 }))
965 })
966
967 const eventLogWrite = (options: {
968 readonly schema: EventLogSchema<any>
969 readonly event: string
970 readonly payload: unknown
971 }) => {
972 const handler = registry.handlers.get(options.event) as Handlers.Item<any> | undefined
973 if (handler === undefined) {
974 return Effect.die(`Event handler not found for: "${options.event}"`)
975 }
976 return writeHandler(handler, options)
977 }
978
979 yield* registry.handleRemote(runRemote)
980
981 return EventLog.of({
982 write: eventLogWrite as EventLog["Service"]["write"],
983 entries: journal.entries,
984 destroy: journal.destroy
985 })
986})
987
988/**
989 * Provides `EventLog` and `Registry` using the configured `EventJournal` and
990 * `Identity`.
991 *
992 * @stability unstable
993 * @category layers
994 * @since 4.0.0
995 */
996export const layerEventLog: Layer.Layer<EventLog | Registry, never, EventJournal | Identity> = Layer.effect(
997 EventLog,
998 make
999).pipe(
1000 Layer.provide(ReactivityLayer.layer),
1001 Layer.provideMerge(layerRegistry)
1002)
1003
1004/**
1005 * Combines event-group handler layers with the `EventLog` runtime for a schema.
1006 *
1007 * **When to use**
1008 *
1009 * Use when you need one layer that installs the shared `EventLog` runtime for
1010 * an `EventLogSchema` and registers an event-group handler layer for typed
1011 * writes.
1012 *
1013 * **Details**
1014 *
1015 * The supplied handler layer is provided with `layerEventLog`. The returned
1016 * layer provides `EventLog | Registry`, preserves the handler layer's error
1017 * type, and still requires its remaining services plus `EventJournal` and
1018 * `Identity`.
1019 *
1020 * **Gotchas**
1021 *
1022 * The schema argument does not register handlers by itself. Handler registration
1023 * comes from the supplied layer, and writing an event without a registered
1024 * handler dies with `Event handler not found for: "<tag>"`.
1025 *
1026 * @see {@link schema} for creating the schema argument from event groups
1027 * @see {@link group} for building the handler layer consumed by this layer
1028 * @see {@link layerEventLog} for installing the runtime and registry without combining a handler layer
1029 *
1030 * @stability unstable
1031 * @category layers
1032 * @since 4.0.0
1033 */
1034export const layer = <Groups extends EventGroup.Any, E, R>(
1035 _schema: EventLogSchema<Groups>,
1036 layer: Layer.Layer<EventGroup.ToService<Groups>, E, R>
1037): Layer.Layer<
1038 EventLog | Registry,
1039 E,
1040 Exclude<R, EventLog | Registry> | EventJournal | Identity
1041> =>
1042 layer.pipe(
1043 Layer.provideMerge(layerEventLog)
1044 )
1045
1046/**
1047 * Creates a typed client function for writing events defined by an
1048 * `EventLogSchema`.
1049 *
1050 * **Details**
1051 *
1052 * The returned function delegates to the `EventLog` service and preserves each
1053 * event's success and error types.
1054 *
1055 * @stability unstable
1056 * @category constructors
1057 * @since 4.0.0
1058 */
1059export const makeClient = <Groups extends EventGroup.Any>(
1060 schema: EventLogSchema<Groups>
1061): Effect.Effect<
1062 (<Tag extends Event.Tag<EventGroup.Events<Groups>>>(
1063 event: Tag,
1064 payload: Event.PayloadWithTag<EventGroup.Events<Groups>, Tag>
1065 ) => Effect.Effect<
1066 Event.SuccessWithTag<EventGroup.Events<Groups>, Tag>,
1067 Event.ErrorWithTag<EventGroup.Events<Groups>, Tag> | EventJournalError
1068 >),
1069 never,
1070 EventLog
1071> =>
1072 Effect.gen(function*() {
1073 const log = yield* EventLog
1074
1075 return <Tag extends Event.Tag<EventGroup.Events<Groups>>>(
1076 event: Tag,
1077 payload: Event.PayloadWithTag<EventGroup.Events<Groups>, Tag>
1078 ): Effect.Effect<
1079 Event.SuccessWithTag<EventGroup.Events<Groups>, Tag>,
1080 Event.ErrorWithTag<EventGroup.Events<Groups>, Tag> | EventJournalError
1081 > => log.write({ schema, event, payload })
1082 })