1import { assert, describe, it } from "@effect/vitest"
2import { Deferred, Effect, Exit, Fiber, Option } from "effect"
3import { Reactivity } from "effect/reactivity"
4import { TestClock } from "effect/testing"
5import { vi } from "vitest"
6
7const state = vi.hoisted(() => ({ vfsCloseCalls: 0 }))
8
9vi.mock("@effect/wa-sqlite/dist/wa-sqlite.mjs", () => ({ default: async () => ({}) }))
10vi.mock("@effect/wa-sqlite/src/examples/AccessHandlePoolVFS.js", () => ({
11 AccessHandlePoolVFS: {
12 create: async () => ({
13 close: async () => {
14 state.vfsCloseCalls++
15 }
16 })
17 }
18}))
19vi.mock("@effect/wa-sqlite", () => ({
20 SQLITE_ROW: 100,
21 Factory: () => ({
22 bind_collection() {},
23 close() {},
24 column_names: (statement: { readonly columns: ReadonlyArray<string> }) => statement.columns,
25 open_v2: () => 1,
26 row: (statement: { readonly index: number; readonly rows: ReadonlyArray<ReadonlyArray<number>> }) =>
27 statement.rows[statement.index],
28 *statements(_db: number, sql: string) {
29 if (sql !== "SELECT 1 AS first; SELECT 2 AS second, 3 AS third") {
30 throw new Error(`Unexpected SQL: ${sql}`)
31 }
32 yield { columns: ["first"], index: -1, rows: [[1]] }
33 yield { columns: ["second", "third"], index: -1, rows: [[2, 3]] }
34 },
35 step(statement: { index: number; readonly rows: ReadonlyArray<ReadonlyArray<number>> }) {
36 return ++statement.index < statement.rows.length ? 100 : 101
37 },
38 vfs_register() {}
39 })
40}))
41
42import { OpfsWorker, SqliteClient } from "@effect/sql-sqlite-wasm"
43
44class FakePort extends EventTarget {
45 close() {}
46
47 postMessage(message: ReadonlyArray<unknown>): void {
48 if (message[0] === "ready") {
49 queueMicrotask(() => this.dispatchEvent(new MessageEvent("message", { data: ["close"] })))
50 }
51 }
52}
53
54class FakeWorker extends EventTarget {
55 onerror: unknown = null
56
57 constructor(readonly queryPosted: Deferred.Deferred<void>) {
58 super()
59 }
60
61 override addEventListener(
62 type: string,
63 listener: EventListenerOrEventListenerObject | null,
64 options?: unknown
65 ): void {
66 super.addEventListener(type, listener, options as boolean)
67 if (type === "message") {
68 queueMicrotask(() => this.dispatchEvent(new MessageEvent("message", { data: ["ready"] })))
69 }
70 }
71
72 postMessage(message: ReadonlyArray<unknown>): void {
73 if (typeof message[0] === "number") {
74 Effect.runFork(Deferred.succeed(this.queryPosted, undefined))
75 }
76 }
77}
78
79// A MessagePort only delivers queued messages once `start()` is called.
80class GatedPort extends EventTarget {
81 private queued: Array<ReadonlyArray<unknown>> = []
82 private started = false
83
84 start(): void {
85 this.started = true
86 for (const data of this.queued.splice(0)) {
87 this.dispatchEvent(new MessageEvent("message", { data }))
88 }
89 }
90
91 deliver(data: ReadonlyArray<unknown>): void {
92 if (this.started) this.dispatchEvent(new MessageEvent("message", { data }))
93 else this.queued.push(data)
94 }
95
96 postMessage(): void {}
97 close(): void {}
98}
99
100const settled = <A, E>(effect: Effect.Effect<A, E>) =>
101 Effect.gen(function*() {
102 const fiber = yield* effect.pipe(Effect.exit, Effect.timeoutOption("100 millis"), Effect.forkChild)
103 yield* TestClock.adjust("100 millis")
104 return yield* Fiber.join(fiber)
105 })
106
107describe("Client", () => {
108 it.effect("should work", () => Effect.void)
109
110 it.effect("closes the OPFS VFS when the worker loop closes", () =>
111 Effect.gen(function*() {
112 yield* OpfsWorker.run({ port: new FakePort(), dbName: "test.db" })
113 assert.strictEqual(state.vfsCloseCalls, 1)
114 }))
115
116 it.effect("keeps columns scoped to each statement", () =>
117 Effect.gen(function*() {
118 const channel = new MessageChannel()
119 yield* Effect.addFinalizer(() => Effect.sync(() => channel.port2.close()))
120 yield* Effect.forkChild(OpfsWorker.run({ port: channel.port1, dbName: "test.db" }))
121 const sql = yield* SqliteClient.make({ worker: Effect.succeed(channel.port2) })
122
123 const rows = yield* sql`SELECT 1 AS first; SELECT 2 AS second, 3 AS third`
124
125 assert.deepStrictEqual(rows, [{ first: 1 }, { second: 2, third: 3 }])
126 }).pipe(Effect.provide(Reactivity.layer)))
127
128 it.effect("settles an in-flight query when the worker errors and reconnects", () =>
129 Effect.gen(function*() {
130 const queryPosted = yield* Deferred.make<void>()
131 const worker = new FakeWorker(queryPosted)
132 const sql = yield* SqliteClient.make({ worker: Effect.succeed(worker as unknown as Worker) })
133 const fiber = yield* Effect.forkChild(sql`SELECT 1`)
134
135 yield* Deferred.await(queryPosted)
136 worker.dispatchEvent(new Event("error"))
137
138 const joinFiber = yield* Fiber.join(fiber).pipe(Effect.exit, Effect.timeoutOption("100 millis"), Effect.forkChild)
139 yield* TestClock.adjust("100 millis")
140 const result = yield* Fiber.join(joinFiber)
141 assert(Option.isSome(result), "the request remained pending after worker replacement")
142 assert(Exit.isFailure(result.value))
143 }).pipe(Effect.provide(Reactivity.layer)))
144
145 it.effect("starts a MessagePort transport so the ready message arrives", () =>
146 Effect.gen(function*() {
147 const port = new GatedPort()
148 port.deliver(["ready", undefined, undefined])
149 const fiber = yield* Effect.forkChild(SqliteClient.make({ worker: Effect.succeed(port as unknown as Worker) }))
150 const result = yield* settled(Fiber.join(fiber))
151 assert(Option.isSome(result), "the ready message was never delivered")
152 assert(Exit.isSuccess(result.value))
153 }).pipe(Effect.provide(Reactivity.layer)))
154
155 it.effect("starts a MessagePort port so worker requests are consumed", () =>
156 Effect.gen(function*() {
157 const port = new GatedPort()
158 port.deliver(["close"])
159 const fiber = yield* Effect.forkChild(OpfsWorker.run({ port, dbName: "test.db" }))
160 const result = yield* settled(Fiber.join(fiber))
161 assert(Option.isSome(result), "the queued request was never consumed")
162 assert(Exit.isSuccess(result.value))
163 }))
164})