1/**
2 * Stores encrypted event-log server state in SQL.
3 *
4 * This module provides the durable `Storage` implementation used by
5 * `EventLogServerEncrypted` when entries should be stored without exposing
6 * plaintext event data to the database. It persists the server remote id,
7 * session authentication bindings, and encrypted entry tables, assigns stable
8 * sequence numbers, and streams changes. Clients remain responsible for
9 * encrypting writes and decrypting reads.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
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 type * as Scope from "../Scope.ts"
20import * as SqlClient from "../sql/SqlClient.ts"
21import type * as SqlError from "../sql/SqlError.ts"
22import * as Stream from "../Stream.ts"
23import { EntryId, makeRemoteIdUnsafe, type RemoteId } from "./EventJournal.ts"
24import * as EventLogEncryption from "./EventLogEncryption.ts"
25import * as EventLogServerEncrypted from "./EventLogServerEncrypted.ts"
26
27/**
28 * Creates encrypted event-log server `Storage` backed by SQL.
29 *
30 * **Details**
31 *
32 * It persists the server remote id, session authentication bindings, and encrypted
33 * entries in dialect-specific tables, creating per-identity/store entry tables as
34 * needed.
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 EventLogServerEncrypted.Storage["Service"],
46 SqlError.SqlError,
47 SqlClient.SqlClient | EventLogEncryption.EventLogEncryption | Scope.Scope
48> =>
49 Effect.gen(function*() {
50 const encryptions = yield* EventLogEncryption.EventLogEncryption
51 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
52
53 const tablePrefix = options?.entryTablePrefix ?? "effect_events"
54 const remoteIdTable = options?.remoteIdTable ?? "effect_remote_id"
55 const sessionAuthBindingsTable = `${tablePrefix}_session_auth_bindings`
56 const insertBatchSize = options?.insertBatchSize ?? 200
57
58 const remoteIdTableSql = sql(remoteIdTable)
59 const sessionAuthBindingsTableSql = sql(sessionAuthBindingsTable)
60
61 yield* sql.onDialectOrElse({
62 pg: () =>
63 sql`
64 CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
65 remote_id BYTEA PRIMARY KEY
66 )`,
67 mysql: () =>
68 sql`
69 CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
70 remote_id BINARY(16) PRIMARY KEY
71 )`,
72 mssql: () =>
73 sql`
74 CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
75 remote_id VARBINARY(16) PRIMARY KEY
76 )`,
77 orElse: () =>
78 sql`
79 CREATE TABLE IF NOT EXISTS ${remoteIdTableSql} (
80 remote_id BLOB PRIMARY KEY
81 )`
82 })
83
84 yield* sql.onDialectOrElse({
85 pg: () =>
86 sql`
87 CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
88 public_key TEXT PRIMARY KEY,
89 signing_public_key BYTEA NOT NULL
90 )`,
91 mysql: () =>
92 sql`
93 CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
94 public_key VARCHAR(191) PRIMARY KEY,
95 signing_public_key BINARY(32) NOT NULL
96 )`,
97 mssql: () =>
98 sql`
99 CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
100 public_key NVARCHAR(191) PRIMARY KEY,
101 signing_public_key VARBINARY(32) NOT NULL
102 )`,
103 orElse: () =>
104 sql`
105 CREATE TABLE IF NOT EXISTS ${sessionAuthBindingsTableSql} (
106 public_key TEXT PRIMARY KEY,
107 signing_public_key BLOB NOT NULL
108 )`
109 })
110
111 const remoteId = yield* sql<{ remote_id: Uint8Array }>`SELECT remote_id FROM ${remoteIdTableSql}`.pipe(
112 Effect.flatMap((results) => {
113 if (results.length > 0) {
114 return Effect.succeed(results[0].remote_id as RemoteId)
115 }
116 const created = makeRemoteIdUnsafe()
117 return Effect.as(
118 sql`INSERT INTO ${remoteIdTableSql} (remote_id) VALUES (${created})`,
119 created
120 )
121 })
122 )
123
124 const resources = yield* RcMap.make({
125 lookup: Effect.fnUntraced(function*(scopeKey: string) {
126 const scopeHash = (yield* encryptions.sha256String(new TextEncoder().encode(scopeKey))).slice(0, 16)
127 const table = `${tablePrefix}_${scopeHash}`
128 const tableSql = sql(table)
129
130 yield* sql.onDialectOrElse({
131 pg: () =>
132 sql`
133 CREATE TABLE IF NOT EXISTS ${tableSql} (
134 sequence SERIAL PRIMARY KEY,
135 iv BYTEA NOT NULL,
136 entry_id BYTEA UNIQUE NOT NULL,
137 encrypted_entry BYTEA NOT NULL
138 )`,
139 mysql: () =>
140 sql`
141 CREATE TABLE IF NOT EXISTS ${tableSql} (
142 sequence INT AUTO_INCREMENT PRIMARY KEY,
143 iv BINARY(12) NOT NULL,
144 entry_id BINARY(16) UNIQUE NOT NULL,
145 encrypted_entry BLOB NOT NULL
146 )`,
147 mssql: () =>
148 sql`
149 CREATE TABLE IF NOT EXISTS ${tableSql} (
150 sequence INT IDENTITY(1,1) PRIMARY KEY,
151 iv VARBINARY(12) NOT NULL,
152 entry_id VARBINARY(16) UNIQUE NOT NULL,
153 encrypted_entry VARBINARY(MAX) NOT NULL
154 )`,
155 orElse: () =>
156 sql`
157 CREATE TABLE IF NOT EXISTS ${tableSql} (
158 sequence INTEGER PRIMARY KEY AUTOINCREMENT,
159 iv BLOB NOT NULL,
160 entry_id BLOB UNIQUE NOT NULL,
161 encrypted_entry BLOB NOT NULL
162 )`
163 })
164
165 const pubsub = yield* Effect.acquireRelease(
166 PubSub.unbounded<EventLogEncryption.EncryptedRemoteEntry>(),
167 PubSub.shutdown
168 )
169 return { pubsub, table } as const
170 }, withTracerDisabled),
171 idleTimeToLive: "5 minutes"
172 })
173
174 const getSessionAuthBinding = (publicKey: string) =>
175 sql`
176 SELECT public_key, signing_public_key
177 FROM ${sessionAuthBindingsTableSql}
178 WHERE public_key = ${publicKey}
179 `.pipe(
180 Effect.flatMap(decodeSessionAuthBindings),
181 Effect.map((rows) => {
182 const row = rows[0]
183 return row === undefined ? undefined : row.signing_public_key as Uint8Array<ArrayBuffer>
184 }),
185 Effect.orDie
186 )
187
188 return EventLogServerEncrypted.Storage.of({
189 getId: Effect.succeed(remoteId),
190 getOrCreateSessionAuthBinding: Effect.fnUntraced(
191 function*(publicKey, signingPublicKey) {
192 const existing = yield* getSessionAuthBinding(publicKey)
193 if (existing !== undefined) {
194 return existing
195 }
196 return yield* sql`
197 INSERT INTO ${sessionAuthBindingsTableSql} (public_key, signing_public_key)
198 VALUES (${publicKey}, ${signingPublicKey})
199 `.pipe(
200 Effect.as(signingPublicKey)
201 )
202 },
203 sql.withTransaction,
204 withTracerDisabled,
205 Effect.orDie
206 ),
207 write: Effect.fnUntraced(
208 function*(publicKey, storeId, entries) {
209 if (entries.length === 0) return []
210 const scopeKey = makeEncryptedScopeKey(publicKey, storeId)
211 const { pubsub, table } = yield* RcMap.get(resources, scopeKey)
212 const forInsert: Array<{
213 readonly ids: Array<EntryId>
214 readonly entries: Array<{
215 iv: Uint8Array
216 entry_id: Uint8Array
217 encrypted_entry: Uint8Array
218 }>
219 }> = [{ ids: [], entries: [] }]
220 let currentBatch = forInsert[0]
221 for (const entry of entries) {
222 currentBatch.ids.push(entry.entryId)
223 currentBatch.entries.push({
224 iv: entry.iv,
225 entry_id: entry.entryId,
226 encrypted_entry: entry.encryptedEntry
227 })
228 if (currentBatch.entries.length === insertBatchSize) {
229 currentBatch = { ids: [], entries: [] }
230 forInsert.push(currentBatch)
231 }
232 }
233
234 const allEntries: Array<EventLogEncryption.EncryptedRemoteEntry> = []
235 for (const batch of forInsert) {
236 if (batch.entries.length === 0) continue
237 const encryptedEntries = yield* sql`
238 INSERT INTO ${sql(table)} ${sql.insert(batch.entries)} ON CONFLICT DO NOTHING
239 `.pipe(
240 Effect.andThen(
241 sql`SELECT * FROM ${sql(table)} WHERE ${sql.in("entry_id", batch.ids)} ORDER BY sequence ASC`
242 ),
243 Effect.flatMap(decodeEntries)
244 )
245 yield* PubSub.publishAll(pubsub, encryptedEntries)
246 allEntries.push(...encryptedEntries)
247 }
248 return allEntries
249 },
250 Effect.orDie,
251 Effect.scoped,
252 withTracerDisabled
253 ),
254 changes: Effect.fnUntraced(
255 function*(publicKey, storeId, startSequence) {
256 const scopeKey = makeEncryptedScopeKey(publicKey, storeId)
257 const { pubsub, table } = yield* RcMap.get(resources, scopeKey)
258 const subscription = yield* PubSub.subscribe(pubsub)
259 const initial = yield* sql`
260 SELECT * FROM ${sql(table)} WHERE sequence >= ${startSequence} ORDER BY sequence ASC
261 `.pipe(
262 Effect.flatMap(decodeEntries)
263 )
264 return Stream.fromArray(initial).pipe(Stream.concat(Stream.fromSubscription(subscription)))
265 },
266 Effect.orDie,
267 withTracerDisabled,
268 Stream.unwrap
269 )
270 })
271 }).pipe(withTracerDisabled)
272
273const EncryptedRemoteEntrySql = Schema.Struct({
274 sequence: Schema.Natural,
275 iv: Schema.Uint8Array,
276 entry_id: EntryId,
277 encrypted_entry: Schema.Uint8Array
278})
279
280type EncryptedRemoteEntrySql = Schema.Schema.Type<typeof EncryptedRemoteEntrySql>
281
282const SessionAuthBindingSql = Schema.Struct({
283 public_key: Schema.String,
284 signing_public_key: Schema.Uint8Array
285})
286
287type SessionAuthBindingSql = Schema.Schema.Type<typeof SessionAuthBindingSql>
288
289const decodeEntryRows = Schema.decodeUnknownEffect(Schema.Array(EncryptedRemoteEntrySql))
290const decodeSessionAuthBindingRows = Schema.decodeUnknownEffect(Schema.Array(SessionAuthBindingSql))
291
292const toEncryptedRemoteEntry = (row: EncryptedRemoteEntrySql): EventLogEncryption.EncryptedRemoteEntry => ({
293 sequence: row.sequence,
294 iv: row.iv as Uint8Array<ArrayBuffer>,
295 entryId: row.entry_id,
296 encryptedEntry: row.encrypted_entry as Uint8Array<ArrayBuffer>
297})
298
299const decodeEntries = (
300 rows: unknown
301): Effect.Effect<ReadonlyArray<EventLogEncryption.EncryptedRemoteEntry>, Schema.SchemaError> =>
302 decodeEntryRows(rows).pipe(Effect.map((entries) => entries.map(toEncryptedRemoteEntry)))
303
304const decodeSessionAuthBindings = (
305 rows: unknown
306): Effect.Effect<ReadonlyArray<SessionAuthBindingSql>, Schema.SchemaError> => decodeSessionAuthBindingRows(rows)
307
308/**
309 * Provides encrypted server `Storage` using the SQL-backed implementation.
310 *
311 * @stability unstable
312 * @category layers
313 * @since 4.0.0
314 */
315export const layerStorage = (options?: {
316 readonly entryTablePrefix?: string
317 readonly remoteIdTable?: string
318 readonly insertBatchSize?: number
319}): Layer.Layer<
320 EventLogServerEncrypted.Storage,
321 SqlError.SqlError,
322 SqlClient.SqlClient | EventLogEncryption.EventLogEncryption
323> => Layer.effect(EventLogServerEncrypted.Storage)(makeStorage(options))
324
325/**
326 * Provides SQL-backed encrypted server `Storage` and supplies the default Web
327 * Crypto `EventLogEncryption` layer.
328 *
329 * @stability unstable
330 * @category layers
331 * @since 4.0.0
332 */
333export const layerStorageSubtle = (options?: {
334 readonly entryTablePrefix?: string
335 readonly remoteIdTable?: string
336 readonly insertBatchSize?: number
337}): Layer.Layer<EventLogServerEncrypted.Storage, SqlError.SqlError, SqlClient.SqlClient> =>
338 layerStorage(options).pipe(
339 Layer.provide(EventLogEncryption.layerSubtle)
340 )
341
342const makeEncryptedScopeKey = (
343 publicKey: string,
344 storeId: string
345): string => `${publicKey}/${storeId}`
346
347const withTracerDisabled = Effect.withTracerEnabled(false)