1/**
2 * ClickHouse driver for Effect SQL, backed by `@clickhouse/client`.
3 *
4 * This module provides both the ClickHouse-specific {@link ClickhouseClient}
5 * service and the generic {@link Client.SqlClient} service. `make` creates a
6 * scoped client, checks the connection with `ping()`, maps ClickHouse errors
7 * to `SqlError`, and aborts in-flight queries when interrupted. The
8 * ClickHouse-specific service adds typed parameters, command execution, insert
9 * queries, query id and settings helpers, a statement compiler, and direct or
10 * config-backed layers.
11 *
12 * @stability unstable
13 * @since 4.0.0
14 */
15import * as Clickhouse from "@clickhouse/client"
16import * as NodeStream from "@effect/platform-node/NodeStream"
17import * as Config from "effect/Config"
18import * as Context from "effect/Context"
19import * as Duration from "effect/Duration"
20import * as Effect from "effect/Effect"
21import * as Fiber from "effect/Fiber"
22import { dual } from "effect/Function"
23import * as Layer from "effect/Layer"
24import * as Reactivity from "effect/reactivity/Reactivity"
25import type * as Scope from "effect/Scope"
26import * as Client from "effect/sql/SqlClient"
27import type { Connection } from "effect/sql/SqlConnection"
28import {
29 AuthenticationError,
30 AuthorizationError,
31 ConnectionError,
32 SqlError,
33 SqlSyntaxError,
34 StatementTimeoutError,
35 UnknownError
36} from "effect/sql/SqlError"
37import * as Statement from "effect/sql/Statement"
38import * as Stream from "effect/Stream"
39import * as Crypto from "node:crypto"
40import type { Readable } from "node:stream"
41
42const ATTR_DB_SYSTEM_NAME = "db.system.name"
43const ATTR_DB_NAMESPACE = "db.namespace"
44
45const clickhouseCodeFromCause = (cause: unknown): number | undefined => {
46 if (typeof cause !== "object" || cause === null || !("code" in cause)) {
47 return undefined
48 }
49 const code = cause.code
50 if (typeof code === "number") {
51 return code
52 }
53 if (typeof code === "string") {
54 const parsed = Number(code)
55 return Number.isNaN(parsed) ? undefined : parsed
56 }
57 return undefined
58}
59
60const clickhouseSyntaxErrorCodes = new Set([36, 60, 62, 242])
61
62const classifyError = (
63 cause: unknown,
64 message: string,
65 operation: string,
66 fallback: "connection" | "unknown" = "unknown"
67) => {
68 const props = { cause, message, operation }
69 const code = clickhouseCodeFromCause(cause)
70 if (code !== undefined) {
71 if (code === 516) {
72 return new AuthenticationError(props)
73 }
74 if (code === 497) {
75 return new AuthorizationError(props)
76 }
77 if (clickhouseSyntaxErrorCodes.has(code)) {
78 return new SqlSyntaxError(props)
79 }
80 if (code === 159 || code === 469) {
81 return new StatementTimeoutError(props)
82 }
83 }
84 return fallback === "connection" ? new ConnectionError(props) : new UnknownError(props)
85}
86
87/**
88 * Unique runtime identifier used to tag `ClickhouseClient` values.
89 *
90 * @stability unstable
91 * @category type IDs
92 * @since 4.0.0
93 */
94export const TypeId: TypeId = "~@effect/sql-clickhouse/ClickhouseClient"
95
96/**
97 * Type-level literal for the `ClickhouseClient` runtime identifier.
98 *
99 * @stability unstable
100 * @category type IDs
101 * @since 4.0.0
102 */
103export type TypeId = "~@effect/sql-clickhouse/ClickhouseClient"
104
105/**
106 * ClickHouse-specific `SqlClient` extension with access to its configuration,
107 * typed parameter fragments, command-mode execution, insert queries, and
108 * per-effect query ID and ClickHouse settings.
109 *
110 * @stability unstable
111 * @category services
112 * @since 4.0.0
113 */
114export interface ClickhouseClient extends Client.SqlClient {
115 readonly [TypeId]: TypeId
116 readonly config: ClickhouseClientConfig
117 readonly param: (dataType: string, value: unknown) => Statement.Fragment
118 readonly asCommand: <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
119 readonly insertQuery: <T = unknown>(options: {
120 readonly table: string
121 readonly values: Clickhouse.InsertValues<Readable, T>
122 readonly format?: Clickhouse.DataFormat
123 readonly columns?: NonNullable<Clickhouse.InsertParams<Readable, T>["columns"]>
124 }) => Effect.Effect<Clickhouse.InsertResult, SqlError>
125 readonly withQueryId: {
126 (queryId: string): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
127 <A, E, R>(effect: Effect.Effect<A, E, R>, queryId: string): Effect.Effect<A, E, R>
128 }
129 readonly withClickhouseSettings: {
130 (
131 settings: NonNullable<Clickhouse.BaseQueryParams["clickhouse_settings"]>
132 ): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
133 <A, E, R>(
134 effect: Effect.Effect<A, E, R>,
135 settings: NonNullable<Clickhouse.BaseQueryParams["clickhouse_settings"]>
136 ): Effect.Effect<A, E, R>
137 }
138}
139
140/**
141 * Service tag for the active ClickHouse SQL client.
142 *
143 * **When to use**
144 *
145 * Use to access or provide a ClickHouse SQL client through the Effect context.
146 *
147 * @stability unstable
148 * @category services
149 * @since 4.0.0
150 */
151export const ClickhouseClient = Context.Service<ClickhouseClient>("@effect/sql-clickhouse/ClickhouseClient")
152
153/**
154 * Configuration for creating a ClickHouse client, combining
155 * `@clickhouse/client` options with optional span attributes and query/result
156 * name transforms.
157 *
158 * @stability unstable
159 * @category models
160 * @since 4.0.0
161 */
162export interface ClickhouseClientConfig extends Clickhouse.ClickHouseClientConfigOptions {
163 readonly spanAttributes?: Record<string, unknown> | undefined
164 readonly transformResultNames?: ((str: string) => string) | undefined
165 readonly transformQueryNames?: ((str: string) => string) | undefined
166}
167
168/**
169 * Creates a scoped `ClickhouseClient`, verifies connectivity with `ping()`,
170 * closes the underlying client when the scope ends, maps ClickHouse failures
171 * to `SqlError`, and aborts plus kills in-flight queries when interrupted.
172 *
173 * @stability unstable
174 * @category constructors
175 * @since 4.0.0
176 */
177export const make = (
178 options: ClickhouseClientConfig
179): Effect.Effect<ClickhouseClient, SqlError, Scope.Scope | Reactivity.Reactivity> =>
180 Effect.gen(function*() {
181 const compiler = makeCompiler(options.transformQueryNames)
182 const transformRows = options.transformResultNames
183 ? Statement.defaultTransforms(options.transformResultNames).array
184 : undefined
185
186 const client = yield* Effect.acquireRelease(
187 Effect.sync(() => Clickhouse.createClient(options)),
188 (client) => Effect.promise(() => client.close())
189 )
190
191 yield* Effect.tryPromise({
192 try: async () => {
193 const result = await client.ping()
194 if (!result.success) {
195 throw result.error
196 }
197 return result
198 },
199 catch: (cause) =>
200 new SqlError({ reason: classifyError(cause, "ClickhouseClient: Failed to connect", "connect", "connection") })
201 }).pipe(
202 Effect.timeoutOrElse({
203 duration: Duration.seconds(5),
204 orElse: () =>
205 Effect.fail(
206 new SqlError({
207 reason: new ConnectionError({
208 message: "ClickhouseClient: Connection timeout",
209 cause: new Error("connection timeout"),
210 operation: "connect"
211 })
212 })
213 )
214 })
215 )
216
217 class ConnectionImpl implements Connection {
218 private conn: Clickhouse.ClickHouseClient
219 constructor(conn: Clickhouse.ClickHouseClient) {
220 this.conn = conn
221 }
222
223 private runRaw(sql: string, params: ReadonlyArray<unknown>, format: Clickhouse.DataFormat = "JSON") {
224 const paramsObj: Record<string, unknown> = {}
225 for (let i = 0; i < params.length; i++) {
226 paramsObj[`p${i + 1}`] = params[i]
227 }
228 return Effect.withFiber<Clickhouse.ResultSet<"JSON"> | Clickhouse.CommandResult, SqlError>((fiber) => {
229 const method = fiber.getRef(ClientMethod)
230 return Effect.callback<Clickhouse.ResultSet<"JSON"> | Clickhouse.CommandResult, SqlError>((resume) => {
231 const queryId = fiber.getRef(QueryId) ?? Crypto.randomUUID()
232 const settings = fiber.getRef(ClickhouseSettings)
233 const controller = new AbortController()
234 if (method === "command") {
235 this.conn.command({
236 query: sql,
237 query_params: paramsObj,
238 abort_signal: controller.signal,
239 query_id: queryId,
240 clickhouse_settings: settings
241 }).then(
242 (result) => resume(Effect.succeed(result)),
243 (cause) =>
244 resume(
245 Effect.fail(
246 new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") })
247 )
248 )
249 )
250 } else {
251 this.conn.query({
252 query: sql,
253 query_params: paramsObj,
254 abort_signal: controller.signal,
255 query_id: queryId,
256 clickhouse_settings: settings,
257 format
258 }).then(
259 (result) => resume(Effect.succeed(result)),
260 (cause) =>
261 resume(
262 Effect.fail(
263 new SqlError({ reason: classifyError(cause, "Failed to execute statement", "execute") })
264 )
265 )
266 )
267 }
268 return Effect.suspend(() => {
269 controller.abort()
270 return Effect.promise(() =>
271 this.conn.command({
272 query: "KILL QUERY WHERE query_id = {queryId:String}",
273 query_params: { queryId }
274 })
275 )
276 })
277 })
278 })
279 }
280
281 private run(sql: string, params: ReadonlyArray<unknown>, format?: Clickhouse.DataFormat) {
282 return this.runRaw(sql, params, format).pipe(
283 Effect.flatMap((result) => {
284 if ("json" in result) {
285 return Effect.tryPromise({
286 try: () => result.json().then((result) => "data" in result ? result.data : result as any),
287 catch: (cause) =>
288 new SqlError({ reason: classifyError(cause, "Failed to parse result", "parseResult") })
289 })
290 }
291 return Effect.succeed([])
292 })
293 )
294 }
295
296 execute(
297 sql: string,
298 params: ReadonlyArray<unknown>,
299 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
300 ) {
301 return transformRows
302 ? Effect.map(this.run(sql, params), transformRows)
303 : this.run(sql, params)
304 }
305 executeRaw(sql: string, params: ReadonlyArray<unknown>) {
306 return this.runRaw(sql, params)
307 }
308 executeValues(sql: string, params: ReadonlyArray<unknown>) {
309 return this.run(sql, params, "JSONCompact")
310 }
311 executeValuesUnprepared(sql: string, params: ReadonlyArray<unknown>) {
312 return this.executeValues(sql, params)
313 }
314 executeUnprepared(sql: string, params: ReadonlyArray<unknown>, transformRows?: any) {
315 return this.execute(sql, params, transformRows)
316 }
317 executeStream(
318 sql: string,
319 params: ReadonlyArray<unknown>,
320 transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
321 ) {
322 return this.runRaw(sql, params, "JSONEachRow").pipe(
323 Effect.map((result) => {
324 if (!("stream" in result)) {
325 return Stream.empty
326 }
327 return NodeStream.fromReadable<ReadonlyArray<Clickhouse.Row<any, "JSONEachRow">>, SqlError>({
328 evaluate: () => result.stream() as any,
329 onError: (cause) => new SqlError({ reason: classifyError(cause, "Failed to execute stream", "stream") })
330 })
331 }),
332 Stream.unwrap,
333 Stream.chunks,
334 Stream.mapEffect((chunk) => {
335 const promises: Array<Promise<any>> = []
336 for (const rows of chunk) {
337 for (const row of rows) {
338 promises.push(row.json())
339 }
340 }
341 return Effect.tryPromise({
342 try: () => Promise.all(promises).then((rows) => transformRows ? transformRows(rows) : rows),
343 catch: (cause) => new SqlError({ reason: classifyError(cause, "Failed to parse row", "parseRow") })
344 })
345 }),
346 Stream.flattenIterable
347 )
348 }
349 }
350
351 const connection = new ConnectionImpl(client)
352
353 return Object.assign(
354 yield* Client.make({
355 acquirer: Effect.succeed(connection),
356 compiler,
357 spanAttributes: [
358 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
359 [ATTR_DB_SYSTEM_NAME, "clickhouse"],
360 [ATTR_DB_NAMESPACE, options.database ?? "default"]
361 ],
362 beginTransaction: "BEGIN TRANSACTION",
363 transformRows
364 }),
365 {
366 [TypeId]: TypeId as TypeId,
367 config: options,
368 param(dataType: string, value: unknown) {
369 return Statement.fragment([clickhouseParam(dataType, value)])
370 },
371 asCommand<A, E, R>(effect: Effect.Effect<A, E, R>) {
372 return Effect.provideService(effect, ClientMethod, "command")
373 },
374 insertQuery<T = unknown>(options: {
375 readonly table: string
376 readonly values: Clickhouse.InsertValues<Readable, T>
377 readonly format?: Clickhouse.DataFormat
378 readonly columns?: NonNullable<Clickhouse.InsertParams<Readable, T>["columns"]>
379 }) {
380 return Effect.callback<Clickhouse.InsertResult, SqlError>((resume) => {
381 const fiber = Fiber.getCurrent()!
382 const queryId = fiber.getRef(QueryId) ?? Crypto.randomUUID()
383 const settings = fiber.getRef(ClickhouseSettings)
384 const controller = new AbortController()
385 client.insert({
386 format: "JSONEachRow",
387 ...options,
388 abort_signal: controller.signal,
389 query_id: queryId,
390 clickhouse_settings: settings
391 }).then(
392 (result) => resume(Effect.succeed(result)),
393 (cause) =>
394 resume(Effect.fail(new SqlError({ reason: classifyError(cause, "Failed to insert data", "insert") })))
395 )
396 return Effect.suspend(() => {
397 controller.abort()
398 return Effect.promise(() =>
399 client.command({
400 query: "KILL QUERY WHERE query_id = {queryId:String}",
401 query_params: { queryId }
402 })
403 )
404 })
405 })
406 },
407 withQueryId: dual(
408 2,
409 <A, E, R>(effect: Effect.Effect<A, E, R>, queryId: string) => Effect.provideService(effect, QueryId, queryId)
410 ),
411 withClickhouseSettings: dual(
412 2,
413 <A, E, R>(
414 effect: Effect.Effect<A, E, R>,
415 settings: NonNullable<Clickhouse.BaseQueryParams["clickhouse_settings"]>
416 ) => Effect.provideService(effect, ClickhouseSettings, settings)
417 )
418 }
419 )
420 })
421
422/**
423 * Fiber reference read by the low-level ClickHouse connection to choose query
424 * or command execution for statements; defaults to `query`.
425 *
426 * @stability unstable
427 * @category services
428 * @since 4.0.0
429 */
430export const ClientMethod = Context.Reference<"query" | "command" | "insert">(
431 "@effect/sql-clickhouse/ClickhouseClient/ClientMethod",
432 {
433 defaultValue: () => "query"
434 }
435)
436
437/**
438 * Fiber reference for the ClickHouse `query_id` applied to queries and
439 * inserts; a random UUID is generated when no query ID is set.
440 *
441 * @stability unstable
442 * @category services
443 * @since 4.0.0
444 */
445export const QueryId = Context.Reference<string | undefined>(
446 "@effect/sql-clickhouse/ClickhouseClient/QueryId",
447 { defaultValue: () => undefined }
448)
449
450/**
451 * Fiber reference containing ClickHouse settings to attach to queries,
452 * commands, and inserts.
453 *
454 * @stability unstable
455 * @category services
456 * @since 4.0.0
457 */
458export const ClickhouseSettings: Context.Reference<
459 NonNullable<Clickhouse.BaseQueryParams["clickhouse_settings"]>
460> = Context.Reference("@effect/sql-clickhouse/ClickhouseClient/ClickhouseSettings", {
461 defaultValue: () => ({})
462})
463
464/**
465 * Provides both `ClickhouseClient` and generic `SqlClient` services from a
466 * `Config`-backed ClickHouse client configuration.
467 *
468 * @stability unstable
469 * @category layers
470 * @since 4.0.0
471 */
472export const layerConfig: (
473 config: Config.Wrap<ClickhouseClientConfig>
474) => Layer.Layer<ClickhouseClient | Client.SqlClient, Config.ConfigError | SqlError> = (
475 config: Config.Wrap<ClickhouseClientConfig>
476): Layer.Layer<ClickhouseClient | Client.SqlClient, Config.ConfigError | SqlError> =>
477 Layer.effectContext(
478 Config.unwrap(config).pipe(
479 Effect.flatMap(make),
480 Effect.map((client) =>
481 Context.make(ClickhouseClient, client).pipe(
482 Context.add(Client.SqlClient, client)
483 )
484 )
485 )
486 ).pipe(Layer.provide(Reactivity.layer))
487
488/**
489 * Provides both `ClickhouseClient` and generic `SqlClient` services from a
490 * ClickHouse client configuration.
491 *
492 * @stability unstable
493 * @category layers
494 * @since 4.0.0
495 */
496export const layer = (
497 config: ClickhouseClientConfig
498): Layer.Layer<ClickhouseClient | Client.SqlClient, Config.ConfigError | SqlError> =>
499 Layer.effectContext(
500 Effect.map(make(config), (client) =>
501 Context.make(ClickhouseClient, client).pipe(
502 Context.add(Client.SqlClient, client)
503 ))
504 ).pipe(Layer.provide(Reactivity.layer))
505
506const typeFromUnknown = (value: unknown): string => {
507 if (Statement.isFragment(value)) {
508 return typeFromUnknown(value.segments[0])
509 } else if (isClickhouseParam(value)) {
510 return value.paramA
511 } else if (Array.isArray(value)) {
512 return `Array(${typeFromUnknown(value[0])})`
513 }
514 switch (typeof value) {
515 case "number":
516 return "Float64"
517 case "bigint":
518 return "Int64"
519 case "boolean":
520 return "Bool"
521 case "object":
522 if (value instanceof Date) {
523 return "DateTime()"
524 }
525 return "String"
526 default:
527 return "String"
528 }
529}
530
531/**
532 * Creates the SQL statement compiler for ClickHouse, emitting typed
533 * `{pN: Type}` placeholders and escaping identifiers with an optional query
534 * name transform.
535 *
536 * @stability unstable
537 * @category constructors
538 * @since 4.0.0
539 */
540export const makeCompiler = (transform?: (_: string) => string) =>
541 Statement.makeCompiler<ClickhouseCustom>({
542 dialect: "clickhouse",
543 placeholder(i, u) {
544 return `{p${i}: ${typeFromUnknown(u)}}`
545 },
546 onIdentifier: transform ?
547 function(value, withoutTransform) {
548 return withoutTransform ? escape(value) : escape(transform(value))
549 } :
550 escape,
551 onRecordUpdate() {
552 return ["", []]
553 },
554 onCustom(type, placeholder) {
555 return [placeholder(type), [type.paramB]]
556 }
557 })
558
559// compiler helpers
560
561const escape = Statement.defaultEscape("\"")
562
563/**
564 * Custom SQL fragment type used for ClickHouse typed parameters created by
565 * `ClickhouseClient.param`.
566 *
567 * @stability unstable
568 * @category models
569 * @since 4.0.0
570 */
571export type ClickhouseCustom = ClickhouseParam
572
573/**
574 * @category models
575 * @since 4.0.0
576 */
577interface ClickhouseParam extends Statement.Custom<"ClickhouseParam", string, unknown> {}
578
579const clickhouseParam = Statement.custom<ClickhouseParam>("ClickhouseParam")
580const isClickhouseParam = Statement.isCustom<ClickhouseParam>("ClickhouseParam")