1/**
2 * Plaintext server implementation for the event-log remote protocol.
3 *
4 * This module accepts unencrypted event batches from remote clients, runs the
5 * registered event handlers, stores journal entries, and streams backlog plus
6 * live changes through the shared `EventLogServer` RPC protocol. It is intended
7 * for trusted deployments, local development, and tests where event data does
8 * not need a server-side encryption layer. The module also provides the
9 * services and layers needed to authorize requests, map stores, persist entries,
10 * and install the plaintext server.
11 *
12 * @stability unstable
13 * @since 4.0.0
14 */
15import * as Arr from "../Array.ts"
16import * as Context from "../Context.ts"
17import * as Data from "../Data.ts"
18import * as Effect from "../Effect.ts"
19import * as Layer from "../Layer.ts"
20import * as Option from "../Option.ts"
21import * as PubSub from "../PubSub.ts"
22import * as RcMap from "../RcMap.ts"
23import * as Redacted from "../Redacted.ts"
24import type * as Rpc from "../rpc/Rpc.ts"
25import type * as RpcGroup from "../rpc/RpcGroup.ts"
26import * as RpcServer from "../rpc/RpcServer.ts"
27import * as Schema from "../Schema.ts"
28import type * as Scope from "../Scope.ts"
29import * as Semaphore from "../Semaphore.ts"
30import * as Stream from "../Stream.ts"
31import type * as Event from "./Event.ts"
32import type * as EventGroup from "./EventGroup.ts"
33import * as EventJournal from "./EventJournal.ts"
34import { Entry, makeEntryIdUnsafe, makeRemoteIdUnsafe, RemoteEntry, type RemoteId } from "./EventJournal.ts"
35import * as EventLog from "./EventLog.ts"
36import {
37 ChangesRpc,
38 type EventLogAuthentication,
39 EventLogProtocolError,
40 EventLogRemoteRpcs,
41 type StoreId,
42 WriteEntriesUnencrypted
43} from "./EventLogMessage.ts"
44import * as EventLogServer from "./EventLogServer.ts"
45
46/**
47 * Service that writes plaintext event-log entries directly to
48 * unencrypted storage through registered event handlers.
49 *
50 * **When to use**
51 *
52 * Use to access or provide the server service that handles plaintext
53 * event-log writes.
54 *
55 * @stability unstable
56 * @category services
57 * @since 4.0.0
58 */
59export class EventLogServerUnencrypted extends Context.Service<EventLogServerUnencrypted, {
60 readonly makeWrite: <Groups extends EventGroup.Any>(
61 schema: EventLog.EventLogSchema<Groups>
62 ) => <
63 Tag extends EventGroup.Events<Groups>["tag"],
64 Event extends Event.Any = Event.WithTag<EventGroup.Events<Groups>, Tag>
65 >(options: {
66 readonly storeId: StoreId
67 readonly event: Tag
68 readonly payload: Event.Payload<Event>
69 }) => Effect.Effect<
70 Event.Success<Event>,
71 EventLogServerStoreError | Event.Error<Event>
72 >
73}>()("effect/eventlog/EventLogServerUnencrypted") {}
74
75/**
76 * Creates a typed server-side write function for events in the supplied
77 * `EventLogSchema`.
78 *
79 * @stability unstable
80 * @category constructors
81 * @since 4.0.0
82 */
83export const makeWrite = <Groups extends EventGroup.Any>(
84 schema: EventLog.EventLogSchema<Groups>
85): Effect.Effect<
86 <
87 Tag extends EventGroup.Events<Groups>["tag"],
88 Event extends Event.Any = Event.WithTag<EventGroup.Events<Groups>, Tag>
89 >(options: {
90 readonly storeId: StoreId
91 readonly event: Tag
92 readonly payload: Event.Payload<Event>
93 }) => Effect.Effect<
94 Event.Success<Event>,
95 EventLogServerStoreError | Event.Error<Event>
96 >,
97 never,
98 EventLogServerUnencrypted
99> => EventLogServerUnencrypted.useSync((_) => _.makeWrite(schema))
100
101/**
102 * Provides RPC handlers for the unencrypted event-log server.
103 *
104 * **Details**
105 *
106 * Incoming plaintext entries are authorized, mapped to a server store, checked
107 * for conflicts, run through registered handlers, and persisted; change streams
108 * include compacted backlog entries when compactors are registered.
109 *
110 * @stability unstable
111 * @category layers
112 * @since 4.0.0
113 */
114export const layerRpcHandlers: Layer.Layer<
115 Rpc.ToHandler<RpcGroup.Rpcs<typeof EventLogRemoteRpcs>> | EventLogAuthentication,
116 never,
117 Storage | StoreMapping | EventLogServerAuthorization | EventLog.Registry
118> = Layer.unwrap(Effect.gen(function*() {
119 const storage = yield* Storage
120 const mapping = yield* StoreMapping
121 const auth = yield* EventLogServerAuthorization
122 const registry = yield* EventLog.Registry
123 const handler = yield* makeServerHandler
124 const remoteId = yield* storage.getId
125
126 const processEntries = Effect.fnUntraced(function*(options: {
127 readonly publicKey: string
128 readonly storeId: StoreId
129 readonly entries: Arr.NonEmptyReadonlyArray<Entry>
130 }) {
131 const entries = Arr.sort(options.entries, Entry.Order)
132 let history = yield* storage.entriesAfter(options.storeId, entries[0])
133 const persistedEntries = Arr.empty<Entry>()
134 for (const entry of entries) {
135 const [duplicate, conflicts, newHistory] = toConflicts(history, entry)
136 if (duplicate) continue
137 history = newHistory
138 yield* handler({
139 publicKey: options.publicKey,
140 storeId: options.storeId,
141 entry,
142 conflicts
143 })
144 persistedEntries.push(entry)
145 }
146 yield* storage.write(options.storeId, persistedEntries)
147 }, storage.withTransaction)
148
149 return EventLogServer.layerRpcHandlers({
150 remoteId,
151 authorizeIdentity: (publicKey) =>
152 auth.authorizeIdentity({ publicKey }).pipe(
153 Effect.mapError((error) =>
154 new EventLogProtocolError({
155 requestTag: "Authenticate",
156 publicKey,
157 code: error.reason === "Forbidden" ? "Forbidden" : "Unauthorized",
158 message: error.message
159 })
160 )
161 ),
162 getOrCreateSessionAuthBinding: (publicKey, signingPublicKey) =>
163 storage.getOrCreateSessionAuthBinding(publicKey, signingPublicKey),
164 onWrite: Effect.fnUntraced(function*(data, authenticatedPublicKeys) {
165 const request = yield* WriteEntriesUnencrypted.decode(data).pipe(
166 Effect.mapError((_) =>
167 new EventLogProtocolError({
168 requestTag: "WriteEntries",
169 publicKey: undefined,
170 code: "InternalServerError",
171 message: "Decoding failure"
172 })
173 )
174 )
175 if (!authenticatedPublicKeys.has(request.publicKey)) {
176 return yield* new EventLogProtocolError({
177 requestTag: "WriteEntries",
178 publicKey: request.publicKey,
179 code: "Forbidden",
180 message: "Identity is not authenticated"
181 })
182 }
183 if (!Arr.isReadonlyArrayNonEmpty(request.entries)) return
184
185 const resolvedStoreId = yield* mapping.resolve({
186 publicKey: request.publicKey,
187 storeId: request.storeId
188 }).pipe(
189 Effect.mapError((_) =>
190 new EventLogProtocolError({
191 requestTag: "WriteEntries",
192 publicKey: request.publicKey,
193 storeId: request.storeId,
194 code: "Unauthorized",
195 message: _.message
196 })
197 )
198 )
199 yield* auth.authorizeWrite({
200 publicKey: request.publicKey,
201 storeId: resolvedStoreId,
202 entries: request.entries
203 }).pipe(
204 Effect.mapError((_) =>
205 new EventLogProtocolError({
206 requestTag: "WriteEntries",
207 publicKey: request.publicKey,
208 storeId: request.storeId,
209 code: "Unauthorized",
210 message: _.message
211 })
212 )
213 )
214 yield* processEntries({
215 publicKey: request.publicKey,
216 storeId: resolvedStoreId,
217 entries: request.entries
218 }).pipe(
219 Effect.catchCause((_) =>
220 Effect.fail(
221 new EventLogProtocolError({
222 requestTag: "WriteEntries",
223 publicKey: request.publicKey,
224 code: "InternalServerError",
225 message: "Persistence failure"
226 })
227 )
228 ),
229 Effect.provideService(EventLog.Identity, makeClientIdentity(request.publicKey))
230 )
231 }),
232 changes: Effect.fnUntraced(function*(request) {
233 const storeId = yield* mapping.resolve({
234 publicKey: request.publicKey,
235 storeId: request.storeId
236 })
237 yield* auth.authorizeRead({
238 publicKey: request.publicKey,
239 storeId
240 })
241 return storage.changes({
242 storeId,
243 startSequence: request.startSequence,
244 compactors: registry.compactors
245 }).pipe(
246 Stream.mapArrayEffect((entries) => Effect.map(ChangesRpc.encodeUnencrypted(entries), Arr.of))
247 )
248 }, Stream.unwrap)
249 })
250}))
251
252/**
253 * Error raised by unencrypted server storage and store mapping operations.
254 *
255 * @stability unstable
256 * @category errors
257 * @since 4.0.0
258 */
259export class EventLogServerStoreError extends Data.TaggedError("EventLogServerStoreError")<{
260 readonly reason: "NotFound" | "PersistenceFailure"
261 readonly publicKey?: string | undefined
262 readonly storeId?: StoreId | undefined
263 readonly message?: string | undefined
264}> {}
265
266/**
267 * Error raised when unencrypted server authorization rejects an identity or store
268 * operation.
269 *
270 * @stability unstable
271 * @category errors
272 * @since 4.0.0
273 */
274export class EventLogServerAuthError extends Data.TaggedError("EventLogServerAuthError")<{
275 readonly reason: "Unauthorized" | "Forbidden"
276 readonly publicKey: string
277 readonly storeId?: StoreId | undefined
278 readonly message?: string | undefined
279}> {}
280
281/**
282 * Service that validates unencrypted event-log server
283 * write access, read access, and identities.
284 *
285 * **When to use**
286 *
287 * Use to provide authorization checks for plaintext event-log writes, reads,
288 * and identity authentication.
289 *
290 * @stability unstable
291 * @category services
292 * @since 4.0.0
293 */
294export class EventLogServerAuthorization extends Context.Service<EventLogServerAuthorization, {
295 readonly authorizeWrite: (options: {
296 readonly publicKey: string
297 readonly storeId: StoreId
298 readonly entries: ReadonlyArray<Entry>
299 }) => Effect.Effect<void, EventLogServerAuthError>
300 readonly authorizeRead: (options: {
301 readonly publicKey: string
302 readonly storeId: StoreId
303 }) => Effect.Effect<void, EventLogServerAuthError>
304 readonly authorizeIdentity: (options: {
305 readonly publicKey: string
306 }) => Effect.Effect<void, EventLogServerAuthError>
307}>()("effect/eventlog/EventLogServerUnencrypted/EventLogServerAuthorization") {}
308
309/**
310 * Service that resolves client-requested store ids to server store ids and checks
311 * whether a store exists.
312 *
313 * **When to use**
314 *
315 * Use to map client-visible store identifiers to server storage identifiers
316 * before authorizing or serving unencrypted event-log requests.
317 *
318 * @stability unstable
319 * @category services
320 * @since 4.0.0
321 */
322export class StoreMapping extends Context.Service<StoreMapping, {
323 readonly resolve: (
324 options: {
325 readonly publicKey: string
326 readonly storeId: StoreId
327 }
328 ) => Effect.Effect<StoreId, EventLogServerStoreError>
329 readonly hasStore: (options: {
330 readonly publicKey: string
331 readonly storeId: StoreId
332 }) => Effect.Effect<boolean, EventLogServerStoreError>
333}>()("effect/eventlog/EventLogServerUnencrypted/StoreMapping") {}
334
335const toStoreNotFoundError = (options: {
336 readonly storeId: StoreId
337 readonly publicKey?: string | undefined
338}) =>
339 new EventLogServerStoreError({
340 reason: "NotFound",
341 publicKey: options.publicKey,
342 storeId: options.storeId,
343 message: options.publicKey === undefined
344 ? `No provisioned store found for store id: ${options.storeId}`
345 : `No provisioned store found for public key: ${options.publicKey} and store id: ${options.storeId}`
346 })
347
348/**
349 * Provides a `StoreMapping` that accepts only one configured store id and fails
350 * all other store ids as not found.
351 *
352 * @stability unstable
353 * @category layers
354 * @since 4.0.0
355 */
356export const layerStoreMappingStatic = (options: {
357 readonly storeId: StoreId
358}): Layer.Layer<StoreMapping> =>
359 Layer.succeed(StoreMapping, {
360 resolve(request) {
361 if (request.storeId === options.storeId) {
362 return Effect.succeed(options.storeId)
363 }
364 return Effect.fail(toStoreNotFoundError(request))
365 },
366 hasStore: ({ storeId }) => Effect.succeed(storeId === options.storeId)
367 })
368
369/**
370 * Defines the backing store service used by the unencrypted event-log server.
371 *
372 * **When to use**
373 *
374 * Use to provide durable event-log persistence for an unencrypted event-log
375 * server layer.
376 *
377 * **Details**
378 *
379 * It provides the server remote id, stores session authentication bindings,
380 * allocates remote sequence numbers, persists entries, streams changes, and
381 * exposes a transaction boundary.
382 *
383 * @stability unstable
384 * @category services
385 * @since 4.0.0
386 */
387export class Storage extends Context.Service<Storage, {
388 readonly getId: Effect.Effect<RemoteId>
389 readonly getOrCreateSessionAuthBinding: (
390 publicKey: string,
391 signingPublicKey: Uint8Array<ArrayBuffer>
392 ) => Effect.Effect<Uint8Array<ArrayBuffer>>
393 readonly entriesAfter: (storeId: StoreId, entry: Entry) => Effect.Effect<Array<Entry>>
394 readonly write: (
395 storeId: StoreId,
396 entries: ReadonlyArray<Entry>
397 ) => Effect.Effect<ReadonlyArray<RemoteEntry>>
398 readonly changes: (options: {
399 readonly storeId: StoreId
400 readonly startSequence: number
401 readonly compactors: ReadonlyMap<string, RegisteredCompactor>
402 }) => Stream.Stream<RemoteEntry>
403 readonly withTransaction: <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
404}>()("effect/eventlog/EventLogServerUnencrypted/Storage") {}
405
406const makeClientIdentity = (publicKey: string): EventLog.Identity["Service"] => ({
407 publicKey,
408 privateKey: constEmptyPrivateKey
409})
410const constEmptyPrivateKey = Redacted.make(new Uint8Array(32))
411
412const makeServerWriteIdentityPublicKey = (storeId: StoreId): string => `effect-eventlog-server-write:${storeId}`
413
414const entriesAfter = (journal: Array<RemoteEntry>, startSequence: number): ReadonlyArray<RemoteEntry> =>
415 journal.filter((entry) => entry.remoteSequence >= startSequence)
416
417const toConflicts = (
418 history: ReadonlyArray<Entry>,
419 originEntry: Entry
420): [duplicate: boolean, conflicts: Array<Entry>, newHistory: Array<Entry>] => {
421 let duplicate = false
422
423 for (let i = 0; i < history.length; i++) {
424 const entry = history[i]
425 if (entry.createdAtMillis < originEntry.createdAtMillis) {
426 continue
427 } else if (entry.idString === originEntry.idString) {
428 duplicate = true
429 continue
430 }
431
432 const newHistory = history.slice(i)
433 let conflicts: Array<Entry> = []
434 for (let j = 0; j < newHistory.length; j++) {
435 const scannedEntry = newHistory[j]!
436 if (scannedEntry.event === originEntry.event && scannedEntry.primaryKey === originEntry.primaryKey) {
437 conflicts.push(scannedEntry)
438 }
439 }
440 return [duplicate, conflicts, newHistory]
441 }
442
443 return [duplicate, [], []]
444}
445
446type RegisteredCompactor = {
447 readonly events: ReadonlySet<string>
448 readonly effect: (options: {
449 readonly entries: ReadonlyArray<Entry>
450 readonly write: (entry: Entry) => Effect.Effect<void>
451 }) => Effect.Effect<void>
452}
453
454const representativeSequences = (options: {
455 readonly remoteEntries: ReadonlyArray<RemoteEntry>
456 readonly compactedCount: number
457}): ReadonlyArray<number> | undefined => {
458 if (options.compactedCount === 0) {
459 return []
460 }
461 if (options.compactedCount > options.remoteEntries.length) {
462 return undefined
463 }
464
465 const maxSequence = options.remoteEntries[options.remoteEntries.length - 1]!.remoteSequence
466 if (options.compactedCount === 1) {
467 return [maxSequence]
468 }
469
470 const selected = options.remoteEntries
471 .slice(0, options.compactedCount - 1)
472 .map((entry) => entry.remoteSequence)
473 selected.push(maxSequence)
474 for (let i = 1; i < selected.length; i++) {
475 if (selected[i]! <= selected[i - 1]!) {
476 return undefined
477 }
478 }
479 return selected
480}
481
482const toCompactedRemoteEntries = (options: {
483 readonly compacted: ReadonlyArray<Entry>
484 readonly remoteEntries: ReadonlyArray<RemoteEntry>
485}): ReadonlyArray<RemoteEntry> | undefined => {
486 const sequences = representativeSequences({
487 remoteEntries: options.remoteEntries,
488 compactedCount: options.compacted.length
489 })
490 if (sequences === undefined) {
491 return undefined
492 }
493
494 return options.compacted.map((entry, index) =>
495 new RemoteEntry({
496 remoteSequence: sequences[index]!,
497 entry
498 }, { disableChecks: true })
499 )
500}
501
502/**
503 * Runs the registered compactors over a backlog of remote entries.
504 *
505 * **When to use**
506 *
507 * Use to reduce stored remote entries before replaying them to an unencrypted
508 * event-log client.
509 *
510 * **Details**
511 *
512 * Contiguous entries handled by the same compactor may be replaced with compacted
513 * entries when the replacement count can be mapped back to increasing remote
514 * sequence numbers; otherwise the original entries are kept.
515 *
516 * @stability unstable
517 * @category compaction
518 * @since 4.0.0
519 */
520export const compactBacklog = Effect.fnUntraced(function*(options: {
521 readonly remoteEntries: ReadonlyArray<RemoteEntry>
522 readonly compactors: ReadonlyMap<string, RegisteredCompactor>
523}) {
524 if (options.compactors.size === 0 || options.remoteEntries.length === 0) {
525 return options.remoteEntries
526 }
527
528 const compactedRemoteEntries: Array<RemoteEntry> = []
529 let index = 0
530
531 while (index < options.remoteEntries.length) {
532 const remoteEntry = options.remoteEntries[index]!
533 const compactor = options.compactors.get(remoteEntry.entry.event)
534 if (compactor === undefined) {
535 compactedRemoteEntries.push(remoteEntry)
536 index++
537 continue
538 }
539
540 const entries: Array<Entry> = [remoteEntry.entry]
541 const remoteGroup: Array<RemoteEntry> = [remoteEntry]
542 const compacted: Array<Entry> = []
543 index++
544
545 while (index < options.remoteEntries.length) {
546 const nextRemoteEntry = options.remoteEntries[index]!
547 const nextCompactor = options.compactors.get(nextRemoteEntry.entry.event)
548 if (nextCompactor !== compactor) {
549 break
550 }
551 entries.push(nextRemoteEntry.entry)
552 remoteGroup.push(nextRemoteEntry)
553 index++
554 }
555
556 yield* compactor.effect({
557 entries,
558 write(entry) {
559 return Effect.sync(() => {
560 compacted.push(entry)
561 })
562 }
563 }).pipe(Effect.orDie)
564
565 const projected = toCompactedRemoteEntries({
566 compacted,
567 remoteEntries: remoteGroup
568 })
569
570 if (projected === undefined) {
571 compactedRemoteEntries.push(...remoteGroup)
572 continue
573 }
574 compactedRemoteEntries.push(...projected)
575 }
576
577 return compactedRemoteEntries
578})
579
580/**
581 * Creates an in-memory unencrypted server `Storage`.
582 *
583 * **Details**
584 *
585 * The implementation keeps per-store journals and session authentication bindings
586 * in memory, publishes live changes, and serializes transactions with a
587 * semaphore.
588 *
589 * @stability unstable
590 * @category constructors
591 * @since 4.0.0
592 */
593export const makeStorageMemory: Effect.Effect<Storage["Service"], never, Scope.Scope> = Effect.gen(function*() {
594 const knownIds = new Map<string, Map<string, number>>()
595 const journals = new Map<string, Array<RemoteEntry>>()
596 const sessionAuthBindings = new Map<string, Uint8Array<ArrayBuffer>>()
597 const remoteId = makeRemoteIdUnsafe()
598
599 const ensureKnownIds = (storeId: StoreId): Map<string, number> => {
600 let storeKnownIds = knownIds.get(storeId)
601 if (storeKnownIds) return storeKnownIds
602 storeKnownIds = new Map()
603 knownIds.set(storeId, storeKnownIds)
604 return storeKnownIds
605 }
606
607 const ensureJournal = (storeId: StoreId): Array<RemoteEntry> => {
608 let journal = journals.get(storeId)
609 if (journal) return journal
610 journal = []
611 journals.set(storeId, journal)
612 return journal
613 }
614
615 const pubsubs = yield* RcMap.make({
616 lookup: (_storeId: string) =>
617 Effect.acquireRelease(
618 PubSub.unbounded<RemoteEntry>(),
619 PubSub.shutdown
620 ),
621 idleTimeToLive: 60000
622 })
623
624 const write = Effect.fnUntraced(function*(storeId: StoreId, entries: ReadonlyArray<Entry>) {
625 const sequenceNumbers: Array<number> = []
626 const committed: Array<RemoteEntry> = []
627 const storeKnownIds = ensureKnownIds(storeId)
628 const journal = ensureJournal(storeId)
629 let lastSequenceNumber = Arr.last(journal).pipe(
630 Option.map((entry) => entry.remoteSequence),
631 Option.getOrElse(() => 0)
632 )
633 if (entries.some((entry) => storeKnownIds.has(entry.idString))) {
634 return yield* Effect.die("Duplicate entries")
635 }
636
637 for (const entry of entries) {
638 const remoteEntry = new RemoteEntry({
639 remoteSequence: ++lastSequenceNumber,
640 entry
641 }, { disableChecks: true })
642
643 sequenceNumbers.push(remoteEntry.remoteSequence)
644 committed.push(remoteEntry)
645 journal.push(remoteEntry)
646 storeKnownIds.set(entry.idString, remoteEntry.remoteSequence)
647 }
648
649 const pubsub = yield* RcMap.get(pubsubs, storeId)
650 yield* PubSub.publishAll(pubsub, committed)
651
652 return committed
653 }, Effect.scoped)
654
655 const transactionSemaphore = yield* Semaphore.make(1)
656
657 return Storage.of({
658 getId: Effect.succeed(remoteId),
659 getOrCreateSessionAuthBinding: (publicKey, signingPublicKey) =>
660 Effect.sync(() => {
661 const existing = sessionAuthBindings.get(publicKey)
662 if (existing) return existing
663 sessionAuthBindings.set(publicKey, signingPublicKey)
664 return signingPublicKey
665 }),
666 entriesAfter: (storeId, entry) =>
667 Effect.sync(() => {
668 const journal = ensureJournal(storeId)
669 return journal.filter((e) => Entry.Order(e.entry, entry) >= 0).map((e) => e.entry)
670 }),
671 write,
672 changes: Effect.fnUntraced(function*({ storeId, startSequence, compactors }) {
673 const pubsub = yield* RcMap.get(pubsubs, storeId)
674 const subscription = yield* PubSub.subscribe(pubsub)
675
676 const backlog = yield* compactBacklog({
677 remoteEntries: entriesAfter(ensureJournal(storeId), startSequence),
678 compactors
679 })
680 const replayedUpTo = backlog.length > 0 ? backlog[backlog.length - 1].remoteSequence : startSequence - 1
681
682 return Stream.fromArray(backlog).pipe(
683 Stream.concat(
684 Stream.fromSubscription(subscription).pipe(
685 Stream.filter((entry) => entry.remoteSequence > replayedUpTo)
686 )
687 )
688 )
689 }, Stream.unwrap),
690 withTransaction: transactionSemaphore.withPermits(1)
691 })
692})
693
694/**
695 * Provides unencrypted server `Storage` using the in-memory implementation.
696 *
697 * @stability unstable
698 * @category layers
699 * @since 4.0.0
700 */
701export const layerStorageMemory: Layer.Layer<Storage> = Layer.effect(Storage)(makeStorageMemory)
702
703/**
704 * Creates the `EventLogServerUnencrypted` service from the configured storage and
705 * registered event handlers.
706 *
707 * **When to use**
708 *
709 * Use when you need the unencrypted event-log server service from provided
710 * `Storage` and an event-log `Registry`.
711 *
712 * **Details**
713 *
714 * The constructed service exposes `makeWrite`, which builds a typed server-side
715 * write function from an `EventLogSchema`. Each write encodes the payload with
716 * the event schema, runs the registered handler, and persists the generated
717 * entry inside `Storage.withTransaction`.
718 *
719 * **Gotchas**
720 *
721 * The write function dies if the requested event tag is not present in the
722 * schema passed to `makeWrite`; it does not report that case as a typed failure.
723 *
724 * @see {@link makeWrite} for the accessor that retrieves the typed server-side write function from the service environment
725 * @see {@link layerServer} for the layer form that provides this service together with an event-log `Registry`
726 *
727 * @stability unstable
728 * @category constructors
729 * @since 4.0.0
730 */
731export const make = Effect.gen(function*() {
732 const storage = yield* Storage
733 const handler = yield* makeServerHandler
734
735 return EventLogServerUnencrypted.of({
736 makeWrite<Groups extends EventGroup.Any>(schema: EventLog.EventLogSchema<Groups>) {
737 const events = new Map<string, Event.AnyWithProps>()
738 for (const group of schema.groups as unknown as ReadonlyArray<EventGroup.EventGroup<Event.Any>>) {
739 for (const [tag, event] of Object.entries(group.events)) {
740 events.set(tag, event)
741 }
742 }
743 return Effect.fnUntraced(function*(options: {
744 readonly storeId: StoreId
745 readonly event: string
746 readonly payload: unknown
747 }) {
748 const publicKey = makeServerWriteIdentityPublicKey(options.storeId)
749 const schemaEvent = events.get(options.event)
750 if (schemaEvent === undefined) {
751 return yield* Effect.die(`Event schema not found for: "${options.event}"`)
752 }
753
754 const entry = new EventJournal.Entry({
755 id: makeEntryIdUnsafe(),
756 event: options.event,
757 primaryKey: schemaEvent.primaryKey(options.payload),
758 payload: yield* Schema.encodeUnknownEffect(schemaEvent.payloadSchemaBinary)(options.payload).pipe(
759 Effect.mapError((_) =>
760 new EventLogServerStoreError({
761 reason: "PersistenceFailure",
762 publicKey: publicKey,
763 storeId: options.storeId,
764 message: "Failed to encode event"
765 })
766 )
767 ) as Effect.Effect<any, EventLogServerStoreError>
768 }, { disableChecks: true })
769
770 const result = yield* handler({
771 publicKey,
772 storeId: options.storeId,
773 entry,
774 conflicts: []
775 }).pipe(
776 Effect.provideService(EventLog.Identity, makeClientIdentity(publicKey))
777 )
778
779 yield* storage.write(options.storeId, [entry])
780
781 return result
782 }, storage.withTransaction) as any
783 }
784 })
785})
786
787/**
788 * Provides `EventLogServerUnencrypted` and an event-log `Registry` using the
789 * configured unencrypted server `Storage`.
790 *
791 * **When to use**
792 *
793 * Use to provide the unencrypted event-log server service together with the
794 * registry needed by event handlers.
795 *
796 * @stability unstable
797 * @category layers
798 * @since 4.0.0
799 */
800export const layerServer: Layer.Layer<
801 EventLogServerUnencrypted | EventLog.Registry,
802 never,
803 Storage
804> = Layer.effect(EventLogServerUnencrypted, make).pipe(
805 Layer.provideMerge(EventLog.layerRegistry)
806)
807
808/**
809 * Builds a full unencrypted event-log RPC server for the supplied schema and
810 * event-group handler layer.
811 *
812 * **When to use**
813 *
814 * Use when you need the full unencrypted event-log RPC server layer with
815 * storage, authorization, RPC protocol, and event-group handler dependencies
816 * supplied externally.
817 *
818 * **Details**
819 *
820 * The layer installs `EventLogRemoteRpcs`, wires `layerRpcHandlers`, registers
821 * the supplied event-group handler layer, and provides `layerServer`, leaving
822 * only the required infrastructure services in the environment.
823 *
824 * **Gotchas**
825 *
826 * Entries are persisted and streamed in plaintext. Protect the backing
827 * `Storage` with the surrounding infrastructure, and use durable storage that
828 * preserves session authentication bindings when the server must survive
829 * restarts.
830 *
831 * @see {@link layerNoRpcServer} for installing the same unencrypted handlers when an `RpcServer.Protocol` is provided elsewhere
832 * @see {@link layerRpcHandlers} for wiring the unencrypted RPC handlers directly
833 * @see {@link layerServer} for constructing the server service and event-log registry without RPC handlers
834 *
835 * @stability unstable
836 * @category layers
837 * @since 4.0.0
838 */
839export const layer = <Groups extends EventGroup.Any, E, R>(
840 _schema: EventLog.EventLogSchema<Groups>,
841 layer: Layer.Layer<EventGroup.ToService<Groups>, E, R>
842): Layer.Layer<
843 never,
844 E,
845 | Exclude<R, EventLogServerUnencrypted | EventLog.Registry>
846 | EventLogServerAuthorization
847 | RpcServer.Protocol
848 | Storage
849 | StoreMapping
850> =>
851 RpcServer.layer(EventLogRemoteRpcs).pipe(
852 Layer.provide(layerRpcHandlers),
853 Layer.provide(layer),
854 Layer.provide(layerServer)
855 )
856
857/**
858 * Builds the unencrypted event-log server handlers without installing an
859 * `RpcServer.Protocol` implementation.
860 *
861 * @stability unstable
862 * @category layers
863 * @since 4.0.0
864 */
865export const layerNoRpcServer = <Groups extends EventGroup.Any, E, R>(
866 _schema: EventLog.EventLogSchema<Groups>,
867 layer: Layer.Layer<EventGroup.ToService<Groups>, E, R>
868): Layer.Layer<
869 Rpc.ToHandler<RpcGroup.Rpcs<typeof EventLogRemoteRpcs>> | EventLogAuthentication,
870 E,
871 | Exclude<R, EventLogServerUnencrypted | EventLog.Registry>
872 | EventLogServerAuthorization
873 | Storage
874 | StoreMapping
875> =>
876 layerRpcHandlers.pipe(
877 Layer.merge(layer),
878 Layer.provide(layerServer)
879 )
880
881const makeServerHandler = Effect.gen(function*() {
882 const registry = yield* EventLog.Registry
883
884 return Effect.fnUntraced(
885 function*(options: {
886 readonly publicKey: string
887 readonly storeId: StoreId
888 readonly entry: Entry
889 readonly conflicts: ReadonlyArray<Entry>
890 readonly payload?: unknown
891 }): Effect.fn.Return<any, any, EventLog.Identity> {
892 const handler = registry.handlers.get(options.entry.event)
893 if (handler === undefined) {
894 return yield* Effect.logDebug(`Event handler not found for: "${options.entry.event}"`)
895 }
896
897 const decodePayload = Schema.decodeUnknownEffect(handler.event.payloadSchemaBinary)
898 const decodedConflicts: Array<{ readonly entry: Entry; readonly payload: unknown }> = []
899
900 for (const conflict of options.conflicts) {
901 decodedConflicts.push({
902 entry: conflict,
903 payload: yield* decodePayload(conflict.payload).pipe(
904 Effect.updateContext((input) => Context.merge(handler.context, input))
905 ) as any
906 })
907 }
908
909 const payloadEffect = "payload" in options
910 ? Effect.succeed(options.payload)
911 : decodePayload(options.entry.payload)
912
913 return yield* payloadEffect.pipe(
914 Effect.mapError((_) =>
915 new EventLogServerStoreError({
916 reason: "PersistenceFailure",
917 publicKey: options.publicKey,
918 storeId: options.storeId,
919 message: "Failed to decode event"
920 })
921 ),
922 Effect.flatMap((payload) =>
923 handler.handler({
924 storeId: options.storeId,
925 payload,
926 entry: options.entry,
927 conflicts: decodedConflicts
928 })
929 ),
930 Effect.updateContext((input) => Context.merge(handler.context, input))
931 ) as Effect.Effect<any, unknown, EventLog.Identity>
932 }
933 )
934})