1import { ClickhouseClient } from "@effect/sql-clickhouse"
2import { assert, describe, it } from "@effect/vitest"
3import { Effect, Fiber } from "effect"
4import * as Reactivity from "effect/reactivity/Reactivity"
5import * as Statement from "effect/sql/Statement"
6import { TestClock } from "effect/testing"
7import { vi } from "vitest"
8
9let closeCalls = 0
10let connectImmediately = false
11const commandCalls: Array<Record<string, unknown>> = []
12const insertCalls: Array<Record<string, unknown>> = []
13let insertImpl: ((options: Record<string, unknown>) => Promise<unknown>) | undefined
14
15vi.mock("@clickhouse/client", () => ({
16 createClient: () => ({
17 ping: () => connectImmediately ? Promise.resolve({ success: true }) : new Promise(() => {}),
18 query: () => new Promise(() => {}),
19 insert: (options: Record<string, unknown>) => {
20 insertCalls.push(options)
21 return insertImpl ? insertImpl(options) : new Promise(() => {})
22 },
23 command: (options: Record<string, unknown>) => {
24 commandCalls.push(options)
25 return Promise.resolve({})
26 },
27 close: () => {
28 closeCalls++
29 return Promise.resolve()
30 }
31 })
32}))
33
34describe("ClickhouseClient", { concurrent: false }, () => {
35 it("preserves fractional JavaScript numbers in inferred parameters", () => {
36 const sql = Statement.make(Effect.void as any, ClickhouseClient.makeCompiler(), [], undefined)
37 const [query] = sql`SELECT ${1.5}`.compile()
38
39 assert.strictEqual(query, "SELECT {p1: Float64}")
40 })
41
42 it("uses the ClickHouse dialect for dialect-specific fragments", () => {
43 const sql = Statement.make(Effect.void as any, ClickhouseClient.makeCompiler(), [], undefined)
44
45 assert.strictEqual(
46 sql.onDialect({
47 sqlite: () => "sqlite",
48 pg: () => "pg",
49 mysql: () => "mysql",
50 mssql: () => "mssql",
51 clickhouse: () => "clickhouse"
52 }),
53 "clickhouse"
54 )
55 })
56
57 it.effect("closes the client when the connection check times out", () =>
58 Effect.gen(function*() {
59 connectImmediately = false
60 closeCalls = 0
61 const fiber = yield* Effect.forkDetach(
62 ClickhouseClient.make({ url: "http://localhost:8123" }).pipe(Effect.scoped)
63 )
64 yield* Effect.yieldNow
65 yield* TestClock.adjust("5 seconds")
66 const result = fiber.pollUnsafe()
67
68 assert.isDefined(result)
69 assert.strictEqual(closeCalls, 1)
70 }).pipe(Effect.provide(Reactivity.layer)))
71
72 it.effect("parameterizes the query id when cancelling a query", () =>
73 Effect.gen(function*() {
74 connectImmediately = true
75 commandCalls.length = 0
76 const queryId = "id' OR 1 = 1 --"
77 const client = yield* ClickhouseClient.make({ url: "http://localhost:8123" })
78 const fiber = yield* client.withQueryId(client.unsafe("SELECT 1"), queryId).pipe(Effect.forkScoped)
79 yield* Effect.yieldNow
80
81 yield* Fiber.interrupt(fiber)
82
83 assert.deepStrictEqual(commandCalls, [{
84 query: "KILL QUERY WHERE query_id = {queryId:String}",
85 query_params: { queryId }
86 }])
87 }).pipe(
88 Effect.scoped,
89 Effect.provide(Reactivity.layer)
90 ))
91
92 it.effect("parameterizes the query id when cancelling an insert", () =>
93 Effect.gen(function*() {
94 connectImmediately = true
95 commandCalls.length = 0
96 const queryId = "id' OR 1 = 1 --"
97 const client = yield* ClickhouseClient.make({ url: "http://localhost:8123" })
98 const fiber = yield* client.withQueryId(
99 client.insertQuery({ table: "test", values: [] }),
100 queryId
101 ).pipe(Effect.forkScoped)
102 yield* Effect.yieldNow
103
104 yield* Fiber.interrupt(fiber)
105
106 assert.deepStrictEqual(commandCalls, [{
107 query: "KILL QUERY WHERE query_id = {queryId:String}",
108 query_params: { queryId }
109 }])
110 }).pipe(
111 Effect.scoped,
112 Effect.provide(Reactivity.layer)
113 ))
114
115 it.effect("passes column selection to insertQuery", () =>
116 Effect.gen(function*() {
117 connectImmediately = true
118 insertCalls.length = 0
119 insertImpl = () => Promise.resolve({ executed: true, query_id: "" })
120 const client = yield* ClickhouseClient.make({ url: "http://localhost:8123" })
121
122 yield* client.insertQuery({ table: "people", values: [{ name: "Alice" }], columns: ["name"] })
123 yield* client.insertQuery({ table: "people", values: [{ name: "Bob" }], columns: { except: ["id"] } })
124
125 assert.strictEqual(insertCalls.length, 2)
126 assert.strictEqual(insertCalls[0].table, "people")
127 assert.strictEqual(insertCalls[0].format, "JSONEachRow")
128 assert.deepStrictEqual(insertCalls[0].columns, ["name"])
129 assert.deepStrictEqual(insertCalls[1].columns, { except: ["id"] })
130 }).pipe(Effect.provide(Reactivity.layer)))
131})