1import type { DurableObjectStorage, SqlStorage } from "@cloudflare/workers-types"
2import { SqliteClient, SqliteMigrator } from "@effect/sql-sqlite-do"
3import { assert, describe, it } from "@effect/vitest"
4import { Deferred, Effect, Exit, Fiber, Stream } from "effect"
5import * as Reactivity from "effect/reactivity/Reactivity"
6import * as SqlClient from "effect/sql/SqlClient"
7
8class FakeCursor {
9 readonly columnNames: ReadonlyArray<string>
10 readonly rows: ReadonlyArray<ReadonlyArray<unknown>>
11
12 constructor(rows: ReadonlyArray<Record<string, unknown>>, columns?: ReadonlyArray<string>) {
13 this.columnNames = columns ?? Object.keys(rows[0] ?? {})
14 this.rows = rows.map((row) => this.columnNames.map((column) => row[column]))
15 }
16
17 *raw() {
18 yield* this.rows.map((row) => [...row])
19 }
20}
21
22class FakeSqlStorage {
23 readonly statements: Array<string> = []
24 tables = new Map<string, Array<Record<string, unknown>>>()
25 columns = new Map<string, Array<string>>()
26
27 snapshot() {
28 return {
29 tables: new Map(Array.from(this.tables, ([table, rows]) => [table, rows.map((row) => ({ ...row }))])),
30 columns: new Map(Array.from(this.columns, ([table, columns]) => [table, [...columns]]))
31 }
32 }
33
34 restore(snapshot: ReturnType<FakeSqlStorage["snapshot"]>) {
35 this.tables = snapshot.tables
36 this.columns = snapshot.columns
37 }
38
39 exec(sql: string, ...params: ReadonlyArray<unknown>) {
40 this.statements.push(sql)
41 const statement = normalizeSql(sql)
42 if (/^(BEGIN|COMMIT|ROLLBACK|SAVEPOINT)\b/i.test(statement)) {
43 throw new Error(`Unsupported transaction SQL: ${statement}`)
44 }
45
46 if (/^CREATE TABLE\b/i.test(statement)) {
47 const match = /^CREATE TABLE(?: IF NOT EXISTS)?\s+("[^"]+"|\w+)\s*\((.*)\)$/i.exec(statement)
48 if (match !== null) {
49 const table = unquote(match[1])
50 if (!this.tables.has(table)) {
51 this.tables.set(table, [])
52 this.columns.set(
53 table,
54 match[2].split(",").map((part) => unquote(part.trim().split(/\s+/)[0])).filter((column) =>
55 column.length > 0 && column.toUpperCase() !== "PRIMARY"
56 )
57 )
58 }
59 }
60 return new FakeCursor([])
61 }
62
63 if (/^INSERT INTO\b/i.test(statement)) {
64 const match = /^INSERT INTO\s+("[^"]+"|\w+)\s*\(([^)]+)\)\s+VALUES\s+(.+)$/i.exec(statement)
65 if (match !== null) {
66 const table = unquote(match[1])
67 const columns = match[2].split(",").map((column) => unquote(column.trim()))
68 const rows = this.tables.get(table) ?? []
69 const tableColumns = this.columns.get(table) ?? [...columns]
70 if (!this.tables.has(table)) {
71 this.tables.set(table, rows)
72 this.columns.set(table, tableColumns)
73 }
74 let paramIndex = 0
75 for (const group of valueGroups(match[3])) {
76 const values = group.split(",").map((value) => parseValue(value.trim(), params, () => paramIndex++))
77 const row: Record<string, unknown> = {}
78 if (tableColumns.includes("id") && !columns.includes("id")) {
79 row.id = rows.length + 1
80 }
81 for (let i = 0; i < columns.length; i++) {
82 row[columns[i]] = values[i]
83 }
84 if (tableColumns.includes("created_at") && row.created_at === undefined) {
85 row.created_at = "current_timestamp"
86 }
87 rows.push(row)
88 }
89 }
90 return new FakeCursor([])
91 }
92
93 if (/^SELECT\b/i.test(statement)) {
94 const match =
95 /^SELECT\s+(.+)\s+FROM\s+("[^"]+"|\w+)(?:\s+WHERE\s+("[^"]+"|\w+)\s*=\s*(\?|('[^']*')|\d+))?(?:\s+ORDER BY\s+("[^"]+"|\w+)\s+DESC)?$/i
96 .exec(statement)
97 if (match !== null) {
98 const table = unquote(match[2])
99 const selectedColumns = match[1].trim() === "*"
100 ? this.columns.get(table) ?? []
101 : match[1].split(",").map((column) => unquote(column.trim()))
102 let rows = [...(this.tables.get(table) ?? [])]
103 if (match[3] !== undefined) {
104 const column = unquote(match[3])
105 const value = match[4] === "?" ? params[0] : parseValue(match[4], [], () => 0)
106 rows = rows.filter((row) => row[column] === value)
107 }
108 if (match[6] !== undefined) {
109 const column = unquote(match[6])
110 rows.sort((a, b) => Number(b[column]) - Number(a[column]))
111 }
112 return new FakeCursor(rows, selectedColumns)
113 }
114 }
115
116 return new FakeCursor([])
117 }
118}
119
120class FakeDurableObjectStorage {
121 readonly sql = new FakeSqlStorage()
122 transactionCalls = 0
123 rollbackCalls = 0
124 readonly snapshots: Array<ReturnType<FakeSqlStorage["snapshot"]>> = []
125
126 transaction<T>(body: (txn: { rollback: () => void }) => Promise<T>): Promise<T> {
127 this.transactionCalls++
128 const snapshot = this.sql.snapshot()
129 this.snapshots.push(snapshot)
130 let rolledBack = false
131 const txn = {
132 rollback: () => {
133 this.rollbackCalls++
134 rolledBack = true
135 }
136 }
137 return Promise.resolve().then(() => body(txn)).then(
138 (value) => {
139 if (rolledBack) {
140 this.sql.restore(snapshot)
141 }
142 return value
143 },
144 (error) => {
145 this.sql.restore(snapshot)
146 throw error
147 }
148 ).finally(() => {
149 assert.strictEqual(this.snapshots.pop(), snapshot)
150 })
151 }
152}
153
154const normalizeSql = (sql: string) => sql.replace(/;$/, "").replace(/\s+/g, " ").trim()
155
156const unquote = (value: string) => value.replace(/^"|"$/g, "")
157
158const valueGroups = (values: string): Array<string> =>
159 values.match(/\(([^)]*)\)/g)?.map((group) => group.slice(1, -1)) ?? []
160
161const parseValue = (
162 value: string,
163 params: ReadonlyArray<unknown>,
164 nextParamIndex: () => number
165): unknown => {
166 if (value === "?") {
167 return params[nextParamIndex()]
168 }
169 if (/^'.*'$/.test(value)) {
170 return value.slice(1, -1)
171 }
172 const number = Number(value)
173 return Number.isNaN(number) ? value : number
174}
175
176const makeClient = (config: SqliteClient.SqliteClientConfig) =>
177 SqliteClient.make(config).pipe(
178 Effect.scoped,
179 Effect.provide(Reactivity.layer)
180 )
181
182const hasForbiddenTransactionSql = (db: FakeSqlStorage) =>
183 db.statements.some((statement) => /^(BEGIN|COMMIT|ROLLBACK|SAVEPOINT)\b/i.test(normalizeSql(statement)))
184
185describe("Client", () => {
186 it.effect("classifies native errors without stable sqlite codes as UnknownError", () =>
187 Effect.gen(function*() {
188 const failingDb = {
189 exec: () => {
190 throw new Error("boom")
191 }
192 } as unknown as SqlStorage
193
194 const client = yield* makeClient({ db: failingDb })
195 const error = yield* Effect.flip(client`SELECT 1`)
196 assert.strictEqual(error.reason._tag, "UnknownError")
197 }))
198
199 it.effect("classifies streaming native errors without stable sqlite codes as UnknownError", () =>
200 Effect.gen(function*() {
201 const driverError = new Error("boom")
202 const failingDb = {
203 exec: () => {
204 throw driverError
205 }
206 } as unknown as SqlStorage
207
208 const client = yield* makeClient({ db: failingDb })
209 const error = yield* client`SELECT 1`.stream.pipe(Stream.runCollect, Effect.flip)
210 assert.strictEqual(error._tag, "SqlError")
211 assert.strictEqual(error.reason._tag, "UnknownError")
212 assert.strictEqual(error.reason.cause, driverError)
213 }))
214
215 it.effect("db-only clients support normal queries", () =>
216 Effect.gen(function*() {
217 const db = new FakeSqlStorage()
218 const sql = yield* makeClient({ db: db as unknown as SqlStorage })
219
220 yield* sql`CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)`
221 yield* sql`INSERT INTO test (name) VALUES (${"hello"})`
222 const rows = yield* sql`SELECT * FROM test`
223
224 assert.deepStrictEqual(rows, [{ id: 1, name: "hello" }])
225 }))
226
227 it.effect("db-only transactions fail clearly without transaction SQL", () =>
228 Effect.gen(function*() {
229 const db = new FakeSqlStorage()
230 const sql = yield* makeClient({ db: db as unknown as SqlStorage })
231
232 yield* sql`CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)`
233 const error = yield* Effect.flip(sql.withTransaction(sql`INSERT INTO test (name) VALUES ('hello')`))
234 const rows = yield* sql`SELECT * FROM test`
235
236 assert.strictEqual(error.reason._tag, "UnknownError")
237 assert.match(error.message, /Transactions require Durable Object storage/)
238 assert.deepStrictEqual(rows, [])
239 assert.strictEqual(hasForbiddenTransactionSql(db), false)
240 }))
241
242 it.effect("storage-backed transactions use DurableObjectStorage.transaction", () =>
243 Effect.gen(function*() {
244 const storage = new FakeDurableObjectStorage()
245 const sql = yield* makeClient({ storage: storage as unknown as DurableObjectStorage })
246
247 yield* sql`CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)`
248 yield* sql.withTransaction(sql`INSERT INTO test (name) VALUES ('hello')`)
249 const rows = yield* sql`SELECT * FROM test`
250
251 assert.strictEqual(storage.transactionCalls, 1)
252 assert.deepStrictEqual(rows, [{ id: 1, name: "hello" }])
253 assert.strictEqual(hasForbiddenTransactionSql(storage.sql), false)
254 }))
255
256 it.effect("storage-backed failed transactions roll back and re-emit the original failure", () =>
257 Effect.gen(function*() {
258 const storage = new FakeDurableObjectStorage()
259 const sql = yield* makeClient({ storage: storage as unknown as DurableObjectStorage })
260
261 yield* sql`CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)`
262 const error = yield* sql`INSERT INTO test (name) VALUES ('hello')`.pipe(
263 Effect.andThen(Effect.fail("boom")),
264 sql.withTransaction,
265 Effect.flip
266 )
267 const rows = yield* sql`SELECT * FROM test`
268
269 assert.strictEqual(error, "boom")
270 assert.deepStrictEqual(rows, [])
271 assert.strictEqual(storage.rollbackCalls, 1)
272 }))
273
274 it.effect("propagates native rejection after the body succeeds before releasing the connection", () =>
275 Effect.gen(function*() {
276 const storage = new FakeDurableObjectStorage()
277 const commitError = new Error("native completion failed")
278 const bodyCompleted = yield* Deferred.make<void>()
279 let complete!: () => void
280 const completion = new Promise<void>((resolve) => {
281 complete = resolve
282 })
283 const sql = yield* makeClient({
284 storage: {
285 sql: storage.sql,
286 transaction: <T>(body: (txn: { rollback: () => void }) => Promise<T>) =>
287 storage.transaction(async (txn) => {
288 await body(txn)
289 Deferred.doneUnsafe(bodyCompleted, Effect.void)
290 await completion
291 throw commitError
292 })
293 } as unknown as DurableObjectStorage
294 })
295
296 yield* sql`CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)`
297 const transaction = yield* sql`INSERT INTO test (name) VALUES ('hello')`.pipe(
298 sql.withTransaction,
299 Effect.flip,
300 Effect.forkChild
301 )
302 yield* Deferred.await(bodyCompleted)
303 const query = yield* sql`SELECT * FROM test`.pipe(Effect.forkChild({ startImmediately: true }))
304 const transactionBeforeCompletion = transaction.pollUnsafe()
305 const queryBeforeCompletion = query.pollUnsafe()
306 complete()
307
308 const error = yield* Fiber.join(transaction)
309 const rows = yield* Fiber.join(query)
310 assert.isUndefined(transactionBeforeCompletion)
311 assert.isUndefined(queryBeforeCompletion)
312 assert.strictEqual(error._tag, "SqlError")
313 assert.strictEqual(error.reason.cause, commitError)
314 assert.strictEqual(storage.rollbackCalls, 0)
315 assert.strictEqual(storage.transactionCalls, 1)
316 assert.deepStrictEqual(rows, [])
317 }))
318
319 it.effect("storage-backed interrupted transactions roll back before release", () =>
320 Effect.gen(function*() {
321 const storage = new FakeDurableObjectStorage()
322 const sql = yield* makeClient({ storage: storage as unknown as DurableObjectStorage })
323 const inserted = yield* Deferred.make<void>()
324
325 yield* sql`CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)`
326 const fiber = yield* sql`INSERT INTO test (name) VALUES ('hello')`.pipe(
327 Effect.tap(() => Deferred.succeed(inserted, void 0)),
328 Effect.andThen(Effect.never),
329 sql.withTransaction,
330 Effect.forkChild
331 )
332 yield* Deferred.await(inserted)
333 yield* Fiber.interrupt(fiber)
334 const rows = yield* sql`SELECT * FROM test`
335
336 assert.deepStrictEqual(rows, [])
337 assert.strictEqual(storage.rollbackCalls, 1)
338 }))
339
340 it.effect("nested transactions commit inner and outer writes without transaction SQL", () =>
341 Effect.gen(function*() {
342 const storage = new FakeDurableObjectStorage()
343 const sql = yield* makeClient({ storage: storage as unknown as DurableObjectStorage })
344
345 yield* sql`CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)`
346 yield* sql.withTransaction(
347 Effect.gen(function*() {
348 yield* sql`INSERT INTO test (name) VALUES ('outer')`
349 yield* sql.withTransaction(sql`INSERT INTO test (name) VALUES ('inner')`)
350 yield* sql`INSERT INTO test (name) VALUES ('after')`
351 })
352 )
353 const rows = yield* sql`SELECT * FROM test`
354
355 assert.deepStrictEqual(rows, [
356 { id: 1, name: "outer" },
357 { id: 2, name: "inner" },
358 { id: 3, name: "after" }
359 ])
360 assert.strictEqual(storage.transactionCalls, 2)
361 assert.strictEqual(storage.rollbackCalls, 0)
362 assert.deepStrictEqual(storage.snapshots, [])
363 assert.strictEqual(hasForbiddenTransactionSql(storage.sql), false)
364 }))
365
366 it.effect("cross-client nested transactions fail before opening storage or running the body", () =>
367 Effect.gen(function*() {
368 const storageA = new FakeDurableObjectStorage()
369 const storageB = new FakeDurableObjectStorage()
370 const sqlA = yield* makeClient({ storage: storageA as unknown as DurableObjectStorage })
371 const sqlB = yield* makeClient({ storage: storageB as unknown as DurableObjectStorage })
372 let bodyRan = false
373
374 const exit = yield* sqlA.withTransaction(
375 sqlB.withTransaction(Effect.sync(() => {
376 bodyRan = true
377 }))
378 ).pipe(Effect.exit)
379
380 assert.strictEqual(storageA.transactionCalls, 1)
381 assert.strictEqual(storageB.transactionCalls, 0)
382 assert.strictEqual(bodyRan, false)
383 const error = yield* Effect.flip(exit)
384 assert.strictEqual(error.reason._tag, "UnknownError")
385 assert.match(error.message, /Transactions cannot use a connection from a different SQLite client/)
386 }))
387
388 it.effect("caught nested failure rolls back only inner writes and lets the outer transaction commit", () =>
389 Effect.gen(function*() {
390 const storage = new FakeDurableObjectStorage()
391 const sql = yield* makeClient({ storage: storage as unknown as DurableObjectStorage })
392
393 yield* sql`CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)`
394 yield* sql.withTransaction(
395 Effect.gen(function*() {
396 yield* sql`INSERT INTO test (name) VALUES ('outer')`
397 const error = yield* sql`INSERT INTO test (name) VALUES ('inner')`.pipe(
398 Effect.andThen(Effect.fail("inner failure")),
399 sql.withTransaction,
400 Effect.flip
401 )
402 assert.strictEqual(error, "inner failure")
403 assert.deepStrictEqual(yield* sql`SELECT * FROM test`, [{ id: 1, name: "outer" }])
404 yield* sql`INSERT INTO test (name) VALUES ('after')`
405 })
406 )
407 const rows = yield* sql`SELECT * FROM test`
408
409 assert.deepStrictEqual(rows, [{ id: 1, name: "outer" }, { id: 2, name: "after" }])
410 assert.strictEqual(storage.transactionCalls, 2)
411 assert.strictEqual(storage.rollbackCalls, 1)
412 assert.deepStrictEqual(storage.snapshots, [])
413 assert.strictEqual(hasForbiddenTransactionSql(storage.sql), false)
414 }))
415
416 it.effect("uncaught nested failure rolls back the whole transaction", () =>
417 Effect.gen(function*() {
418 const storage = new FakeDurableObjectStorage()
419 const sql = yield* makeClient({ storage: storage as unknown as DurableObjectStorage })
420
421 yield* sql`CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)`
422 const error = yield* sql.withTransaction(
423 Effect.gen(function*() {
424 yield* sql`INSERT INTO test (name) VALUES ('outer')`
425 return yield* sql`INSERT INTO test (name) VALUES ('inner')`.pipe(
426 Effect.andThen(Effect.fail("inner failure")),
427 sql.withTransaction
428 )
429 })
430 ).pipe(Effect.flip)
431 const rows = yield* sql`SELECT * FROM test`
432
433 assert.strictEqual(error, "inner failure")
434 assert.deepStrictEqual(rows, [])
435 assert.strictEqual(storage.transactionCalls, 2)
436 assert.strictEqual(storage.rollbackCalls, 2)
437 assert.deepStrictEqual(storage.snapshots, [])
438 assert.strictEqual(hasForbiddenTransactionSql(storage.sql), false)
439 }))
440
441 it.effect("interrupted nested transactions roll back only the child before the outer transaction continues", () =>
442 Effect.gen(function*() {
443 const storage = new FakeDurableObjectStorage()
444 const sql = yield* makeClient({ storage: storage as unknown as DurableObjectStorage })
445 const inserted = yield* Deferred.make<void>()
446
447 yield* sql`CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)`
448 yield* sql.withTransaction(
449 Effect.gen(function*() {
450 yield* sql`INSERT INTO test (name) VALUES ('outer')`
451 const fiber = yield* sql`INSERT INTO test (name) VALUES ('inner')`.pipe(
452 Effect.tap(() => Deferred.succeed(inserted, void 0)),
453 Effect.andThen(Effect.never),
454 sql.withTransaction,
455 Effect.forkChild
456 )
457 // Surface an early transaction failure rather than waiting for an insert that never happens.
458 yield* Effect.raceFirst(Deferred.await(inserted), Fiber.join(fiber))
459 yield* Fiber.interrupt(fiber)
460 assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(fiber)))
461 assert.strictEqual(storage.rollbackCalls, 1)
462 assert.deepStrictEqual(yield* sql`SELECT * FROM test`, [{ id: 1, name: "outer" }])
463 yield* sql`INSERT INTO test (name) VALUES ('after')`
464 })
465 )
466 const rows = yield* sql`SELECT * FROM test`
467
468 assert.deepStrictEqual(rows, [{ id: 1, name: "outer" }, { id: 2, name: "after" }])
469 assert.strictEqual(storage.transactionCalls, 2)
470 assert.strictEqual(storage.rollbackCalls, 1)
471 assert.deepStrictEqual(storage.snapshots, [])
472 assert.strictEqual(hasForbiddenTransactionSql(storage.sql), false)
473 }))
474
475 it.effect("SqliteMigrator.run works with storage-backed transactions", () =>
476 Effect.gen(function*() {
477 const storage = new FakeDurableObjectStorage()
478 const sql = yield* makeClient({ storage: storage as unknown as DurableObjectStorage })
479
480 const completed = yield* SqliteMigrator.run({
481 loader: SqliteMigrator.fromRecord({
482 "1_create": Effect.gen(function*() {
483 const sql = yield* SqlClient.SqlClient
484 yield* sql`CREATE TABLE migrated (id INTEGER PRIMARY KEY, name TEXT)`
485 yield* sql`INSERT INTO migrated (name) VALUES ('ok')`
486 })
487 })
488 }).pipe(Effect.provideService(SqlClient.SqlClient, sql))
489 const rows = yield* sql`SELECT * FROM migrated`
490 const migrations = yield* sql`SELECT * FROM effect_sql_migrations`
491
492 assert.deepStrictEqual(completed, [[1, "create"]])
493 assert.deepStrictEqual(rows, [{ id: 1, name: "ok" }])
494 assert.deepStrictEqual(migrations, [{ migration_id: 1, name: "create", created_at: "current_timestamp" }])
495 assert.strictEqual(storage.transactionCalls, 1)
496 assert.strictEqual(hasForbiddenTransactionSql(storage.sql), false)
497 }))
498})