1import { MssqlClient, Procedure } from "@effect/sql-mssql"
2import { assert, describe, expect, it } from "@effect/vitest"
3import { Deferred, Effect, Fiber } from "effect"
4import * as Reactivity from "effect/reactivity/Reactivity"
5import * as Statement from "effect/sql/Statement"
6import type { EventEmitter } from "node:events"
7import type * as Tedious from "tedious"
8import { vi } from "vitest"
9
10interface MockConnection extends EventEmitter {
11 closed: boolean
12 closedSignal: Deferred.Deferred<void>
13 cancellations: number
14 queries: Array<string>
15 close(): void
16}
17
18const state = vi.hoisted(() => ({
19 cancelCalls: 0,
20 completeRequests: true,
21 type: {},
22 connections: [] as Array<MockConnection>,
23 pendingStarted: undefined as Deferred.Deferred<void> | undefined,
24 errorAfterConnect: false,
25 emissionError: undefined as unknown
26}))
27
28vi.mock("tedious", async (importOriginal) => {
29 const original = await importOriginal<typeof Tedious>()
30 const { EventEmitter } = await import("node:events")
31
32 class MockRequest {
33 readonly listeners: Record<string, (...args: Array<any>) => void> = {}
34
35 constructor(
36 readonly sql: string,
37 readonly callback: (cause: unknown, rowCount: number, rows: ReadonlyArray<any>) => void
38 ) {}
39
40 addParameter() {}
41 addOutputParameter() {}
42 on(event: string, listener: (...args: Array<any>) => void) {
43 this.listeners[event] = listener
44 }
45 }
46
47 class MockConnection extends EventEmitter {
48 closed = false
49 readonly closedSignal = Deferred.makeUnsafe<void>()
50 pending: MockRequest | undefined
51 cancellations = 0
52 readonly queries: Array<string> = []
53
54 constructor() {
55 super()
56 state.connections.push(this)
57 }
58
59 connect(callback: (cause: unknown) => void) {
60 // Resume acquisition, then emit in the same driver turn, before a forked
61 // pool listener can run. Do not install a test-only error listener.
62 Fiber.getCurrent()!.currentDispatcher.scheduleTask(() => {
63 callback(null)
64 if (state.errorAfterConnect) {
65 state.errorAfterConnect = false
66 this.closed = true
67 try {
68 this.emit("error", new Error("Connection lost - read ETIMEDOUT"))
69 } catch (error) {
70 state.emissionError = error
71 }
72 }
73 }, 0)
74 }
75
76 close() {
77 if (!this.closed) {
78 this.closed = true
79 this.emit("end")
80 }
81 Effect.runSync(Deferred.succeed(this.closedSignal, undefined))
82 }
83
84 cancel() {
85 // MssqlClient also calls cancel before each request. Count only an actual
86 // in-flight cancellation, so that cannot satisfy the interruption assertion.
87 state.cancelCalls++
88 if (this.pending) {
89 this.cancellations++
90 this.pending = undefined
91 }
92 }
93
94 execSql(request: MockRequest) {
95 this.queries.push(request.sql)
96 if (this.closed) {
97 request.callback(new Error("Requests can only be made in the LoggedIn state, not the Final state"), 0, [])
98 } else if (request.sql === "SELECT pending" || !state.completeRequests) {
99 this.pending = request
100 if (state.pendingStarted) {
101 Effect.runSync(Deferred.succeed(state.pendingStarted, undefined))
102 }
103 } else {
104 request.callback(null, 1, [[{ metadata: { colName: "value" }, value: 42 }]])
105 }
106 }
107 callProcedure(request: MockRequest) {
108 request.listeners.returnValue("answer", 42)
109 request.callback(null, 0, [])
110 }
111
112 beginTransaction(callback: (cause: unknown) => void) {
113 callback(null)
114 }
115
116 commitTransaction(callback: (cause: unknown) => void) {
117 callback(null)
118 }
119
120 saveTransaction(callback: (cause: unknown) => void) {
121 callback(null)
122 }
123
124 rollbackTransaction(callback: (cause: unknown) => void) {
125 callback(null)
126 }
127 }
128
129 return {
130 ...original,
131 Connection: MockConnection,
132 Request: MockRequest
133 }
134})
135
136const sql = Statement.make(Effect.void as any, MssqlClient.makeCompiler(), [], undefined)
137
138describe("mssql", () => {
139 it("preserves fractional JavaScript numbers with the default parameter mapping", () => {
140 const value = MssqlClient.defaultParameterTypes.number.validate(1.5, undefined)
141
142 expect(value).toBe(1.5)
143 })
144
145 it("preserves Unicode JavaScript strings with the default parameter mapping", () => {
146 const value = MssqlClient.defaultParameterTypes.string.validate("lambda: \u03bb", undefined)
147
148 expect(value).toBe("lambda: \u03bb")
149 })
150
151 it("insert helper", () => {
152 const [query, params] = sql`INSERT INTO ${sql("people")} ${sql.insert({ name: "Tim", age: 10 })}`.compile()
153 expect(query).toEqual(
154 `INSERT INTO [people] ([name],[age]) VALUES (@1,@2)`
155 )
156 expect(params).toEqual(["Tim", 10])
157 })
158
159 it("insert helper returning", () => {
160 const [query, params] = sql`INSERT INTO ${sql("people")} ${sql.insert({ name: "Tim", age: 10 }).returning("*")}`
161 .compile()
162 expect(query).toEqual(
163 `INSERT INTO [people] ([name],[age]) OUTPUT INSERTED.* VALUES (@1,@2)`
164 )
165 expect(params).toEqual(["Tim", 10])
166 })
167
168 it("update helper", () => {
169 const [query, params] = sql`UPDATE people SET name = data.name ${
170 sql.updateValues(
171 [{ name: "Tim" }, { name: "John" }],
172 "data"
173 )
174 }`.compile()
175 expect(query).toEqual(
176 `UPDATE people SET name = data.name FROM (values (@1),(@2)) AS data([name])`
177 )
178 expect(params).toEqual(["Tim", "John"])
179 })
180
181 it("update helper returning", () => {
182 const [query, params] = sql`UPDATE people SET name = data.name ${
183 sql.updateValues(
184 [{ name: "Tim" }, { name: "John" }],
185 "data"
186 ).returning("*")
187 }`.compile()
188 expect(query).toEqual(
189 `UPDATE people SET name = data.name OUTPUT INSERTED.* FROM (values (@1),(@2)) AS data([name])`
190 )
191 expect(params).toEqual(["Tim", "John"])
192 })
193
194 it("update single helper returning", () => {
195 const [query, params] = sql`UPDATE people SET ${sql.update({ name: "Tim" }).returning("*")}`
196 .compile()
197 expect(query).toEqual(
198 `UPDATE people SET [name] = @1 OUTPUT INSERTED.*`
199 )
200 expect(params).toEqual(["Tim"])
201 })
202
203 for (
204 const [name, returning] of [
205 ["Identifier", sql("INSERTED.name")],
206 ["wrapped Identifier", sql`${sql("INSERTED.name")}`]
207 ] as const
208 ) {
209 it(`insert helper returning ${name}`, () => {
210 const result = sql`INSERT INTO ${sql("people")} ${sql.insert({ name: "Rowan", age: 10 }).returning(returning)}`
211 .compile()
212
213 assert.deepStrictEqual(result, [
214 `INSERT INTO [people] ([name],[age]) OUTPUT [INSERTED].[name] VALUES (@1,@2)`,
215 ["Rowan", 10]
216 ])
217 })
218
219 it(`update helper returning ${name}`, () => {
220 const result = sql`UPDATE people SET ${sql.update({ name: "Rowan" }).returning(returning)}`.compile()
221
222 assert.deepStrictEqual(result, [
223 `UPDATE people SET [name] = @1 OUTPUT [INSERTED].[name]`,
224 ["Rowan"]
225 ])
226 })
227
228 it(`updateValues helper returning ${name}`, () => {
229 const result = sql`UPDATE people SET name = data.name ${
230 sql.updateValues([{ name: "Rowan" }, { name: "Mira" }], "data").returning(returning)
231 }`.compile()
232
233 assert.deepStrictEqual(result, [
234 `UPDATE people SET name = data.name OUTPUT [INSERTED].[name] FROM (values (@1),(@2)) AS data([name])`,
235 ["Rowan", "Mira"]
236 ])
237 })
238 }
239
240 it("array helper", () => {
241 const [query, params] = sql`SELECT * FROM ${sql("people")} WHERE id IN ${sql.in([1, 2, "string"])}`.compile()
242 expect(query).toEqual(`SELECT * FROM [people] WHERE id IN (@1,@2,@3)`)
243 expect(params).toEqual([1, 2, "string"])
244 })
245
246 // it("param types", () => {
247 // const [query, params] = sql`SELECT * FROM ${sql("people")} WHERE id = ${
248 // sql.param(
249 // MssqlTypes.BigInt,
250 // 1
251 // )
252 // }`.compile()
253 // expect(query).toEqual(`SELECT * FROM [people] WHERE id = @1`)
254 // expect(isCustom("MssqlParam")(params[0])).toEqual(true)
255 // const param = params[0] as unknown as Custom<
256 // "MsSqlParam",
257 // any,
258 // any,
259 // any
260 // >
261 // expect(param.i0).toEqual(MssqlTypes.BigInt)
262 // expect(param.i1).toEqual(1)
263 // expect(param.i2).toEqual({})
264 // })
265
266 it("escape [", () => {
267 const [query] = sql`SELECT * FROM ${sql("peo[]ple.te[st]ing")}`.compile()
268 expect(query).toEqual(`SELECT * FROM [peo[]]ple].[te[st]]ing]`)
269 })
270
271 it.effect("returns stored procedure output parameters under output", () =>
272 Effect.gen(function*() {
273 const client = yield* MssqlClient.make({ server: "localhost" })
274 const definition = Procedure.outputParam<number>()("answer", state.type as any)(Procedure.make("get_answer"))
275 const result = yield* client.call(Procedure.compile(definition)({}))
276
277 assert.deepStrictEqual(result, { output: { answer: 42 }, rows: [] })
278 }).pipe(
279 Effect.scoped,
280 Effect.provide(Reactivity.layer)
281 ))
282
283 it.effect("cancels an in-flight Tedious request when interrupted", () =>
284 Effect.gen(function*() {
285 state.cancelCalls = 0
286 state.completeRequests = true
287 const client = yield* MssqlClient.make({ server: "localhost" })
288 state.completeRequests = false
289 const fiber = yield* Effect.forkChild(client`WAITFOR DELAY '00:01:00'`)
290 yield* Effect.yieldNow
291 const callsBeforeInterrupt = state.cancelCalls
292 yield* Fiber.interrupt(fiber)
293
294 assert.strictEqual(state.cancelCalls, callsBeforeInterrupt + 1)
295 }).pipe(
296 Effect.scoped,
297 Effect.provide(Reactivity.layer)
298 ))
299})
300
301const reset = () => {
302 state.cancelCalls = 0
303 state.completeRequests = true
304 state.connections = []
305 state.pendingStarted = Deferred.makeUnsafe<void>()
306 state.errorAfterConnect = false
307 state.emissionError = undefined
308}
309
310const makeClient = MssqlClient.make({
311 server: "localhost",
312 minConnections: 0,
313 maxConnections: 1,
314 connectionTTL: "1 hour"
315})
316
317const bounded = <A, E, R>(effect: Effect.Effect<A, E, R>) => effect.pipe(Effect.timeout("2 seconds"))
318
319describe("MssqlClient connection lifecycle", { concurrent: false }, () => {
320 it.live("replaces a connection that ends without error after query cancellation", () =>
321 Effect.gen(function*() {
322 reset()
323 const client = yield* bounded(makeClient)
324 const connection = state.connections[0]
325 const query = yield* Effect.forkScoped(client.unsafe("SELECT pending"))
326 yield* bounded(Deferred.await(state.pendingStarted!))
327 yield* bounded(Fiber.interrupt(query))
328 assert.strictEqual(connection.cancellations, 1)
329
330 // Model tedious' cancel timeout: Final state and end, with no error event.
331 connection.close()
332 yield* Effect.yieldNow
333 const rows = yield* bounded(client.unsafe("SELECT recovered"))
334 assert.deepStrictEqual(rows, [{ value: 42 }])
335 assert.strictEqual(state.connections.length, 2)
336 assert.isFalse(connection.queries.includes("SELECT recovered"))
337 assert.isTrue(state.connections[1].queries.includes("SELECT recovered"))
338 }).pipe(Effect.scoped, Effect.provide(Reactivity.layer)))
339
340 it.live("handles an error immediately after the connect callback and does not reuse the failed connection", () =>
341 Effect.gen(function*() {
342 reset()
343 const client = yield* bounded(makeClient)
344 const original = state.connections[0]
345 yield* Effect.yieldNow
346
347 // Retire the healthy connection through a normal driver error, then
348 // inject the callback race on the next acquisition of this same pool.
349 state.errorAfterConnect = true
350 original.emit("error", new Error("Connection lost"))
351 yield* bounded(Deferred.await(original.closedSignal))
352 yield* bounded(Effect.result(client.unsafe("SELECT during_race")))
353 assert.isUndefined(state.emissionError, "the driver's error emission must not throw")
354
355 const failed = state.connections[1]
356 assert.isTrue(failed.closed)
357 const rows = yield* bounded(client.unsafe("SELECT recovered"))
358 assert.deepStrictEqual(rows, [{ value: 42 }])
359 assert.isFalse(failed.queries.includes("SELECT recovered"))
360 assert.isTrue(state.connections.slice(2).some((connection) => connection.queries.includes("SELECT recovered")))
361 }).pipe(Effect.scoped, Effect.provide(Reactivity.layer)))
362})