1/**
2 * SQL-backed storage for unencrypted event-log servers.
3 *
4 * This module provides the durable `Storage` implementation used by
5 * `EventLogServerUnencrypted` when remote entries should be stored in a SQL
6 * database and streamed back to clients by store sequence. It creates
7 * dialect-specific tables for the server remote id, per-store sequence state,
8 * plaintext entries, and session authentication bindings.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Arr from "../Array.ts"
14import * as Effect from "../Effect.ts"
15import * as Layer from "../Layer.ts"
16import * as PubSub from "../PubSub.ts"
17import * as RcMap from "../RcMap.ts"
18import * as Schema from "../Schema.ts"
19import * as SchemaTransformation from "../SchemaTransformation.ts"
20import type * as Scope from "../Scope.ts"
21import * as SqlClient from "../sql/SqlClient.ts"
22import * as SqlError from "../sql/SqlError.ts"
23import * as Stream from "../Stream.ts"
24import { Entry, EntryId, makeRemoteIdUnsafe, RemoteEntry, type RemoteId } from "./EventJournal.ts"
25import * as EventLogServerUnencrypted from "./EventLogServerUnencrypted.ts"
26
27/**
28 * Creates unencrypted event-log server `Storage` backed by SQL.
29 *
30 * **Details**
31 *
32 * The implementation creates tables for the server remote id, store sequences,
33 * entries, and session authentication bindings, then persists and streams
34 * plaintext remote entries.
35 *
36 * @stability unstable
37 * @category constructors
38 * @since 4.0.0
39 */
40export const makeStorage = (options?: {
41 readonly entryTablePrefix?: string
42 readonly remoteIdTable?: string
43 readonly insertBatchSize?: number
44}): Effect.Effect<
45 EventLogServerUnencrypted.Storage["Service"],
46 SqlError.SqlError,
47 SqlClient.SqlClient | Scope.Scope
48> =>
49 Effect.gen(function*() {
50 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
51
52 const entriesTable = options?.entryTablePrefix ?? "effect_events"
53 const remoteIdTable = options?.remoteIdTable ?? "effect_remote_id"
54 const insertBatchSize = options?.insertBatchSize ?? 200
55 const storesTable = `${entriesTable}_stores`
56 const sessionAuthBindingsTable = `${entriesTable}_session_auth_bindings`
57
58 const remoteIdTableSql = sql(remoteIdTable)
59 const entriesTableSql = sql(entriesTable)
60 const storesTableSql = sql(storesTable)
61 const sessionAuthBindingsTableSql = sql(sessionAuthBindingsTable)
62
63 yield* sql.onDialectOrElse({
64 pg: () =>
65 sql`
66 CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
67 singleton INT PRIMARY KEY,
68 remote_id BYTEA NOT NULL
69 )`,
70 mysql: () =>
71 sql`
72 CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
73 singleton INT PRIMARY KEY,
74 remote_id BINARY(16) NOT NULL
75 )`,
76 mssql: () =>
77 sql`
78 CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
79 singleton INT PRIMARY KEY,
80 remote_id VARBINARY(16) NOT NULL
81 )`,
82 orElse: () =>
83 sql`
84 CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
85 singleton INTEGER PRIMARY KEY,
86 remote_id BLOB NOT NULL
87 )`
88 })
89
90 yield* sql.onDialectOrElse({
91 pg: () =>
92 sql`
93 CREATE TABLE IF NOT EXISTS ${entriesTableSql} (
94 store_id TEXT NOT NULL,
95 sequence BIGINT NOT NULL,
96 entry_id BYTEA NOT NULL,
97 event TEXT NOT NULL,
98 primary_key TEXT NOT NULL,
99 payload BYTEA NOT NULL,
100 PRIMARY KEY (store_id, sequence),
101 UNIQUE (store_id, entry_id)
102 )`,
103 mysql: () =>
104 sql`
105 CREATE TABLE IF NOT EXISTS ${entriesTableSql} (
106 store_id VARCHAR(191) NOT NULL,
107 sequence BIGINT NOT NULL,
108 entry_id BINARY(16) NOT NULL,
109 event TEXT NOT NULL,
110 primary_key TEXT NOT NULL,
111 payload BLOB NOT NULL,
112 PRIMARY KEY (store_id, sequence),
113 UNIQUE (store_id, entry_id)
114 )`,
115 mssql: () =>
116 sql`
117 CREATE TABLE IF NOT EXISTS ${entriesTableSql} (
118 store_id NVARCHAR(191) NOT NULL,
119 sequence BIGINT NOT NULL,
120 entry_id VARBINARY(16) NOT NULL,
121 event NVARCHAR(MAX) NOT NULL,
122 primary_key NVARCHAR(MAX) NOT NULL,
123 payload VARBINARY(MAX) NOT NULL,
124 PRIMARY KEY (store_id, sequence),
125 UNIQUE (store_id, entry_id)
126 )`,
127 orElse: () =>
128 sql`
129 CREATE TABLE IF NOT EXISTS ${entriesTableSql} (
130 store_id TEXT NOT NULL,
131 sequence INTEGER NOT NULL,
132 entry_id BLOB NOT NULL,
133 event TEXT NOT NULL,
134 primary_key TEXT NOT NULL,
135 payload BLOB NOT NULL,
136 PRIMARY KEY (store_id, sequence),
137 UNIQUE (store_id, entry_id)
138 )`
139 })
140
141 yield* sql.onDialectOrElse({
142 pg: () =>
143 sql`
144 CREATE TABLE IF NOT EXISTS ${storesTableSql} (
145 store_id TEXT PRIMARY KEY,
146 next_sequence BIGINT NOT NULL
147 )`,
148 mysql: () =>
149 sql`
150 CREATE TABLE IF NOT EXISTS ${storesTableSql} (
151 store_id VARCHAR(191) PRIMARY KEY,
152 next_sequence BIGINT NOT NULL
153 )`,
154 mssql: () =>
155 sql`
156 CREATE TABLE IF NOT EXISTS ${storesTableSql} (
157 store_id NVARCHAR(191) PRIMARY KEY,
158 next_sequence BIGINT NOT NULL
159 )`,
160 orElse: () =>
161 sql`
162 CREATE TABLE IF NOT EXISTS ${storesTableSql} (
163 store_id TEXT PRIMARY KEY,
164 next_sequence INTEGER NOT NULL
165 )`
166 })
167
168 yield* sql.onDialectOrElse({
169 pg: () =>
170 sql`
171 CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
172 public_key TEXT PRIMARY KEY,
173 signing_public_key BYTEA NOT NULL
174 )`,
175 mysql: () =>
176 sql`
177 CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
178 public_key VARCHAR(191) PRIMARY KEY,
179 signing_public_key BINARY(32) NOT NULL
180 )`,
181 mssql: () =>
182 sql`
183 CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
184 public_key NVARCHAR(191) PRIMARY KEY,
185 signing_public_key VARBINARY(32) NOT NULL
186 )`,
187 orElse: () =>
188 sql`
189 CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
190 public_key TEXT PRIMARY KEY,
191 signing_public_key BLOB NOT NULL
192 )`
193 })
194
195 const selectRemoteId = sql<{ remote_id: Uint8Array }>`
196 SELECT remote_id
197 FROM ${remoteIdTableSql}
198 WHERE singleton = 1
199 `
200
201 const remoteId = yield* selectRemoteId.pipe(
202 Effect.flatMap((rows) => {
203 const existing = rows[0]
204 if (existing !== undefined) {
205 return Effect.succeed(existing.remote_id as RemoteId)
206 }
207
208 const created = makeRemoteIdUnsafe()
209 return sql`
210 INSERT INTO ${remoteIdTableSql} (singleton, remote_id)
211 VALUES (1, ${created})
212 `.pipe(
213 Effect.catchIf(
214 isConstraintConflict,
215 () => Effect.void
216 ),
217 Effect.andThen(selectRemoteId),
218 Effect.map((rows) => rows[0]?.remote_id as RemoteId | undefined),
219 Effect.map((persisted) => persisted ?? created)
220 )
221 })
222 )
223
224 const pubsubs = yield* RcMap.make({
225 lookup: (_storeId: string) =>
226 Effect.acquireRelease(
227 PubSub.unbounded<RemoteEntry>(),
228 PubSub.shutdown
229 ),
230 idleTimeToLive: "5 minutes"
231 })
232
233 const ensureStore = (storeId: string) =>
234 sql.onDialectOrElse({
235 pg: () =>
236 sql`
237 INSERT INTO ${storesTableSql} (store_id, next_sequence)
238 VALUES (${storeId}, 1)
239 ON CONFLICT (store_id) DO NOTHING
240 `,
241 mysql: () =>
242 sql`
243 INSERT INTO ${storesTableSql} (store_id, next_sequence)
244 VALUES (${storeId}, 1)
245 ON DUPLICATE KEY UPDATE store_id = store_id
246 `,
247 mssql: () =>
248 sql`
249 MERGE ${storesTableSql} WITH (HOLDLOCK) AS target
250 USING (SELECT ${storeId} AS store_id, 1 AS next_sequence) AS source
251 ON target.store_id = source.store_id
252 WHEN NOT MATCHED THEN
253 INSERT (store_id, next_sequence)
254 VALUES (source.store_id, source.next_sequence);
255 `,
256 orElse: () =>
257 sql`
258 INSERT INTO ${storesTableSql} (store_id, next_sequence)
259 VALUES (${storeId}, 1)
260 ON CONFLICT DO NOTHING
261 `
262 })
263
264 const lockStore = (storeId: string) =>
265 sql.onDialectOrElse({
266 pg: () =>
267 sql`
268 SELECT next_sequence
269 FROM ${storesTableSql}
270 WHERE store_id = ${storeId}
271 FOR UPDATE
272 `,
273 mysql: () =>
274 sql`
275 SELECT next_sequence
276 FROM ${storesTableSql}
277 WHERE store_id = ${storeId}
278 FOR UPDATE
279 `,
280 mssql: () =>
281 sql`
282 SELECT next_sequence
283 FROM ${storesTableSql} WITH (UPDLOCK, HOLDLOCK)
284 WHERE store_id = ${storeId}
285 `,
286 orElse: () =>
287 sql`
288 UPDATE ${storesTableSql}
289 SET next_sequence = next_sequence
290 WHERE store_id = ${storeId}
291 RETURNING next_sequence
292 `
293 }).pipe(
294 Effect.flatMap(decodeStoreSequence)
295 )
296
297 const setNextSequence = (storeId: string, nextSequence: number) =>
298 sql`
299 UPDATE ${storesTableSql}
300 SET next_sequence = ${nextSequence}
301 WHERE store_id = ${storeId}
302 `
303
304 const selectEntriesAfter = (storeId: string, startSequence: number) =>
305 sql`
306 SELECT sequence, entry_id, event, primary_key, payload
307 FROM ${entriesTableSql}
308 WHERE store_id = ${storeId} AND sequence >= ${startSequence}
309 ORDER BY sequence ASC
310 `.pipe(
311 Effect.flatMap(decodeRemoteEntries)
312 )
313
314 const getSessionAuthBinding = (publicKey: string) =>
315 sql`
316 SELECT public_key, signing_public_key
317 FROM ${sessionAuthBindingsTableSql}
318 WHERE public_key = ${publicKey}
319 `.pipe(
320 Effect.flatMap(decodeSessionAuthBindings),
321 Effect.map((rows) => {
322 const row = rows[0]
323 return row === undefined ? undefined : row.signing_public_key as Uint8Array<ArrayBuffer>
324 }),
325 Effect.orDie
326 )
327
328 return EventLogServerUnencrypted.Storage.of({
329 getId: Effect.succeed(remoteId),
330 getOrCreateSessionAuthBinding: Effect.fnUntraced(
331 function*(publicKey, signingPublicKey) {
332 const existing = yield* getSessionAuthBinding(publicKey)
333 if (existing !== undefined) {
334 return existing
335 }
336 return yield* sql`
337 INSERT INTO ${sessionAuthBindingsTableSql} (public_key, signing_public_key)
338 VALUES (${publicKey}, ${signingPublicKey})
339 `.pipe(
340 Effect.as(signingPublicKey)
341 )
342 },
343 sql.withTransaction,
344 Effect.orDie,
345 withTracerDisabled
346 ),
347 entriesAfter: (storeId, entry) =>
348 sql`
349 SELECT sequence, entry_id, event, primary_key, payload
350 FROM ${entriesTableSql}
351 WHERE store_id = ${storeId} AND entry_id >= ${entry.id}
352 ORDER BY sequence ASC
353 `.pipe(
354 Effect.flatMap(decodeRemoteEntries),
355 Effect.map(Arr.map((r) => r.entry)),
356 Effect.orDie,
357 withTracerDisabled
358 ),
359 write: Effect.fnUntraced(
360 function*(storeId, entries) {
361 if (entries.length === 0) {
362 return []
363 }
364
365 yield* ensureStore(storeId)
366 const currentNextSequence = yield* lockStore(storeId)
367
368 const committed: Array<RemoteEntry> = []
369 let rowsForInsert: Array<{
370 readonly store_id: string
371 readonly sequence: number
372 readonly entry_id: Uint8Array
373 readonly event: string
374 readonly primary_key: string
375 readonly payload: Uint8Array
376 }> = []
377
378 for (let index = 0; index < entries.length; index++) {
379 const entry = entries[index]
380 const remoteSequence = currentNextSequence + index
381 committed.push(
382 new RemoteEntry({
383 remoteSequence,
384 entry
385 }, { disableChecks: true })
386 )
387 rowsForInsert.push({
388 store_id: storeId,
389 sequence: remoteSequence,
390 entry_id: entry.id,
391 event: entry.event,
392 primary_key: entry.primaryKey,
393 payload: entry.payload
394 })
395 if (rowsForInsert.length >= insertBatchSize) {
396 yield* sql`INSERT INTO ${entriesTableSql} ${sql.insert(rowsForInsert)}`
397 rowsForInsert = []
398 }
399 }
400 if (rowsForInsert.length > 0) {
401 yield* sql`INSERT INTO ${entriesTableSql} ${sql.insert(rowsForInsert)}`
402 }
403 const nextSequence = currentNextSequence + entries.length
404 yield* setNextSequence(storeId, nextSequence)
405
406 const pubsub = yield* RcMap.get(pubsubs, storeId)
407 yield* PubSub.publishAll(pubsub, committed)
408
409 return committed
410 },
411 Effect.scoped,
412 sql.withTransaction,
413 withTracerDisabled,
414 Effect.orDie
415 ),
416 changes: Effect.fnUntraced(
417 function*({ storeId, startSequence, compactors }) {
418 const pubsub = yield* RcMap.get(pubsubs, storeId)
419 const subscription = yield* PubSub.subscribe(pubsub)
420 const backlog = yield* EventLogServerUnencrypted.compactBacklog({
421 compactors,
422 remoteEntries: yield* selectEntriesAfter(storeId, startSequence)
423 })
424 let watermark = backlog.length > 0 ? backlog[backlog.length - 1]!.remoteSequence : startSequence - 1
425
426 return Stream.fromArray(backlog).pipe(
427 Stream.concat(
428 Stream.fromSubscription(subscription).pipe(
429 Stream.filter((entry) => entry.remoteSequence > watermark)
430 )
431 )
432 )
433 },
434 Effect.orDie,
435 withTracerDisabled,
436 Stream.unwrap
437 ),
438 withTransaction: (effect) =>
439 sql.withTransaction(effect).pipe(
440 Effect.catchIf(SqlError.isSqlError, Effect.die)
441 )
442 })
443 }).pipe(withTracerDisabled)
444
445/**
446 * Provides unencrypted server `Storage` using the SQL-backed implementation.
447 *
448 * @stability unstable
449 * @category layers
450 * @since 4.0.0
451 */
452export const layerStorage = (options?: {
453 readonly entryTablePrefix?: string
454 readonly remoteIdTable?: string
455 readonly insertBatchSize?: number
456}): Layer.Layer<EventLogServerUnencrypted.Storage, SqlError.SqlError, SqlClient.SqlClient> =>
457 Layer.effect(EventLogServerUnencrypted.Storage)(makeStorage(options))
458
459const EntrySql = Schema.Struct({
460 entry_id: EntryId,
461 event: Schema.String,
462 primary_key: Schema.String,
463 payload: Schema.Uint8Array
464})
465
466type EntrySql = Schema.Schema.Type<typeof EntrySql>
467
468const SqlNatural = Schema.Union([
469 Schema.Natural,
470 Schema.FiniteFromString.check(Schema.isInt(), Schema.isGreaterThanOrEqualTo(0)),
471 Schema.BigInt.pipe(
472 Schema.decodeTo(
473 Schema.Natural,
474 SchemaTransformation.transform({
475 decode: Number,
476 encode: BigInt
477 })
478 )
479 )
480])
481
482const RemoteEntrySql = Schema.Struct({
483 ...EntrySql.fields,
484 sequence: SqlNatural
485})
486
487type RemoteEntrySql = Schema.Schema.Type<typeof RemoteEntrySql>
488
489const StoreSequenceSql = Schema.Struct({
490 next_sequence: SqlNatural
491})
492
493const SessionAuthBindingSql = Schema.Struct({
494 public_key: Schema.String,
495 signing_public_key: Schema.Uint8Array
496})
497
498type SessionAuthBindingSql = Schema.Schema.Type<typeof SessionAuthBindingSql>
499
500const decodeRemoteEntryRows = Schema.decodeUnknownEffect(Schema.mutable(Schema.Array(RemoteEntrySql)))
501const decodeStoreSequenceRows = Schema.decodeUnknownEffect(Schema.Array(StoreSequenceSql))
502const decodeSessionAuthBindingRows = Schema.decodeUnknownEffect(Schema.Array(SessionAuthBindingSql))
503
504const toEntry = (row: EntrySql): Entry =>
505 new Entry({
506 id: row.entry_id,
507 event: row.event,
508 primaryKey: row.primary_key,
509 payload: row.payload
510 }, { disableChecks: true })
511
512const toRemoteEntry = (row: RemoteEntrySql): RemoteEntry =>
513 new RemoteEntry({
514 remoteSequence: row.sequence,
515 entry: toEntry(row)
516 }, { disableChecks: true })
517
518const decodeRemoteEntries = (
519 rows: unknown
520): Effect.Effect<Array<RemoteEntry>, Schema.SchemaError> =>
521 decodeRemoteEntryRows(rows).pipe(
522 Effect.map((rows) => rows.map(toRemoteEntry))
523 )
524
525const decodeStoreSequence = (rows: unknown): Effect.Effect<number, Schema.SchemaError> =>
526 decodeStoreSequenceRows(rows).pipe(
527 Effect.flatMap((rows) => {
528 const row = rows[0]
529 if (row === undefined) {
530 return Effect.die("SqlEventLogServerUnencrypted missing store sequence row")
531 }
532 return Effect.succeed(row.next_sequence)
533 })
534 )
535
536const decodeSessionAuthBindings = (
537 rows: unknown
538): Effect.Effect<ReadonlyArray<SessionAuthBindingSql>, Schema.SchemaError> => decodeSessionAuthBindingRows(rows)
539
540const withTracerDisabled = Effect.withTracerEnabled(false)
541
542const isConstraintConflict = (error: SqlError.SqlError): boolean =>
543 error.reason._tag === "ConstraintError" || error.reason._tag === "UniqueViolation"