1/**
2 * Serves encrypted event-log replication.
3 *
4 * Encrypted `EventLogRemote` clients use this module when they need a remote
5 * synchronization endpoint that never sees plaintext events. The server stores
6 * encrypted entries and replication metadata keyed by client public key and
7 * store id, then streams encrypted changes back to clients for local
8 * decryption. This module defines the RPC handlers, server layer, storage
9 * contract, and in-memory storage layer for that encrypted server path.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Arr from "../Array.ts"
15import * as Context from "../Context.ts"
16import * as Effect from "../Effect.ts"
17import * as Uuid from "../internal/uuid.ts"
18import * as Layer from "../Layer.ts"
19import * as PubSub from "../PubSub.ts"
20import * as RcMap from "../RcMap.ts"
21import * as RpcServer from "../rpc/RpcServer.ts"
22import * as Schema from "../Schema.ts"
23import type * as Scope from "../Scope.ts"
24import * as Stream from "../Stream.ts"
25import * as Transferable from "../workers/Transferable.ts"
26import { EntryId, makeRemoteIdUnsafe, type RemoteId } from "./EventJournal.ts"
27import type { EncryptedRemoteEntry } from "./EventLogEncryption.ts"
28import { ChangesRpc, EventLogProtocolError, EventLogRemoteRpcs, type StoreId, WriteEntries } from "./EventLogMessage.ts"
29import * as EventLogServer from "./EventLogServer.ts"
30
31/**
32 * Provides RPC handlers for the encrypted event-log server.
33 *
34 * **Details**
35 *
36 * Incoming encrypted write payloads are decoded and persisted through `Storage`;
37 * change streams read encrypted entries from storage and encode them for the
38 * remote protocol.
39 *
40 * @stability unstable
41 * @category layers
42 * @since 4.0.0
43 */
44export const layerRpcHandlers = Layer.unwrap(Effect.gen(function*() {
45 const storage = yield* Storage
46 const remoteId = yield* storage.getId
47
48 return EventLogServer.layerRpcHandlers({
49 remoteId,
50 getOrCreateSessionAuthBinding: (publicKey, signingPublicKey) =>
51 storage.getOrCreateSessionAuthBinding(publicKey, signingPublicKey),
52 onWrite: Effect.fnUntraced(function*(data, authenticatedPublicKeys) {
53 const request = yield* WriteEntries.decode(data).pipe(
54 Effect.mapError((_) =>
55 new EventLogProtocolError({
56 requestTag: "WriteEntries",
57 publicKey: undefined,
58 code: "InternalServerError",
59 message: "Decoding failure"
60 })
61 )
62 )
63 if (!authenticatedPublicKeys.has(request.publicKey)) {
64 return yield* new EventLogProtocolError({
65 requestTag: "WriteEntries",
66 publicKey: request.publicKey,
67 code: "Forbidden",
68 message: "Identity is not authenticated"
69 })
70 }
71 if (request.encryptedEntries.length === 0) return
72 const entries = request.encryptedEntries.map(({ encryptedEntry, entryId, iv }) =>
73 new PersistedEntry({
74 entryId,
75 iv,
76 encryptedEntry
77 })
78 )
79 return yield* storage.write(request.publicKey, request.storeId, entries).pipe(
80 Effect.catchCause((_) =>
81 Effect.fail(
82 new EventLogProtocolError({
83 requestTag: "WriteEntries",
84 publicKey: request.publicKey,
85 code: "InternalServerError",
86 message: "Persistence failure"
87 })
88 )
89 )
90 )
91 }),
92 changes: ({ publicKey, storeId, startSequence }) =>
93 storage.changes(publicKey, storeId, startSequence).pipe(
94 Stream.mapArrayEffect((entries) => Effect.map(ChangesRpc.encodeEncrypted(entries), Arr.of))
95 )
96 })
97}))
98
99/**
100 * Provides an encrypted event-log RPC server using `EventLogRemoteRpcs` and the
101 * encrypted server RPC handlers.
102 *
103 * **When to use**
104 *
105 * Use when you need an encrypted event-log RPC server for encrypted
106 * `EventLogRemote` replication over an existing `RpcServer.Protocol`.
107 *
108 * **Details**
109 *
110 * This layer installs `EventLogRemoteRpcs` on the provided RPC server protocol
111 * and wires those RPCs to `layerRpcHandlers`. Encrypted entries, session
112 * authentication bindings, remote ids, and change streams are delegated to
113 * `Storage`.
114 *
115 * @see {@link layerRpcHandlers} for the encrypted handler layer without installing an RPC server protocol
116 * @see {@link Storage} for the storage service required by this layer
117 * @see {@link layerStorageMemory} for the process-local in-memory storage layer
118 *
119 * @stability unstable
120 * @category layers
121 * @since 4.0.0
122 */
123export const layer: Layer.Layer<never, never, RpcServer.Protocol | Storage> = RpcServer.layer(EventLogRemoteRpcs).pipe(
124 Layer.provide(layerRpcHandlers)
125)
126
127/**
128 * Schema for encrypted entries persisted by the encrypted event-log server.
129 *
130 * @stability unstable
131 * @category models
132 * @since 4.0.0
133 */
134export class PersistedEntry extends Schema.Class<PersistedEntry>(
135 "effect/eventlog/EventLogServerEncrypted/PersistedEntry"
136)({
137 entryId: EntryId,
138 iv: Transferable.Uint8Array,
139 encryptedEntry: Transferable.Uint8Array
140}) {
141 /**
142 * String representation of the encrypted entry id.
143 *
144 * @stability unstable
145 * @since 4.0.0
146 */
147 get entryIdString(): string {
148 return Uuid.stringify(this.entryId)
149 }
150}
151
152/**
153 * Defines the backing store service used by the encrypted event-log server.
154 *
155 * **When to use**
156 *
157 * Use to provide durable encrypted event-log persistence for an encrypted
158 * event-log server layer.
159 *
160 * **Details**
161 *
162 * It provides the server remote id, stores session authentication bindings,
163 * persists encrypted entries, and streams encrypted changes for a public key and
164 * store id.
165 *
166 * @stability unstable
167 * @category services
168 * @since 4.0.0
169 */
170export class Storage extends Context.Service<Storage, {
171 readonly getId: Effect.Effect<RemoteId>
172 readonly getOrCreateSessionAuthBinding: (
173 publicKey: string,
174 signingPublicKey: Uint8Array<ArrayBuffer>
175 ) => Effect.Effect<Uint8Array<ArrayBuffer>>
176 readonly write: (
177 publicKey: string,
178 storeId: StoreId,
179 entries: ReadonlyArray<PersistedEntry>
180 ) => Effect.Effect<ReadonlyArray<EncryptedRemoteEntry>>
181 readonly changes: (
182 publicKey: string,
183 storeId: StoreId,
184 startSequence: number
185 ) => Stream.Stream<EncryptedRemoteEntry>
186}>()("effect/eventlog/EventLogServer/Storage") {}
187
188/**
189 * Creates an in-memory encrypted server `Storage`.
190 *
191 * **Details**
192 *
193 * Data, session authentication bindings, and streams are process-local and are
194 * released with the surrounding scope.
195 *
196 * @stability unstable
197 * @category constructors
198 * @since 4.0.0
199 */
200export const makeStorageMemory: Effect.Effect<Storage["Service"], never, Scope.Scope> = Effect.gen(function*() {
201 const knownIds = new Map<string, Map<string, number>>()
202 const journals = new Map<string, Array<EncryptedRemoteEntry>>()
203 const sessionAuthBindings = new Map<string, Uint8Array<ArrayBuffer>>()
204 const remoteId = makeRemoteIdUnsafe()
205 const ensureKnownIds = (scopeKey: string): Map<string, number> => {
206 let storeKnownIds = knownIds.get(scopeKey)
207 if (storeKnownIds) return storeKnownIds
208 storeKnownIds = new Map<string, number>()
209 knownIds.set(scopeKey, storeKnownIds)
210 return storeKnownIds
211 }
212 const ensureJournal = (scopeKey: string) => {
213 let journal = journals.get(scopeKey)
214 if (journal) return journal
215 journal = []
216 journals.set(scopeKey, journal)
217 return journal
218 }
219 const pubsubs = yield* RcMap.make({
220 lookup: (_scopeKey: string) =>
221 Effect.acquireRelease(
222 PubSub.unbounded<EncryptedRemoteEntry>(),
223 PubSub.shutdown
224 ),
225 idleTimeToLive: 60000
226 })
227
228 return Storage.of({
229 getId: Effect.succeed(remoteId),
230 getOrCreateSessionAuthBinding: (publicKey, signingPublicKey) =>
231 Effect.sync(() => {
232 let existing = sessionAuthBindings.get(publicKey)
233 if (existing) return existing
234 sessionAuthBindings.set(publicKey, signingPublicKey)
235 return signingPublicKey
236 }),
237 write: Effect.fnUntraced(function*(publicKey, storeId, entries) {
238 const scopeKey = makeEncryptedScopeKey({ publicKey, storeId })
239 const pubsub = yield* RcMap.get(pubsubs, scopeKey)
240 const storeKnownIds = ensureKnownIds(scopeKey)
241 const journal = ensureJournal(scopeKey)
242 const encryptedEntries: Array<EncryptedRemoteEntry> = []
243 for (const entry of entries) {
244 const idString = entry.entryIdString
245 if (storeKnownIds.has(idString)) continue
246 const encrypted: EncryptedRemoteEntry = {
247 sequence: journal.length,
248 entryId: entry.entryId,
249 iv: entry.iv,
250 encryptedEntry: entry.encryptedEntry
251 }
252 encryptedEntries.push(encrypted)
253 storeKnownIds.set(idString, encrypted.sequence)
254 journal.push(encrypted)
255 PubSub.publishUnsafe(pubsub, encrypted)
256 }
257 return encryptedEntries
258 }, Effect.scoped),
259 changes: Effect.fnUntraced(function*(publicKey, storeId, startSequence) {
260 const scopeKey = makeEncryptedScopeKey({ publicKey, storeId })
261 const pubsub = yield* RcMap.get(pubsubs, scopeKey)
262 const subscription = yield* PubSub.subscribe(pubsub)
263 return Stream.fromArray(ensureJournal(scopeKey).slice(startSequence)).pipe(
264 Stream.concat(Stream.fromSubscription(subscription))
265 )
266 }, Stream.unwrap)
267 })
268})
269
270/**
271 * Provides encrypted server `Storage` using the in-memory implementation.
272 *
273 * @stability unstable
274 * @category layers
275 * @since 4.0.0
276 */
277export const layerStorageMemory: Layer.Layer<Storage> = Layer.effect(Storage)(makeStorageMemory)
278
279const makeEncryptedScopeKey = ({ publicKey, storeId }: {
280 readonly publicKey: string
281 readonly storeId: StoreId
282}): string => {
283 return `${publicKey}/${storeId}`
284}