1/**
2 * SQL-backed persistence for the unstable event-log journal.
3 *
4 * This module implements `EventJournal` on top of a `SqlClient`. It stores event
5 * entries as encoded bytes and stores per-remote sequence metadata in separate
6 * tables, giving event-log programs a durable journal that can be replayed after
7 * restart and synchronized with remote journals.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import * as Arr from "../Array.ts"
13import * as Effect from "../Effect.ts"
14import * as Uuid from "../internal/uuid.ts"
15import * as Layer from "../Layer.ts"
16import * as PubSub from "../PubSub.ts"
17import * as Schema from "../Schema.ts"
18import * as SchemaTransformation from "../SchemaTransformation.ts"
19import * as SqlClient from "../sql/SqlClient.ts"
20import * as SqlError from "../sql/SqlError.ts"
21import * as SqlSchema from "../sql/SqlSchema.ts"
22import * as EventJournal from "./EventJournal.ts"
23
24type WriteFromRemoteOptions = Parameters<EventJournal.EventJournal["Service"]["writeFromRemote"]>[0]
25
26/**
27 * Creates an `EventJournal` backed by a SQL database.
28 *
29 * **Details**
30 *
31 * The constructor creates the entry and remote metadata tables when needed,
32 * persists local and remote entries, and uses the configured `SqlClient`.
33 *
34 * @stability unstable
35 * @category constructors
36 * @since 4.0.0
37 */
38export const make = (options?: {
39 readonly entryTable?: string
40 readonly remotesTable?: string
41}): Effect.Effect<
42 EventJournal.EventJournal["Service"],
43 SqlError.SqlError,
44 SqlClient.SqlClient
45> =>
46 Effect.gen(function*() {
47 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
48
49 const entryTable = options?.entryTable ?? "effect_event_journal"
50 const remotesTable = options?.remotesTable ?? "effect_event_remotes"
51
52 const entryTableSql = sql(entryTable)
53 const remotesTableSql = sql(remotesTable)
54
55 yield* sql.onDialectOrElse({
56 pg: () =>
57 sql`
58 CREATE TABLE IF NOT EXISTS ${entryTableSql} (
59 id UUID PRIMARY KEY,
60 event TEXT NOT NULL,
61 primary_key TEXT NOT NULL,
62 payload BYTEA NOT NULL,
63 timestamp BIGINT NOT NULL
64 )`,
65 mysql: () =>
66 sql`
67 CREATE TABLE IF NOT EXISTS ${entryTableSql} (
68 id BINARY(16) PRIMARY KEY,
69 event TEXT NOT NULL,
70 primary_key TEXT NOT NULL,
71 payload BLOB NOT NULL,
72 timestamp BIGINT NOT NULL
73 )`,
74 mssql: () =>
75 sql`
76 CREATE TABLE IF NOT EXISTS ${entryTableSql} (
77 id UNIQUEIDENTIFIER PRIMARY KEY,
78 event NVARCHAR(MAX) NOT NULL,
79 primary_key NVARCHAR(MAX) NOT NULL,
80 payload VARBINARY(MAX) NOT NULL,
81 timestamp BIGINT NOT NULL
82 )`,
83 orElse: () =>
84 sql`
85 CREATE TABLE IF NOT EXISTS ${entryTableSql} (
86 id BLOB PRIMARY KEY,
87 event TEXT NOT NULL,
88 primary_key TEXT NOT NULL,
89 payload BLOB NOT NULL,
90 timestamp INTEGER NOT NULL
91 )`
92 }).pipe(withTracerDisabled)
93
94 yield* sql.onDialectOrElse({
95 pg: () =>
96 sql`
97 CREATE TABLE IF NOT EXISTS ${remotesTableSql} (
98 remote_id UUID NOT NULL,
99 entry_id UUID NOT NULL,
100 sequence INT NOT NULL,
101 PRIMARY KEY (remote_id, entry_id)
102 )`,
103 mysql: () =>
104 sql`
105 CREATE TABLE IF NOT EXISTS ${remotesTableSql} (
106 remote_id BINARY(16) NOT NULL,
107 entry_id BINARY(16) NOT NULL,
108 sequence INT NOT NULL,
109 PRIMARY KEY (remote_id, entry_id)
110 )`,
111 mssql: () =>
112 sql`
113 CREATE TABLE IF NOT EXISTS ${remotesTableSql} (
114 remote_id UNIQUEIDENTIFIER NOT NULL,
115 entry_id UNIQUEIDENTIFIER NOT NULL,
116 sequence INT NOT NULL,
117 PRIMARY KEY (remote_id, entry_id)
118 )`,
119 orElse: () =>
120 sql`
121 CREATE TABLE IF NOT EXISTS ${remotesTableSql} (
122 remote_id BLOB NOT NULL,
123 entry_id BLOB NOT NULL,
124 sequence INT NOT NULL,
125 PRIMARY KEY (remote_id, entry_id)
126 )`
127 }).pipe(withTracerDisabled)
128
129 const decodeEntryRows = Schema.decodeUnknownEffect(EntryRowArray)
130 const decodeEntryIdRows = Schema.decodeUnknownEffect(EntryIdRowArray)
131 const toEntries = (rows: ReadonlyArray<EntryRow>): ReadonlyArray<EventJournal.Entry> => rows.map(toEntry)
132
133 const insertEntry = SqlSchema.void({
134 Request: EntryRow,
135 execute: (entry) => sql`INSERT INTO ${entryTableSql} ${sql.insert(entry)} ON CONFLICT DO NOTHING`
136 })
137 const insertEntries = SqlSchema.void({
138 Request: EntryRowArray,
139 execute: (entries) => sql`INSERT INTO ${entryTableSql} ${sql.insert(entries)} ON CONFLICT DO NOTHING`
140 })
141 const insertRemotes = SqlSchema.void({
142 Request: RemoteRowArray,
143 execute: (entries) => sql`INSERT INTO ${remotesTableSql} ${sql.insert(entries)} ON CONFLICT DO NOTHING`
144 })
145
146 const pubsub = yield* PubSub.unbounded<EventJournal.Entry>()
147
148 const writeFromRemote = Effect.fnUntraced(function*(options: WriteFromRemoteOptions): Effect.fn.Return<
149 {
150 readonly duplicateEntries: ReadonlyArray<EventJournal.Entry>
151 },
152 EventJournal.EventJournalError | Schema.SchemaError | SqlError.SqlError
153 > {
154 const entries = options.entries.map((remoteEntry) => remoteEntry.entry)
155 const remoteRows = options.entries.map((remoteEntry) => ({
156 remote_id: options.remoteId,
157 entry_id: remoteEntry.entry.id,
158 sequence: remoteEntry.remoteSequence
159 }))
160
161 const existingIds = new Set<string>()
162 if (entries.length > 0) {
163 yield* sql`SELECT id FROM ${entryTableSql} WHERE ${
164 sql.in(
165 "id",
166 entries.map((entry) => entry.id)
167 )
168 }`.pipe(
169 Effect.flatMap(decodeEntryIdRows),
170 Effect.tap((rows) =>
171 Effect.sync(() => {
172 for (const row of rows) {
173 existingIds.add(Uuid.stringify(row.id))
174 }
175 })
176 )
177 )
178 }
179 if (entries.length > 0) {
180 yield* insertEntries(entries.map(toEntryRow))
181 }
182 if (remoteRows.length > 0) {
183 yield* insertRemotes(remoteRows)
184 }
185
186 const uncommitted = options.entries.filter((entry) => !existingIds.has(entry.entry.idString))
187 const duplicateEntries = options.entries
188 .filter((entry) => existingIds.has(entry.entry.idString))
189 .map((entry) => entry.entry)
190 const compacted = options.compact
191 ? yield* options.compact(uncommitted)
192 : uncommitted.map((remoteEntry) => remoteEntry.entry)
193
194 for (const entry of compacted) {
195 const conflicts = yield* sql`
196 SELECT *
197 FROM ${entryTableSql}
198 WHERE event = ${entry.event} AND
199 primary_key = ${entry.primaryKey} AND
200 timestamp >= ${entry.createdAtMillis}
201 ORDER BY timestamp ASC
202 `.pipe(
203 Effect.flatMap(decodeEntryRows),
204 Effect.map(toEntries)
205 )
206 yield* options.effect({ entry, conflicts })
207 }
208
209 return {
210 duplicateEntries
211 }
212 })
213
214 return EventJournal.EventJournal.of({
215 entries: sql`SELECT * FROM ${entryTableSql} ORDER BY timestamp ASC`.pipe(
216 withTracerDisabled,
217 Effect.flatMap(decodeEntryRows),
218 Effect.map(toEntries),
219 Effect.mapError((cause) => new EventJournal.EventJournalError({ cause, method: "entries" }))
220 ),
221 write: Effect.fnUntraced(
222 function*({ effect, event, payload, primaryKey }) {
223 const entry = new EventJournal.Entry({
224 id: EventJournal.makeEntryIdUnsafe(),
225 event,
226 primaryKey,
227 payload
228 }, { disableChecks: true })
229 return yield* Effect.uninterruptibleMask((restore) =>
230 restore(effect(entry)).pipe(
231 Effect.tap(
232 insertEntry(toEntryRow(entry)).pipe(
233 Effect.mapError((cause) => new EventJournal.EventJournalError({ cause, method: "write" }))
234 )
235 ),
236 Effect.tap(PubSub.publish(pubsub, entry))
237 )
238 )
239 },
240 withTracerDisabled
241 ),
242 writeFromRemote: (options) =>
243 writeFromRemote(options).pipe(
244 withTracerDisabled,
245 Effect.catchIf(
246 (e) => e._tag !== "EventJournalError",
247 (cause) => Effect.fail(new EventJournal.EventJournalError({ cause, method: "writeFromRemote" }))
248 )
249 ),
250 withRemoteUncommited: Effect.fnUntraced(
251 function*(remoteId, f) {
252 const entries = yield* sql`
253 SELECT *
254 FROM ${entryTableSql}
255 WHERE id NOT IN (SELECT entry_id FROM ${remotesTableSql} WHERE remote_id = ${remoteId})
256 ORDER BY timestamp ASC
257 `.pipe(
258 Effect.flatMap(decodeEntryRows),
259 Effect.map(toEntries),
260 Effect.mapError((cause) => new EventJournal.EventJournalError({ cause, method: "withRemoteUncommited" }))
261 )
262 if (!Arr.isReadonlyArrayNonEmpty(entries)) return yield* Effect.succeedNone
263 return yield* Effect.asSome(f(entries))
264 },
265 withTracerDisabled
266 ),
267 nextRemoteSequence: (remoteId) =>
268 sql<{ max: number | null }>`SELECT MAX(sequence) AS max FROM ${remotesTableSql} WHERE remote_id = ${remoteId}`
269 .pipe(
270 Effect.map((rows) => {
271 const value = rows[0]?.max
272 if (value === null || value === undefined) return 0
273 return Number(value) + 1
274 }),
275 withTracerDisabled,
276 Effect.mapError((cause) => new EventJournal.EventJournalError({ cause, method: "nextRemoteSequence" }))
277 ),
278 changes: PubSub.subscribe(pubsub),
279 destroy: Effect.gen(function*() {
280 yield* sql`DROP TABLE ${entryTableSql}`
281 yield* sql`DROP TABLE ${remotesTableSql}`
282 }).pipe(
283 withTracerDisabled,
284 Effect.mapError((cause) => new EventJournal.EventJournalError({ cause, method: "destroy" }))
285 ),
286 withLock(_storeId) {
287 return (effect) =>
288 sql.withTransaction(effect).pipe(
289 Effect.catchIf(SqlError.isSqlError, Effect.die)
290 )
291 }
292 })
293 })
294
295/**
296 * Provides `EventJournal` using the SQL-backed implementation created by
297 * `make`.
298 *
299 * **When to use**
300 *
301 * Use when composing a Layer graph that should provide a persistent SQL-backed
302 * `EventJournal` from an existing `SqlClient` service.
303 *
304 * **Details**
305 *
306 * The layer delegates to `make(options)`, so the same optional `entryTable` and
307 * `remotesTable` settings are used and construction requires `SqlClient` and
308 * may fail with `SqlError`.
309 *
310 * **Gotchas**
311 *
312 * Layer construction performs the same minimal `CREATE TABLE IF NOT EXISTS`
313 * setup as `make`; manage indexes and schema migrations outside this layer when
314 * your SQL schema needs more than the built-in tables.
315 *
316 * @see {@link make} for constructing the SQL-backed service directly
317 * @see {@link EventJournal.layerMemory} for an in-memory `EventJournal` layer
318 * @see {@link EventJournal.layerIndexedDb} for an IndexedDB-backed `EventJournal` layer
319 *
320 * @stability unstable
321 * @category layers
322 * @since 4.0.0
323 */
324export const layer = (options?: {
325 readonly entryTable?: string
326 readonly remotesTable?: string
327}): Layer.Layer<EventJournal.EventJournal, SqlError.SqlError, SqlClient.SqlClient> =>
328 Layer.effect(EventJournal.EventJournal)(make(options))
329
330const ArrayBuffer = Schema.instanceOf(globalThis.ArrayBuffer, {
331 expected: "ArrayBuffer"
332})
333
334const SqlUint8Array = Schema.Union([Schema.Uint8Array, ArrayBuffer]).pipe(
335 Schema.decodeTo(
336 Schema.Uint8Array,
337 SchemaTransformation.transform({
338 decode: (value) => value instanceof globalThis.ArrayBuffer ? new globalThis.Uint8Array(value) : value,
339 encode: (value) => value
340 })
341 )
342)
343
344const SqlEntryId = SqlUint8Array.pipe(Schema.decodeTo(EventJournal.EntryId))
345
346const EntryRow = Schema.Struct({
347 id: SqlEntryId,
348 event: Schema.String,
349 primary_key: Schema.String,
350 payload: SqlUint8Array,
351 timestamp: Schema.Int
352})
353
354const EntryRowArray = Schema.Array(EntryRow)
355const EntryIdRowArray = Schema.Array(Schema.Struct({ id: SqlEntryId }))
356
357type EntryRow = Schema.Schema.Type<typeof EntryRow>
358
359const toEntry = (row: EntryRow): EventJournal.Entry =>
360 new EventJournal.Entry({
361 id: row.id,
362 event: row.event,
363 primaryKey: row.primary_key,
364 payload: row.payload
365 }, { disableChecks: true })
366
367const toEntryRow = (entry: EventJournal.Entry): EntryRow => ({
368 id: entry.id,
369 event: entry.event,
370 primary_key: entry.primaryKey,
371 payload: entry.payload,
372 timestamp: EventJournal.entryIdMillis(entry.id)
373})
374
375const RemoteRow = Schema.Struct({
376 remote_id: EventJournal.RemoteId,
377 entry_id: EventJournal.EntryId,
378 sequence: Schema.Natural
379})
380
381const RemoteRowArray = Schema.Array(RemoteRow)
382
383const withTracerDisabled = Effect.withTracerEnabled(false)