1/**
2 * Connects a local event log to a remote replica.
3 *
4 * `EventLogRemote` writes local entries to another journal, receives remote
5 * change streams from a sequence number, and can wait until the current
6 * event-log identity has completed remote authentication. The encrypted
7 * constructor is the default for browser, edge, or service replicas crossing an
8 * untrusted network. The unencrypted constructor is intended for trusted
9 * transports or tests.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Cache from "../Cache.ts"
15import * as Context from "../Context.ts"
16import * as Data from "../Data.ts"
17import * as Effect from "../Effect.ts"
18import * as Layer from "../Layer.ts"
19import * as Predicate from "../Predicate.ts"
20import * as Queue from "../Queue.ts"
21import * as Redacted from "../Redacted.ts"
22import * as RpcClient from "../rpc/RpcClient.ts"
23import type { RpcClientError } from "../rpc/RpcClientError.ts"
24import type * as RpcGroup from "../rpc/RpcGroup.ts"
25import type * as Schema from "../Schema.ts"
26import type * as Scope from "../Scope.ts"
27import type { Entry, RemoteEntry, RemoteId } from "./EventJournal.ts"
28import { type Identity, Registry } from "./EventLog.ts"
29import { EventLogEncryption, layerSubtle } from "./EventLogEncryption.ts"
30import {
31 Authenticate,
32 ChangesRpc,
33 ChunkedMessage,
34 EventLogProtocolError,
35 EventLogRemoteRpcs,
36 type HelloResponse,
37 type StoreId,
38 WriteEntries,
39 WriteEntriesUnencrypted
40} from "./EventLogMessage.ts"
41import { encodeSessionAuthPayload, signSessionAuthPayloadBytes } from "./EventLogSessionAuth.ts"
42import { makeGetIdentityRootSecretMaterial } from "./internal/identityRootSecretDerivation.ts"
43
44/**
45 * Service that represents a remote event-log replica.
46 *
47 * **When to use**
48 *
49 * Use to access or provide a remote event-log replica that can write local
50 * entries and stream remote changes.
51 *
52 * **Details**
53 *
54 * It can write local entries to the remote, stream remote changes from a sequence
55 * number, and run effects only after the supplied identity has authenticated.
56 *
57 * @stability unstable
58 * @category services
59 * @since 4.0.0
60 */
61export class EventLogRemote extends Context.Service<EventLogRemote, {
62 readonly id: RemoteId
63 readonly changes: (options: {
64 readonly identity: Identity["Service"]
65 readonly storeId: StoreId
66 readonly startSequence: number
67 }) => Effect.Effect<Queue.Dequeue<RemoteEntry, EventLogRemoteError>, never, Scope.Scope>
68 readonly write: (options: {
69 readonly identity: Identity["Service"]
70 readonly storeId: StoreId
71 readonly entries: ReadonlyArray<Entry>
72 }) => Effect.Effect<void, EventLogRemoteError>
73 readonly whenAuthenticated: <A, E, R>(
74 effect: Effect.Effect<A, E, R>
75 ) => Effect.Effect<A, E | EventLogRemoteError, R | Identity>
76}>()("effect/eventlog/EventLogRemote") {}
77
78/**
79 * Error raised by `EventLogRemote` operations, recording the failed method and
80 * underlying cause.
81 *
82 * @stability unstable
83 * @category errors
84 * @since 4.0.0
85 */
86export class EventLogRemoteError extends Data.TaggedError("EventLogRemoteError")<{
87 readonly method: string
88 readonly cause: unknown
89}> {
90 /**
91 * Returns `true` when the value is an `EventLogRemoteError`.
92 *
93 * @stability unstable
94 * @since 4.0.0
95 */
96 static is(u: unknown): u is EventLogRemoteError {
97 return Predicate.isTagged(u, "EventLogRemoteError")
98 }
99}
100
101const getIdentityRootSecretMaterial = makeGetIdentityRootSecretMaterial(globalThis.crypto)
102
103const makeAuthenticate = Effect.fnUntraced(function*(options: {
104 readonly identity: Identity["Service"]
105 readonly hello: HelloResponse
106}) {
107 const rootSecretMaterial = yield* getIdentityRootSecretMaterial(options.identity)
108 const payload = yield* encodeSessionAuthPayload({
109 remoteId: options.hello.remoteId,
110 challenge: options.hello.challenge,
111 publicKey: options.identity.publicKey,
112 signingPublicKey: rootSecretMaterial.signingPublicKey
113 })
114 const signature = yield* signSessionAuthPayloadBytes({
115 payload,
116 signingPrivateKey: Redacted.value(rootSecretMaterial.signingPrivateKey)
117 })
118
119 return new Authenticate({
120 publicKey: options.identity.publicKey,
121 signingPublicKey: rootSecretMaterial.signingPublicKey,
122 signature,
123 algorithm: "Ed25519"
124 })
125})
126
127/**
128 * Service that provides a typed RPC client for the `EventLogRemoteRpcs` protocol.
129 *
130 * **When to use**
131 *
132 * Use to provide the RPC client used by remote event-log replicas to
133 * authenticate, write entries, and subscribe to changes.
134 *
135 * @stability unstable
136 * @category services
137 * @since 4.0.0
138 */
139export class EventLogRemoteClient extends Context.Service<
140 EventLogRemoteClient,
141 RpcClient.RpcClient<RpcGroup.Rpcs<typeof EventLogRemoteRpcs>, RpcClientError>
142>()(
143 "effect/eventlog/EventLogRemote/EventLogRemoteClient"
144) {
145 static readonly layer = Layer.effect(
146 EventLogRemoteClient,
147 RpcClient.make(EventLogRemoteRpcs, {
148 disableTracing: true
149 })
150 )
151}
152
153/**
154 * Creates an `EventLogRemote` from custom write encoding and change decoding
155 * functions.
156 *
157 * **Details**
158 *
159 * The remote performs the hello/authentication handshake, retries after forbidden
160 * responses by re-authenticating, chunks large writes, and registers itself with
161 * the `Registry` for the current scope.
162 *
163 * @stability unstable
164 * @category constructors
165 * @since 4.0.0
166 */
167export const makeWith = Effect.fnUntraced(function*({ encodeWrite, decodeChanges }: {
168 readonly encodeWrite: (options: {
169 readonly identity: Identity["Service"]
170 readonly entries: ReadonlyArray<Entry>
171 readonly storeId: StoreId
172 }) => Effect.Effect<Uint8Array<ArrayBuffer>, Schema.SchemaError>
173 readonly decodeChanges: (
174 identity: Identity["Service"],
175 data: Uint8Array<ArrayBuffer>
176 ) => Effect.Effect<ReadonlyArray<RemoteEntry>, Schema.SchemaError>
177}): Effect.fn.Return<EventLogRemote["Service"], EventLogRemoteError, Scope.Scope | EventLogRemoteClient | Registry> {
178 const client = yield* EventLogRemoteClient
179 const registry = yield* Registry
180
181 let hello: HelloResponse | null = yield* client["EventLog.Hello"]().pipe(
182 Effect.mapError((cause) => new EventLogRemoteError({ method: "hello", cause }))
183 )
184
185 const identities = new Map<string, Identity["Service"]>()
186 const ensureIdentity = (identity: Identity["Service"]) => {
187 let entry = identities.get(identity.publicKey)
188 if (!entry) {
189 entry = identity
190 identities.set(identity.publicKey, entry)
191 }
192 return entry
193 }
194
195 const authCache = yield* Cache.make({
196 lookup: Effect.fnUntraced(function*(publicKey: string) {
197 const identity = identities.get(publicKey)!
198 hello ??= yield* client["EventLog.Hello"]().pipe(
199 Effect.mapError((cause) => new EventLogRemoteError({ method: "hello", cause }))
200 )
201 const authenticate = yield* makeAuthenticate({
202 identity,
203 hello
204 })
205 yield* client["EventLog.Authenticate"](authenticate)
206 }, Effect.mapError((cause) => new EventLogRemoteError({ method: "authenticate", cause }))),
207 capacity: Number.MAX_SAFE_INTEGER
208 })
209
210 const ensureAuthenticated = (identity: Identity["Service"]) => {
211 ensureIdentity(identity)
212 return Cache.get(authCache, identity.publicKey)
213 }
214
215 const retryForbidden = <A, E, R>(
216 effect: Effect.Effect<A, E, R>,
217 options: {
218 readonly identity: Identity["Service"]
219 }
220 ) =>
221 Effect.retry(effect, {
222 while(e) {
223 hello = null
224 const error = EventLogRemoteError.is(e) && e.method === "authenticate"
225 ? e.cause
226 : e
227 const isForbidden = EventLogProtocolError.is(error) && error.code === "Forbidden"
228 return Cache.invalidate(authCache, options.identity.publicKey).pipe(
229 Effect.as(isForbidden)
230 )
231 },
232 times: 5
233 })
234
235 let chunkedIdCounter = 0
236
237 const remote = EventLogRemote.of({
238 id: hello.remoteId,
239 write: Effect.fnUntraced(
240 function*(options) {
241 yield* ensureAuthenticated(options.identity)
242 const encoded = yield* encodeWrite(options)
243 if (encoded.byteLength <= ChunkedMessage.chunkSize) {
244 return yield* client["EventLog.WriteSingle"]({ data: encoded })
245 }
246 for (const part of ChunkedMessage.split(chunkedIdCounter++, encoded)) {
247 yield* client["EventLog.WriteChunked"](part)
248 }
249 },
250 retryForbidden,
251 Effect.mapError((cause) => new EventLogRemoteError({ method: "write", cause }))
252 ),
253 changes: Effect.fnUntraced(function*(options) {
254 const outgoing = yield* Queue.make<RemoteEntry, EventLogRemoteError>()
255
256 yield* Effect.gen(function*() {
257 yield* ensureAuthenticated(options.identity)
258
259 const chunkedState = ChunkedMessage.initialJoinState()
260 const incoming = yield* client["EventLog.Changes"]({
261 publicKey: options.identity.publicKey,
262 storeId: options.storeId,
263 startSequence: options.startSequence
264 }, { asQueue: true })
265
266 while (true) {
267 const parts = yield* Queue.takeAll(incoming)
268 for (let i = 0; i < parts.length; i++) {
269 const part = parts[i]
270 if (part._tag === "Single") {
271 yield* Queue.offerAll(outgoing, yield* decodeChanges(options.identity, part.data))
272 continue
273 }
274 const data = ChunkedMessage.join(chunkedState, part)
275 if (!data) continue
276 yield* Queue.offerAll(outgoing, yield* decodeChanges(options.identity, data))
277 }
278 }
279 }).pipe(
280 (effect) => retryForbidden(effect, options),
281 Effect.mapError((cause) => {
282 if (cause._tag === "EventLogRemoteError") {
283 return cause
284 }
285 return new EventLogRemoteError({
286 method: "changes",
287 cause
288 })
289 }),
290 Effect.catchCause((cause) => Queue.failCause(outgoing, cause)),
291 Effect.forkScoped
292 )
293
294 return outgoing
295 }),
296 whenAuthenticated: (effect) =>
297 IdentityService.use((identity) => Effect.flatMap(ensureAuthenticated(identity), () => effect))
298 })
299
300 yield* registry.registerRemote(remote)
301
302 return remote
303})
304
305/** @effect-diagnostics-next-line classSelfMismatch:off */
306class IdentityService extends Context.Service<Identity, Identity["Service"]>()(
307 "effect/eventlog/EventLog/Identity" satisfies Identity["key"]
308) {}
309
310/**
311 * Creates an `EventLogRemote` that encrypts outgoing entries and decrypts
312 * incoming changes with `EventLogEncryption`.
313 *
314 * @stability unstable
315 * @category constructors
316 * @since 4.0.0
317 */
318export const makeEncrypted = Effect.gen(function*(): Effect.fn.Return<
319 EventLogRemote["Service"],
320 EventLogRemoteError,
321 Scope.Scope | EventLogRemoteClient | EventLogEncryption | Registry
322> {
323 const encryption = yield* EventLogEncryption
324
325 return yield* makeWith({
326 encodeWrite: (options) =>
327 encryption.encrypt(options.identity, options.entries).pipe(
328 Effect.flatMap((encryptedEntries) =>
329 new WriteEntries({
330 publicKey: options.identity.publicKey,
331 storeId: options.storeId,
332 encryptedEntries: encryptedEntries.map((entry, i) => ({
333 entryId: options.entries[i].id,
334 iv: entry.iv,
335 encryptedEntry: entry.encryptedEntry
336 }))
337 }).encoded
338 )
339 ),
340 decodeChanges: (identity, data) =>
341 ChangesRpc.decodeEncrypted(data).pipe(
342 Effect.flatMap((entries) => encryption.decrypt(identity, entries))
343 )
344 })
345})
346
347/**
348 * Creates an `EventLogRemote` that sends and receives plaintext entry payloads.
349 *
350 * @stability unstable
351 * @category constructors
352 * @since 4.0.0
353 */
354export const makeUnencrypted: Effect.Effect<
355 EventLogRemote["Service"],
356 EventLogRemoteError,
357 Scope.Scope | EventLogRemoteClient | Registry
358> = makeWith({
359 encodeWrite: (options) =>
360 new WriteEntriesUnencrypted({
361 publicKey: options.identity.publicKey,
362 storeId: options.storeId,
363 entries: options.entries
364 }).encoded,
365 decodeChanges: (_identity, data) => ChangesRpc.decodeUnencrypted(data)
366})
367
368/**
369 * Provides an encrypted `EventLogRemote` using the remote RPC client and the
370 * default Web Crypto encryption layer.
371 *
372 * @stability unstable
373 * @category layers
374 * @since 4.0.0
375 */
376export const layerEncrypted: Layer.Layer<
377 EventLogRemote,
378 EventLogRemoteError,
379 RpcClient.Protocol | Registry
380> = Layer.effect(EventLogRemote, makeEncrypted).pipe(
381 Layer.provide(EventLogRemoteClient.layer),
382 Layer.provide(layerSubtle)
383)
384
385/**
386 * Provides an unencrypted `EventLogRemote` using the remote RPC client.
387 *
388 * @stability unstable
389 * @category layers
390 * @since 4.0.0
391 */
392export const layerUnencrypted: Layer.Layer<
393 EventLogRemote,
394 EventLogRemoteError,
395 RpcClient.Protocol | Registry
396> = Layer.effect(EventLogRemote, makeUnencrypted).pipe(
397 Layer.provide(EventLogRemoteClient.layer)
398)