diff --git a/.changeset/clickhouse-http-and-spans.md b/.changeset/clickhouse-http-and-spans.md new file mode 100644 index 0000000..184802e --- /dev/null +++ b/.changeset/clickhouse-http-and-spans.md @@ -0,0 +1,9 @@ +--- +"@maple-dev/effect-orm": minor +--- + +A ClickHouse client for Cloudflare Workers and any `fetch` runtime, and OpenTelemetry-shaped spans for every statement. + +- `@maple-dev/effect-orm/clickhouse-http` is a ClickHouse `SqlClient` over Effect's `HttpClient`, so `Database` runs where `@effect/sql-clickhouse` (Node only) cannot. Interrupting a statement aborts its request and sends `KILL QUERY` for its `query_id`; failures carry the same `SqlError` reasons as Effect's driver; credentials travel in a header; `asCommand`, `withQueryId` and `withSettings` work as there. +- `Database` opens one client span per statement, named and attributed per the OpenTelemetry database conventions: `db.system.name`, `db.operation.name`, `db.collection.name`, `db.query.summary`, `db.response.returned_rows` and `error.type`. `withQuerySummary(name)` names the statements an effect runs. +- `CompiledQuery.collection` is the table a statement targets: a SELECT's `FROM` table, or the table a write changes. diff --git a/README.md b/README.md index 8d13f2e..20e0b25 100644 --- a/README.md +++ b/README.md @@ -228,6 +228,7 @@ regressions live in [`src/docs-examples.test.ts`](./src/docs-examples.test.ts). | `@maple-dev/effect-orm/schema` | Migration tooling over `table` values: DDL rendering, snapshots, the schema diff. Pure. | | `@maple-dev/effect-orm/kit`, `/migrate` | `effect-orm generate` and `check`; applying migrations through a driver you provide. See [Schema and migrations](./docs/migrations.md). | | `@maple-dev/effect-orm/database` | `Database` over your `SqlClient`: `run`, `execute`, `transaction` with retry. | +| `@maple-dev/effect-orm/clickhouse-http` | A ClickHouse `SqlClient` over `fetch`, for Cloudflare Workers and any runtime with `fetch`; give it to `Database`. | ## Extending with custom functions diff --git a/docs/database.md b/docs/database.md index a98b84f..2e0cccd 100644 --- a/docs/database.md +++ b/docs/database.md @@ -259,6 +259,31 @@ fails at BEGIN today: through the query path with a syntax error, and through `a `NOT_IMPLEMENTED`. The full measurements are in [`design/transactions.md`](../design/transactions.md). +## Tracing + +Each statement `Database` runs is one client span, named and attributed the way the +[OpenTelemetry database conventions](https://opentelemetry.io/docs/specs/semconv/database/database-spans/) +describe. A compiled query knows its operation and table, so the span needs no SQL parsing: + +| Attribute | Value | +| --------------------------- | -------------------------------------------------------------------- | +| `db.system.name` | `clickhouse`, or `postgresql` | +| `db.operation.name` | `SELECT`, `INSERT`, `UPDATE` or `DELETE` | +| `db.collection.name` | The `FROM` or written table; absent over a subquery, union or CTE | +| `db.query.summary` | `SELECT events`, or the name given with `withQuerySummary` | +| `db.response.returned_rows` | The rows it returned | +| `error.type` | On failure: the `DatabaseError` reason, such as `SqlSyntaxError` | + +The span is named after the summary, so a trace reads `SELECT events` rather than `sql.execute`. +Effect's own `sql.execute` span runs inside it and carries the query text. To give a group of +statements a stable name of your own, wrap them: + +```ts +Db.run(query, params).pipe(Db.withQuerySummary("dashboard.revenue_by_day")) +``` + +The table is also on the compiled query, as `compiled.collection`. + ## Writes The builder compiles SELECTs, [INSERTs](./inserts.md) and diff --git a/docs/reference.md b/docs/reference.md index ae09e57..89c2faf 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -513,6 +513,8 @@ placeholders, double-quoted identifiers. | `run`, `status`, `verify`, `baseline`, `MigrationDriver` and the migrate errors | `/migrate` | | `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `retryContention`, `Transaction` | `/database` | | `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` | +| `withQuerySummary`, `QuerySummary` — the `db.query.summary` of the statements an effect runs, for spans | `/database` | +| `ClickhouseHttpClient`, `layer`, `make`, `ClickhouseHttpClientConfig` — a ClickHouse `SqlClient` over `fetch`, for Workers | `/clickhouse-http` | | `defineConfig`, `generate`, `check`, `loadSchema`, `readMigrations`, `analyze`, `KitError` | `/kit` | | `defineSuite`, `query`, `caseFromCompiled`, `runSuite`, `compareRuns`, `compareBudgets` | `/benchmark` | | `Suite`, `RunOutput`, `BenchmarkError` and benchmark contracts | `/benchmark` | diff --git a/docs/running-queries.md b/docs/running-queries.md index 0b020de..717fde4 100644 --- a/docs/running-queries.md +++ b/docs/running-queries.md @@ -60,11 +60,62 @@ SQL failures as `SqlError`. not validate or escape arbitrary SQL; keep untrusted values in `CH.param` bindings. Leave result name transforms disabled so aliases still match the compiled decoder. -`@effect/sql-clickhouse` wraps the official Node.js client. For Workers or other fetch-based -runtimes, use a runtime-compatible adapter such as `@clickhouse/client-web` with an Effect -integration. Keep database credentials on your server. +`@effect/sql-clickhouse` wraps the official Node.js client, so it does not run in Cloudflare +Workers, Deno or a browser. For those, and anywhere `fetch` exists, use +[`@maple-dev/effect-orm/clickhouse-http`](#on-workers-and-other-fetch-runtimes) below. Keep +database credentials on your server. See the [Effect ClickHouse driver source](https://github.com/Effect-TS/effect/blob/main/packages/sql/clickhouse/src/ClickhouseClient.ts). +### On Workers and other fetch runtimes + +`@maple-dev/effect-orm/clickhouse-http` is a ClickHouse `SqlClient` over Effect's `HttpClient`. +Provide `FetchHttpClient.layer` and it runs wherever `fetch` does; give it to `Database` like +any other client: + +```ts +import { Effect, Layer, Redacted } from "effect" +import { FetchHttpClient } from "effect/http" +import * as SqlClient from "effect/sql/SqlClient" +import * as CH from "@maple-dev/effect-orm/clickhouse" +import * as ClickhouseHttp from "@maple-dev/effect-orm/clickhouse-http" +import * as Db from "@maple-dev/effect-orm/database" + +const ClickHouse = ClickhouseHttp.layer({ + url: "https://abc.clickhouse.cloud:8443", + username: "reader", + password: Redacted.make("…"), + database: "analytics", +}).pipe(Layer.provide(FetchHttpClient.layer)) + +const DatabaseLive = Layer.effect( + Db.Database, + Effect.gen(function* () { + return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: CH.clickhouseDialect }) + }), +).pipe(Layer.provideMerge(ClickHouse)) +``` + +What it does for you: + +- **Interrupting cancels the query.** A timeout or an interrupted fiber aborts the request and + sends `KILL QUERY` for its `query_id`: a closed connection alone does not stop a query + ClickHouse is still computing. That holds while the body is read too, and for a `.stream` + that ends before its body does (interrupted, failed, or stopped early with `Stream.take`). + Queries also send `cancel_http_readonly_queries_on_client_close`. +- **Failures carry Effect's `SqlError` reasons** (`SqlSyntaxError`, `AuthenticationError`, + `StatementTimeoutError`, …), read from ClickHouse's exception code, the same as + `@effect/sql-clickhouse`. +- **Credentials travel in a header,** never the URL; the URL may not contain them. A path in the + URL is kept, for a ClickHouse behind a proxy prefix. +- **Spans name the peer:** `server.address`, `server.port` and `peer.service` (`clickhouse`; + override it with `spanAttributes`), so a service map draws the edge to the database. +- `asCommand` runs DDL and writes, `withQueryId` sets the `query_id`, and `withSettings` adds + ClickHouse settings over the configured ones. Statements bind `sql` template values on the + server, as `{pN: Type}` placeholders. + +It opens no connection at startup, so a Worker's first request pays nothing extra. Its spans +are `Database`'s, described in [Running statements: tracing](./database.md#tracing). + For a client that implements Effect's `SqlClient`, the opt-in [`@maple-dev/effect-orm/database`](./database.md) entry point does this loop for you (`Db.run(compiled)`), and adds transactions on Postgres. diff --git a/package.json b/package.json index 59d405c..da976a3 100644 --- a/package.json +++ b/package.json @@ -38,6 +38,11 @@ "import": "./dist/clickhouse.mjs", "default": "./dist/clickhouse.mjs" }, + "./clickhouse-http": { + "types": "./dist/clickhouse-http.d.mts", + "import": "./dist/clickhouse-http.mjs", + "default": "./dist/clickhouse-http.mjs" + }, "./expr": { "types": "./dist/expr.d.mts", "import": "./dist/expr.mjs", @@ -132,4 +137,4 @@ "patchedDependencies": { "@effect/vitest@4.0.0": "patches/@effect%2Fvitest@4.0.0.patch" } -} \ No newline at end of file +} diff --git a/src/ch/compile.ts b/src/ch/compile.ts index 5709864..add77b4 100644 --- a/src/ch/compile.ts +++ b/src/ch/compile.ts @@ -270,6 +270,16 @@ interface CompiledQueryBase { * meant: the SQL may even run, with ClickHouse quoting and inlined params. */ readonly dialect?: string + /** + * The table the statement targets: a SELECT's `FROM` table, or the table an + * INSERT, UPDATE or DELETE writes. Absent when there is no single table, as + * for a `FROM` over a subquery, a union or a CTE, and for handwritten SQL. + * + * What a tracer reports as `db.collection.name`, and with `kind` the + * statement's `db.query.summary` (`SELECT events`), so a span names its + * table without parsing SQL. + */ + readonly collection?: string /** Runtime decode of raw query results. Queries built from handwritten SQL * should provide a row schema so schema drift is caught before consumers * read fields from `Record`. Without a schema this is an @@ -413,6 +423,7 @@ const makeCompiledQuery = ( kind: CompiledQueryBase["kind"] = "select", returning?: ReadonlyArray, crossTenantDeclared = false, + collection?: string, ): CompiledQuery => { let cachedDecodeRow: ((row: unknown) => Effect.Effect) | undefined let decoderBuilt = false @@ -489,6 +500,7 @@ const makeCompiledQuery = ( rowSchemaMismatch, ...(rawSql !== undefined ? { rawSql } : undefined), ...(dialect !== undefined ? { dialect } : undefined), + ...(collection !== undefined ? { collection } : undefined), ...(!(route === undefined) ? { route } : undefined), decodeRows, encodeRows, @@ -997,6 +1009,10 @@ function compileInner< "select", undefined, state.crossTenant === true, + // A table, not a CTE of the same name and not a subquery or union. + state.fromQuery || state.fromUnion || visibleCtes.some((c) => c.name === state.tableName) + ? undefined + : state.tableName, ), tenantScope === "single-tenant" ? scope.bound : undefined, ) @@ -1697,6 +1713,7 @@ const writeCompiledQuery = ( tenantBound: string | undefined, returning: ReturnType, crossTenant: boolean | undefined, + collection: string, ): CompiledQuery => { const returnedSchema = returning !== undefined && "schema" in returning.derived ? returning.derived.schema : undefined return withTenantBound( @@ -1714,6 +1731,7 @@ const writeCompiledQuery = ( kind, returning?.aliases, crossTenant === true, + collection, ), crossTenant === true ? undefined : tenantBound, ) @@ -1948,7 +1966,7 @@ function compileInsert(insert: CHInsert, params: Record +} + +type Reply = { readonly status?: number; readonly body: string; readonly headers?: Record } + +/** A ClickHouse HTTP client over a fake HttpClient that records each request. */ +const fake = (reply: (sent: Sent) => Reply, config: Partial = {}) => { + const sent: Array = [] + const http = HttpClient.make((request, url) => { + const body = request.body._tag === "Uint8Array" ? (request.body.text ?? new TextDecoder().decode(request.body.body)) : "" + const recorded = { url, body, headers: { ...request.headers } } + sent.push(recorded) + const { status = 200, body: text, headers } = reply(recorded) + return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(text, { status, headers }))) + }) + const live = layer({ + url: "https://clickhouse.example:8443", + username: "reader", + password: Redacted.make("s3cret"), + database: "analytics", + settings: { max_execution_time: 30 }, + ...config, + }).pipe(Layer.provide(Layer.succeed(HttpClient.HttpClient, http))) + return { sent, live } +} + +const json = (data: ReadonlyArray): Reply => ({ body: JSON.stringify({ meta: [], data, rows: data.length }) }) + +describe("ClickhouseHttpClient", () => { + it.effect("POSTs the SQL with its format, credentials as a header, and settings as URL parameters", () => { + const { sent, live } = fake(() => json([{ x: 1 }])) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const rows = yield* sql.unsafe("SELECT 1 AS x") + expect(rows).toEqual([{ x: 1 }]) + const [request] = sent + assert(request !== undefined) + expect(request.body).toBe("SELECT 1 AS x\nFORMAT JSON") + expect(request.headers.authorization).toBe(`Basic ${btoa("reader:s3cret")}`) + expect(request.url.origin).toBe("https://clickhouse.example:8443") + expect(request.url.username).toBe("") + expect(request.url.searchParams.get("database")).toBe("analytics") + expect(request.url.searchParams.get("max_execution_time")).toBe("30") + // An interrupted query stops on the server, not only in the client. + expect(request.url.searchParams.get("cancel_http_readonly_queries_on_client_close")).toBe("1") + expect(request.url.searchParams.get("query_id")).toMatch(/^[0-9a-f-]{36}$/) + }).pipe(Effect.provide(live)) + }) + + it.effect("binds parameters server-side, typed, as param_pN", () => { + const { sent, live } = fake(() => json([])) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + yield* sql`SELECT ${"o'brien"} AS name, ${42} AS n` + const [request] = sent + assert(request !== undefined) + expect(request.body).toBe("SELECT {p1: String} AS name, {p2: Float64} AS n\nFORMAT JSON") + expect(request.url.searchParams.get("param_p1")).toBe("o'brien") + expect(request.url.searchParams.get("param_p2")).toBe("42") + }).pipe(Effect.provide(live)) + }) + + it.effect("sends a command without a format, and returns no rows", () => { + const { sent, live } = fake(() => ({ body: "" })) + return Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + const rows = yield* client.asCommand(client.unsafe("CREATE TABLE t (x UInt8) ENGINE = Memory")) + expect(rows).toEqual([]) + const [request] = sent + assert(request !== undefined) + expect(request.body).toBe("CREATE TABLE t (x UInt8) ENGINE = Memory") + expect(request.url.searchParams.has("cancel_http_readonly_queries_on_client_close")).toBe(false) + }).pipe(Effect.provide(live)) + }) + + it.effect("applies a query id and per-call settings over the configured ones", () => { + const { sent, live } = fake(() => json([])) + return Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + yield* client.unsafe("SELECT 1").pipe(client.withSettings({ max_execution_time: 5 }), client.withQueryId("q-1")) + const [request] = sent + assert(request !== undefined) + expect(request.url.searchParams.get("query_id")).toBe("q-1") + expect(request.url.searchParams.get("max_execution_time")).toBe("5") + }).pipe(Effect.provide(live)) + }) + + it.effect.each([ + [62, "SqlSyntaxError"], + [516, "AuthenticationError"], + [497, "AuthorizationError"], + [159, "StatementTimeoutError"], + [241, "UnknownError"], + ] as const)("classifies ClickHouse code %s as %s", ([code, reason]) => { + const { live } = fake(() => ({ + status: 500, + body: `Code: ${code}. DB::Exception: fixture failure. (FIXTURE)`, + headers: { "x-clickhouse-exception-code": String(code) }, + })) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const exit = yield* Effect.exit(sql.unsafe("SELECT 1")) + assert(Exit.isFailure(exit)) + const error = exit.cause.reasons.find((r) => r._tag === "Fail") + assert(error?._tag === "Fail" && isSqlError(error.error)) + expect(error.error.reason._tag).toBe(reason) + expect(error.error.message).toContain("fixture failure") + }).pipe(Effect.provide(live)) + }) + + it.effect("reads the code from the body when the header is missing", () => { + const { live } = fake(() => ({ status: 400, body: "Code: 62. DB::Exception: Syntax error" })) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const error = yield* Effect.flip(sql.unsafe("SELEKT 1")) + expect(error.reason._tag).toBe("SqlSyntaxError") + }).pipe(Effect.provide(live)) + }) + + it.effect("streams JSONEachRow rows", () => { + const { sent, live } = fake(() => ({ body: '{"n":1}\n{"n":2}\n{"n":3}\n' })) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const rows = yield* Stream.runCollect(sql.unsafe("SELECT number AS n FROM numbers(3)").stream) + expect(rows).toEqual([{ n: 1 }, { n: 2 }, { n: 3 }]) + expect(sent[0]?.body).toBe("SELECT number AS n FROM numbers(3)\nFORMAT JSONEachRow") + }).pipe(Effect.provide(live)) + }) + + it.effect("runs effect-orm queries through Database, decoding 64-bit integers", () => { + const { sent, live } = fake(() => json([{ id: "18446744073709551615", n: "7" }])) + const Events = CH.table("events", { external: true, columns: { Id: CH.uint64, OrgId: CH.string } }) + const query = CH.from(Events) + .select(($) => ({ id: CH.toString($.Id), n: CH.count() })) + .where(($) => [$.OrgId.eq(CH.param.string("org"))]) + .groupBy("id") + return Effect.gen(function* () { + const rows = yield* Db.run(query, { org: "o'1" }) + expect(rows).toEqual([{ id: "18446744073709551615", n: 7 }]) + expect(sent[0]?.body).toContain("events.OrgId = 'o\\'1'") + }).pipe( + Effect.provide(Layer.effect(Db.Database, Effect.gen(function* () { + return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: CH.clickhouseDialect }) + })).pipe(Layer.provideMerge(live))), + ) + }) + + it.effect("kills the query by id when its fiber is interrupted", () => { + const sent: Array = [] + const http = HttpClient.make((request, url) => { + const body = request.body._tag === "Uint8Array" ? (request.body.text ?? "") : "" + sent.push({ url, body, headers: { ...request.headers } }) + // The query never answers; the kill does. + return body.startsWith("KILL") + ? Effect.succeed(HttpClientResponse.fromWeb(request, new Response(""))) + : Effect.never + }) + const live = layer({ url: "https://clickhouse.example" }).pipe(Layer.provide(Layer.succeed(HttpClient.HttpClient, http))) + return Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + const fiber = yield* Effect.forkChild(client.unsafe("SELECT sleep(3)").pipe(client.withQueryId("slow-1"))) + yield* Effect.yieldNow + yield* Fiber.interrupt(fiber) + const kill = sent.find((request) => request.body.startsWith("KILL")) + assert(kill !== undefined, "no KILL QUERY was sent") + expect(kill.body).toBe("KILL QUERY WHERE query_id = {id:String} ASYNC") + expect(kill.url.searchParams.get("param_id")).toBe("slow-1") + }).pipe(Effect.provide(live)) + }) + + /** A client whose queries answer 200 with `first` and then a body that never ends. */ + const hanging = (first: string) => { + const sent: Array = [] + const http = HttpClient.make((request, url) => { + const body = request.body._tag === "Uint8Array" ? (request.body.text ?? "") : "" + sent.push({ url, body, headers: { ...request.headers } }) + if (body.startsWith("KILL")) return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(""))) + const stream = new ReadableStream({ + start(controller) { + if (first !== "") controller.enqueue(new TextEncoder().encode(first)) + }, + }) + return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(stream))) + }) + const live = layer({ url: "https://clickhouse.example" }).pipe(Layer.provide(Layer.succeed(HttpClient.HttpClient, http))) + const kills = () => sent.filter((request) => request.body.startsWith("KILL")) + return { live, kills } + } + + it.effect("kills the query when a stream is stopped before its body ends", () => { + const { live, kills } = hanging('{"n":1}\n') + return Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + const rows = yield* Stream.runCollect( + client.unsafe("SELECT number AS n FROM numbers(1e9)").stream.pipe(Stream.take(1)), + ).pipe(client.withQueryId("early-1")) + expect(rows).toEqual([{ n: 1 }]) + expect(kills().map((kill) => kill.url.searchParams.get("param_id"))).toEqual(["early-1"]) + }).pipe(Effect.provide(live)) + }) + + it.effect("kills the query when interrupted while reading a response body", () => { + const { live, kills } = hanging("") + return Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + const fiber = yield* Effect.forkChild(client.unsafe("SELECT sleep(3)").pipe(client.withQueryId("body-1"))) + for (let i = 0; i < 5; i++) yield* Effect.yieldNow + yield* Fiber.interrupt(fiber) + expect(kills().map((kill) => kill.url.searchParams.get("param_id"))).toEqual(["body-1"]) + }).pipe(Effect.provide(live)) + }) + + it.effect("sends no kill for a stream read to the end", () => { + const { sent, live } = fake(() => ({ body: '{"n":1}\n{"n":2}\n' })) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + expect(yield* Stream.runCollect(sql.unsafe("SELECT 1").stream)).toEqual([{ n: 1 }, { n: 2 }]) + expect(sent.filter((request) => request.body.startsWith("KILL"))).toEqual([]) + }).pipe(Effect.provide(live)) + }) + + it.effect("keeps a path in the url, for ClickHouse behind a proxy prefix", () => { + const { sent, live } = fake(() => json([]), { url: "https://gateway.example/clickhouse" }) + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + yield* sql.unsafe("SELECT 1") + expect(sent[0]?.url.pathname).toBe("/clickhouse") + }).pipe(Effect.provide(live)) + }) + + it.effect("names the peer on its spans, which the caller can override", () => { + const spans: Array = [] + const recording = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan({ ...options, sampled: true }) + spans.push(span) + return span + }, + }) + const run = (config: Partial) => { + const { live } = fake(() => json([]), config) + spans.length = 0 + return Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + yield* sql.unsafe("SELECT 1") + const span = spans.find((s) => s.name === "sql.execute") + assert(span !== undefined) + return Object.fromEntries(span.attributes) + }).pipe(Effect.provide(live), Effect.withTracer(recording)) + } + return Effect.gen(function* () { + expect(yield* run({})).toMatchObject({ + "db.system.name": "clickhouse", + "db.namespace": "analytics", + "server.address": "clickhouse.example", + "server.port": 8443, + "peer.service": "clickhouse", + }) + expect((yield* run({ url: "https://ch.internal", spanAttributes: { "peer.service": "events-clickhouse" } }))).toMatchObject({ + "server.port": 443, + "peer.service": "events-clickhouse", + }) + }) + }) + + it("refuses credentials, a query string or a fragment in the url", () => { + for (const url of ["https://u:p@clickhouse.example", "https://clickhouse.example/?x=1", "https://clickhouse.example/#f"]) { + const { live } = fake(() => json([]), { url }) + expect(() => Effect.runSync(Effect.scoped(Layer.build(live)))).toThrow(/url/) + } + }) +}) diff --git a/src/clickhouse-http/client.ts b/src/clickhouse-http/client.ts new file mode 100644 index 0000000..d771e33 --- /dev/null +++ b/src/clickhouse-http/client.ts @@ -0,0 +1,315 @@ +// A ClickHouse `SqlClient` over Effect's `HttpClient`. +// +// Effect's own driver, `@effect/sql-clickhouse`, wraps the Node client, so it +// cannot run in Cloudflare Workers, Deno or a browser. This one speaks +// ClickHouse's HTTP interface through whatever `HttpClient` the caller provides +// (`FetchHttpClient.layer` anywhere `fetch` exists), and keeps that driver's +// shape: the same error reasons, `asCommand`, query ids and settings, so +// `Database` and the migration runner treat the two alike. +import { Context, Effect, Layer, Redacted, Stream } from "effect" +import { dual } from "effect/Function" +import * as Base64 from "effect/encoding/Base64" +import * as HttpClient from "effect/http/HttpClient" +import * as HttpClientRequest from "effect/http/HttpClientRequest" +import type * as HttpClientResponse from "effect/http/HttpClientResponse" +import * as Reactivity from "effect/reactivity/Reactivity" +import type * as Scope from "effect/Scope" +import * as Client from "effect/sql/SqlClient" +import type { Connection } from "effect/sql/SqlConnection" +import { + AuthenticationError, + AuthorizationError, + ConnectionError, + SqlError, + SqlSyntaxError, + StatementTimeoutError, + UnknownError, +} from "effect/sql/SqlError" +import * as Statement from "effect/sql/Statement" + +/** A ClickHouse setting value as the HTTP interface takes it. */ +export type ClickhouseSetting = string | number | boolean + +export interface ClickhouseHttpClientConfig { + /** + * The server, e.g. `https://abc.clickhouse.cloud:8443`. A path is kept, for a + * ClickHouse behind a proxy prefix (`https://gateway/clickhouse`); credentials, + * a query string or a fragment are refused. + */ + readonly url: string + readonly username?: string | undefined + readonly password?: Redacted.Redacted | string | undefined + /** The database unqualified names resolve in; `db.namespace` on spans. Defaults to `default`. */ + readonly database?: string | undefined + /** + * Settings sent with every statement, under the per-call ones + * (`withSettings`). Queries already send + * `cancel_http_readonly_queries_on_client_close=1`, so an interrupted fiber + * stops its query on the server as well as the request. + */ + readonly settings?: Readonly> | undefined + readonly spanAttributes?: Readonly> | undefined +} + +/** The ClickHouse HTTP client: a `SqlClient`, plus what statements over HTTP need. */ +export interface ClickhouseHttpClient extends Client.SqlClient { + readonly config: ClickhouseHttpClientConfig + /** Run the statements in `effect` as commands: DDL and writes, which send back no rows. */ + readonly asCommand: (effect: Effect.Effect) => Effect.Effect + /** The `query_id` of the statements in `effect`. Each statement gets a random one otherwise. */ + readonly withQueryId: { + (queryId: string): (effect: Effect.Effect) => Effect.Effect + (effect: Effect.Effect, queryId: string): Effect.Effect + } + /** ClickHouse settings for the statements in `effect`, over the configured ones. */ + readonly withSettings: { + (settings: Readonly>): (effect: Effect.Effect) => Effect.Effect + (effect: Effect.Effect, settings: Readonly>): Effect.Effect + } +} + +export const ClickhouseHttpClient = Context.Service("@maple-dev/effect-orm/ClickhouseHttpClient") + +const ClientMethod = Context.Reference<"query" | "command">("@maple-dev/effect-orm/ClickhouseHttpClient/ClientMethod", { + defaultValue: () => "query", +}) +const QueryId = Context.Reference("@maple-dev/effect-orm/ClickhouseHttpClient/QueryId", { + defaultValue: () => undefined, +}) +const Settings = Context.Reference>>( + "@maple-dev/effect-orm/ClickhouseHttpClient/Settings", + { defaultValue: () => ({}) }, +) + +// The codes Effect's Node driver classifies, so a failure means the same thing +// whichever client ran it. +const SYNTAX_ERROR_CODES = new Set([36, 60, 62, 242]) + +const classify = (code: number | undefined, message: string, cause: unknown, operation: string) => { + const props = { cause, message, operation } + if (code === 516) return new AuthenticationError(props) + if (code === 497) return new AuthorizationError(props) + if (code !== undefined && SYNTAX_ERROR_CODES.has(code)) return new SqlSyntaxError(props) + if (code === 159 || code === 469) return new StatementTimeoutError(props) + return new UnknownError(props) +} + +/** `Code: 62. DB::Exception: Syntax error …` when the header is missing. */ +const codeFromBody = (body: string): number | undefined => { + const match = /^Code:\s*(\d+)/.exec(body.trim()) + return match ? Number(match[1]) : undefined +} + +const validateUrl = (raw: string): URL => { + const url = new URL(raw) + if (url.username !== "" || url.password !== "") throw new Error("ClickhouseHttpClient: pass credentials as username and password, not in the url") + if (url.search !== "" || url.hash !== "") throw new Error("ClickhouseHttpClient: the url takes no query string or fragment") + return url +} + +const typeFromUnknown = (value: unknown): string => { + if (Array.isArray(value)) return `Array(${typeFromUnknown(value[0])})` + switch (typeof value) { + case "number": + return "Float64" + case "bigint": + return "Int64" + case "boolean": + return "Bool" + case "object": + return value instanceof Date ? "DateTime64(3)" : "String" + default: + return "String" + } +} + +/** `{pN: Type}` placeholders, sent as `param_pN`: ClickHouse's server-side binding. */ +const compiler = Statement.makeCompiler({ + dialect: "clickhouse", + placeholder: (i, value) => `{p${i}: ${typeFromUnknown(value)}}`, + onIdentifier: Statement.defaultEscape('"'), + onRecordUpdate: () => ["", []], + onCustom: () => ["", []], +}) + +const paramValue = (value: unknown): string => + value instanceof Date ? value.toISOString().replace("T", " ").replace("Z", "") : Array.isArray(value) ? JSON.stringify(value) : String(value) + +export const make = ( + config: ClickhouseHttpClientConfig, +): Effect.Effect => + Effect.gen(function* () { + const url = validateUrl(config.url) + const database = config.database ?? "default" + const password = config.password === undefined ? "" : Redacted.isRedacted(config.password) ? Redacted.value(config.password) : config.password + const authorization = + config.username === undefined ? undefined : `Basic ${Base64.encode(`${config.username}:${password}`)}` + const http = yield* HttpClient.HttpClient + + const post = (urlParams: Record, body: string) => { + const request = HttpClientRequest.post(url).pipe( + HttpClientRequest.setUrlParams(urlParams), + HttpClientRequest.bodyText(body, "text/plain; charset=utf-8"), + ) + return authorization === undefined ? request : HttpClientRequest.setHeader(request, "authorization", authorization) + } + // The statement span is the record of the request: an HTTP span would add + // nothing, and its URL carries bound parameter values. + const untraced = (effect: Effect.Effect) => + Effect.provideService(effect, HttpClient.TracerDisabledWhen, () => true) + // Best effort, and bounded: a kill that fails leaves the query to finish or + // to time out on its own, which is what would happen without it. + const kill = (queryId: string) => + untraced( + http.execute(post({ param_id: queryId }, "KILL QUERY WHERE query_id = {id:String} ASYNC")), + ).pipe(Effect.scoped, Effect.timeout("5 seconds"), Effect.ignore) + + const send = ( + sql: string, + params: ReadonlyArray, + format: "JSON" | "JSONCompact" | "JSONEachRow" | undefined, + operation: string, + ): Effect.Effect<{ readonly response: HttpClientResponse.HttpClientResponse; readonly queryId: string }, SqlError, Scope.Scope> => + Effect.withFiber((fiber) => { + const command = fiber.getRef(ClientMethod) === "command" + const queryId = fiber.getRef(QueryId) ?? crypto.randomUUID() + const urlParams: Record = { database, query_id: queryId } + if (!command) urlParams.cancel_http_readonly_queries_on_client_close = "1" + for (const [key, value] of Object.entries({ ...config.settings, ...fiber.getRef(Settings) })) { + urlParams[key] = String(value) + } + params.forEach((value, i) => { + urlParams[`param_p${i + 1}`] = paramValue(value) + }) + const request = post(urlParams, command || format === undefined ? sql : `${sql}\nFORMAT ${format}`) + return untraced(http.execute(request)).pipe( + // Aborting the request does not stop the query: ClickHouse notices a + // closed connection only when it next writes to it, which a query still + // computing its first block may not do for a long time. So kill it by id, + // as Effect's Node driver does. The body is read after this returns (the + // headers come with the first block), so readers keep the same guard. + Effect.onInterrupt(() => kill(queryId)), + Effect.mapError( + (cause) => + new SqlError({ + reason: new ConnectionError({ cause, message: "ClickhouseHttpClient: the request failed", operation }), + }), + ), + Effect.flatMap((response) => + response.status === 200 + ? Effect.succeed({ response, queryId }) + : Effect.flatMap(Effect.orElseSucceed(response.text, () => ""), (body) => { + const header = response.headers["x-clickhouse-exception-code"] + const code = header === undefined ? codeFromBody(body) : Number(header) + const message = body.trim().slice(0, 1000) || `ClickHouse answered ${response.status}` + return Effect.fail(new SqlError({ reason: classify(code, message, { status: response.status, code }, operation) })) + }), + ), + ) + }) + + const parseFailure = (cause: unknown, operation: string) => + new SqlError({ reason: new UnknownError({ cause, message: "ClickhouseHttpClient: the response is not valid JSON", operation }) }) + + const json = (sql: string, params: ReadonlyArray, format: "JSON" | "JSONCompact") => + Effect.scoped( + Effect.withFiber((fiber) => + fiber.getRef(ClientMethod) === "command" + ? Effect.as(send(sql, params, undefined, "execute"), [] as ReadonlyArray) + : send(sql, params, format, "execute").pipe( + Effect.flatMap(({ response, queryId }) => + response.text.pipe( + Effect.mapError((cause) => parseFailure(cause, "execute")), + Effect.onInterrupt(() => kill(queryId)), + ), + ), + Effect.flatMap((text) => + Effect.try({ + try: () => (JSON.parse(text) as { readonly data: ReadonlyArray }).data, + catch: (cause) => parseFailure(cause, "execute"), + }), + ), + ), + ), + ) + + const connection: Connection = { + execute: (sql, params, transformRows) => + transformRows ? Effect.map(json(sql, params, "JSON"), transformRows) : json(sql, params, "JSON"), + executeRaw: (sql, params) => json(sql, params, "JSON"), + executeValues: (sql, params) => json(sql, params, "JSONCompact"), + executeValuesUnprepared: (sql, params) => json(sql, params, "JSONCompact"), + executeUnprepared: (sql, params, transformRows) => + transformRows ? Effect.map(json(sql, params, "JSON"), transformRows) : json(sql, params, "JSON"), + executeStream: (sql, params, transformRows) => + send(sql, params, "JSONEachRow", "stream").pipe( + Effect.map(({ response, queryId }) => { + // A stream can end before the body does: interrupted, failed, or stopped + // early by its consumer (`Stream.take`). Any of those leaves the query + // running on the server, so it is killed unless the body was read to the end. + let drained = false + return response.stream.pipe( + Stream.onEnd(Effect.sync(() => void (drained = true))), + Stream.mapError( + (cause) => + new SqlError({ + reason: new ConnectionError({ cause, message: "ClickhouseHttpClient: the stream failed", operation: "stream" }), + }), + ), + Stream.decodeText(), + Stream.splitLines, + Stream.filter((line) => line.length > 0), + Stream.mapEffect((line) => + Effect.try({ try: () => JSON.parse(line) as object, catch: (cause) => parseFailure(cause, "stream") }), + ), + Stream.map((row) => (transformRows ? transformRows([row])[0] : row)), + Stream.ensuring(Effect.suspend(() => (drained ? Effect.void : kill(queryId)))), + ) + }), + Stream.unwrap, + ), + } + + const client = yield* Client.make({ + acquirer: Effect.succeed(connection), + compiler, + // The peer, for service maps: with the HTTP span off, nothing else names it. + // The caller's attributes come last, so they can rename the peer. + spanAttributes: [ + ["db.system.name", "clickhouse"], + ["db.namespace", database], + ["server.address", url.hostname], + ["server.port", Number(url.port || (url.protocol === "https:" ? 443 : 80))], + ["peer.service", "clickhouse"], + ...Object.entries(config.spanAttributes ?? {}), + ], + // ClickHouse has no transactions over HTTP without a session; effect-orm's + // ClickHouse dialect refuses them before this is ever sent. + beginTransaction: "BEGIN TRANSACTION", + }) + + return Object.assign(client, { + config, + asCommand: (effect: Effect.Effect) => Effect.provideService(effect, ClientMethod, "command"), + withQueryId: dual(2, (effect: Effect.Effect, queryId: string) => + Effect.provideService(effect, QueryId, queryId), + ), + withSettings: dual(2, (effect: Effect.Effect, settings: Readonly>) => + Effect.provideService(effect, Settings, settings), + ), + }) as ClickhouseHttpClient + }) + +/** + * `ClickhouseHttpClient` and `SqlClient` from `config`. Provide an + * `HttpClient`: `FetchHttpClient.layer` from `effect/http` runs anywhere + * `fetch` does, Cloudflare Workers included. + */ +export const layer = ( + config: ClickhouseHttpClientConfig, +): Layer.Layer => + Layer.effectContext( + Effect.map(make(config), (client) => + Context.make(ClickhouseHttpClient, client).pipe(Context.add(Client.SqlClient, client)), + ), + ).pipe(Layer.provide(Reactivity.layer)) diff --git a/src/database.ts b/src/database.ts index d01d8ec..ba66164 100644 --- a/src/database.ts +++ b/src/database.ts @@ -38,5 +38,6 @@ export { TransactionUnsupported, type TransactionError, } from "./database/errors" +export { QuerySummary, withQuerySummary } from "./database/tracing" export { sql, type SqlIdentifier, type SqlRawText, type SqlTemplate } from "./database/sql" export type { DialectTransactions, IsolationLevel, TransactionSettings } from "./ch/dialect" diff --git a/src/database/database.ts b/src/database/database.ts index 8e2dd76..aaf89d4 100644 --- a/src/database/database.ts +++ b/src/database/database.ts @@ -31,6 +31,7 @@ import { } from "./errors" import { isSqlTemplate, renderTemplate, type SqlTemplate } from "./sql" import { firstLine, reasonOf, sqlStateOf, toDatabaseError } from "./sql-error" +import { traceStatement, type TracedStatement } from "./tracing" /** SQL and the values bound to its placeholders. A `CompiledQuery` is one. */ export interface Statement { @@ -243,11 +244,16 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp const resolve = (statement: StatementInput): Effect.Effect => isSqlTemplate(statement) ? renderTemplate(statement, dialect) : checkScope(statement) - const rows = (statement: Statement) => - guard(statement).pipe( - Effect.andThen(raw.unsafe>(statement.sql, statement.parameters ?? [])), - Effect.map((rows): ReadonlyArray> => rows), - Effect.mapError(toDatabaseError(statement.sql)), + const rows = (statement: Statement & TracedStatement) => + traceStatement( + dialect, + statement, + guard(statement).pipe( + Effect.andThen(raw.unsafe>(statement.sql, statement.parameters ?? [])), + Effect.map((rows): ReadonlyArray> => rows), + Effect.mapError(toDatabaseError(statement.sql)), + ), + (rows) => rows.length, ) const decodeWith = @@ -271,11 +277,15 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp )) as DatabaseApi["query"] const execute: DatabaseApi["execute"] = (statement) => - Effect.flatMap(resolve(statement), (resolved) => - guard(resolved).pipe( - Effect.andThen(command(raw.unsafe(resolved.sql, resolved.parameters ?? []))), - Effect.asVoid, - Effect.mapError(toDatabaseError(resolved.sql)), + Effect.flatMap(resolve(statement), (resolved: Statement & TracedStatement) => + traceStatement( + dialect, + resolved, + guard(resolved).pipe( + Effect.andThen(command(raw.unsafe(resolved.sql, resolved.parameters ?? []))), + Effect.asVoid, + Effect.mapError(toDatabaseError(resolved.sql)), + ), ), ) diff --git a/src/database/tracing.test.ts b/src/database/tracing.test.ts new file mode 100644 index 0000000..9bc2fa3 --- /dev/null +++ b/src/database/tracing.test.ts @@ -0,0 +1,130 @@ +// The client span `Database` opens per statement, checked on PGlite with a +// tracer that records spans, against the OpenTelemetry database conventions. +import { PgliteClient } from "@effect/sql-pglite" +import { assert, describe, expect, it, layer } from "@effect/vitest" +import { Effect, Exit, Layer, Option, Tracer } from "effect" +import * as SqlClient from "effect/sql/SqlClient" +import * as PG from "../postgres" +import { postgresDialect } from "../pg/dialect" +import * as Db from "../database" + +const spans: Array = [] +const recording = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan({ ...options, sampled: true }) + spans.push(span) + return span + }, +}) + +const Live = Layer.effect( + Db.Database, + Effect.gen(function* () { + return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: postgresDialect }) + }), +).pipe(Layer.provideMerge(PgliteClient.layer({ postgresqlconf: "timezone = 'UTC'" }))) + +const Items = PG.table("traced_items", { columns: { id: PG.int4, name: PG.text }, primaryKey: ["id"] }) + +/** Runs `effect` traced, and returns the spans it opened, by name. */ +const traced = (effect: Effect.Effect) => + Effect.gen(function* () { + spans.length = 0 + const exit = yield* Effect.exit(Effect.withTracer(effect, recording)) + return { exit, spans: [...spans] } + }) + +const dbSpan = (recorded: ReadonlyArray) => { + const span = recorded.find((s) => s.attributes.has("db.system.name") && s.name !== "sql.execute") + assert(span !== undefined, `no statement span among ${recorded.map((s) => s.name).join(", ")}`) + return span +} + +layer(Live)("Database statement spans", (it) => { + it.effect("creates the table", () => + Db.execute(Db.sql`CREATE TABLE traced_items (id int4 PRIMARY KEY, name text NOT NULL)`), + ) + + it.effect("names a SELECT after its operation and table, as a client span", () => + Effect.gen(function* () { + const { exit, spans } = yield* traced( + Db.run(PG.from(Items).select(($) => ({ name: $.name })).where(($) => [$.id.eq(PG.param.int("id"))]), { id: 1 }), + ) + assert(Exit.isSuccess(exit)) + const span = dbSpan(spans) + expect(span.name).toBe("SELECT traced_items") + expect(span.kind).toBe("client") + expect(Object.fromEntries(span.attributes)).toMatchObject({ + "db.system.name": "postgresql", + "db.operation.name": "SELECT", + "db.collection.name": "traced_items", + "db.query.summary": "SELECT traced_items", + "db.response.returned_rows": 0, + }) + // Effect's own span runs inside it, carrying the query text. + const execute = spans.find((s) => s.name === "sql.execute") + assert(execute !== undefined) + expect(Option.getOrUndefined(execute.parent)).toBe(span) + }), + ) + + it.effect("names a write after its target table and counts RETURNING rows", () => + Effect.gen(function* () { + const { exit, spans } = yield* traced( + Db.run(PG.insertInto(Items).values({ id: 2, name: "two" }).returning(($) => ({ id: $.id }))), + ) + assert(Exit.isSuccess(exit)) + const span = dbSpan(spans) + expect(span.name).toBe("INSERT traced_items") + expect(span.attributes.get("db.collection.name")).toBe("traced_items") + expect(span.attributes.get("db.response.returned_rows")).toBe(1) + }), + ) + + it.effect("leaves the table out when there is no single one", () => + Effect.gen(function* () { + const inner = PG.from(Items).select(($) => ({ name: $.name })) + const { spans } = yield* traced(Db.run(PG.fromQuery(inner, "sub").select(($) => ({ name: $.name })))) + const span = dbSpan(spans) + expect(span.name).toBe("SELECT") + expect(span.attributes.has("db.collection.name")).toBe(false) + }), + ) + + it.effect("takes the summary from withQuerySummary", () => + Effect.gen(function* () { + const { spans } = yield* traced( + Db.run(PG.from(Items).select(($) => ({ id: $.id }))).pipe(Db.withQuerySummary("items.list")), + ) + const span = dbSpan(spans) + expect(span.name).toBe("items.list") + expect(span.attributes.get("db.query.summary")).toBe("items.list") + expect(span.attributes.get("db.operation.name")).toBe("SELECT") + }), + ) + + it.effect("records error.type from a DatabaseError's reason", () => + Effect.gen(function* () { + const { exit, spans } = yield* traced(Db.execute(Db.sql`SELEKT 1`)) + assert(Exit.isFailure(exit)) + const span = dbSpan(spans) + expect(span.attributes.get("error.type")).toBe("SqlSyntaxError") + expect(span.status._tag).toBe("Ended") + }), + ) +}) + +describe("collection", () => { + it("is the FROM table, or the written table, and absent otherwise", () => { + const select = PG.compileUnsafe(PG.from(Items).select(($) => ({ id: $.id }))) + expect(select.collection).toBe("traced_items") + const write = PG.compileUnsafe(PG.deleteFrom(Items).where(($) => [$.id.eq(1)])) + expect(write.collection).toBe("traced_items") + const cte = PG.compileUnsafe( + PG.from(PG.table("recent", { external: true, columns: { id: PG.int4 } })) + .withCTE("recent", PG.from(Items).select(($) => ({ id: $.id }))) + .select(($) => ({ id: $.id })), + ) + expect(cte.collection).toBeUndefined() + }) +}) diff --git a/src/database/tracing.ts b/src/database/tracing.ts new file mode 100644 index 0000000..97237ac --- /dev/null +++ b/src/database/tracing.ts @@ -0,0 +1,82 @@ +// One client span per statement, named and attributed the way the OpenTelemetry +// database semantic conventions describe: +// https://opentelemetry.io/docs/specs/semconv/database/database-spans/ +// +// Effect's own `sql.execute` span still runs inside it with the query text, but +// it cannot carry these: it names the operation after the client method +// (`execute`), and only a compiled query knows its table. +import { Context, Effect } from "effect" +import type { CompiledQuery } from "../ch/compile" +import type { Dialect } from "../ch/dialect" + +/** What the spans read off a statement. A `CompiledQuery` has all of it. */ +export type TracedStatement = { readonly sql: string } & Partial, "kind" | "collection">> + +/** + * A low-cardinality name for the statements run inside an effect, recorded as + * `db.query.summary` and used as the span name. Without one the summary is + * the operation and table (`SELECT events`), as the conventions recommend. + */ +export const QuerySummary = Context.Reference("@maple-dev/effect-orm/QuerySummary", { + defaultValue: () => undefined, +}) + +/** Names the statements `effect` runs, for spans: `db.query.summary`. */ +export const withQuerySummary: { + (summary: string): (effect: Effect.Effect) => Effect.Effect + (effect: Effect.Effect, summary: string): Effect.Effect +} = ((...args: ReadonlyArray) => + typeof args[0] === "string" + ? (effect: Effect.Effect) => Effect.provideService(effect, QuerySummary, args[0] as string) + : Effect.provideService(args[0] as Effect.Effect, QuerySummary, args[1] as string)) as never + +/** `db.system.name` values the conventions define, by dialect name. */ +const SYSTEM_NAMES: Record = { clickhouse: "clickhouse", postgres: "postgresql" } + +const OPERATIONS: Record, string> = { + select: "SELECT", + insert: "INSERT", + update: "UPDATE", + delete: "DELETE", +} + +/** + * Runs `effect`, which sends `statement`, in its client span. The span name is + * the summary when there is one, then `{operation} {table}`, then the system. + * `rowsOf` reads `db.response.returned_rows` from the result, when it has rows. + */ +export const traceStatement = ( + dialect: Dialect, + statement: TracedStatement, + effect: Effect.Effect, + rowsOf?: (result: A) => number, +): Effect.Effect => + Effect.withFiber((fiber) => { + const configured = fiber.getRef(QuerySummary) + const system = SYSTEM_NAMES[dialect.name] ?? dialect.name + const operation = statement.kind === undefined ? undefined : OPERATIONS[statement.kind] + const summary = + configured ?? (operation === undefined ? undefined : statement.collection === undefined ? operation : `${operation} ${statement.collection}`) + const attributes: Record = { "db.system.name": system } + if (operation !== undefined) attributes["db.operation.name"] = operation + if (statement.collection !== undefined) attributes["db.collection.name"] = statement.collection + if (summary !== undefined) attributes["db.query.summary"] = summary + return effect.pipe( + Effect.tap((result) => + rowsOf === undefined ? Effect.void : Effect.annotateCurrentSpan("db.response.returned_rows", rowsOf(result)), + ), + // `error.type` is the failure's class: a DatabaseError's reason, else its tag. + Effect.tapError((error) => { + const reason = + typeof error === "object" && error !== null + ? "reason" in error && typeof error.reason === "string" + ? error.reason + : "_tag" in error && typeof error._tag === "string" + ? error._tag + : undefined + : undefined + return reason === undefined ? Effect.void : Effect.annotateCurrentSpan("error.type", reason) + }), + Effect.withSpan(summary ?? system, { kind: "client", attributes }), + ) + }) diff --git a/tests/clickhouse-http.clickhouse.test.ts b/tests/clickhouse-http.clickhouse.test.ts new file mode 100644 index 0000000..909cc34 --- /dev/null +++ b/tests/clickhouse-http.clickhouse.test.ts @@ -0,0 +1,91 @@ +// The ClickHouse HTTP client against a real server, over `fetch`. Opt in with +// EFFECT_ORM_CLICKHOUSE_URL (and _USER, _PASSWORD); like the other live suites +// it creates no tables and writes no data. +import { assert, describe, expect, it } from "@effect/vitest" +import { Duration, Effect, Exit, Layer, Schedule, Stream } from "effect" +import { FetchHttpClient } from "effect/http" +import * as TestClock from "effect/testing/TestClock" +import * as SqlClient from "effect/sql/SqlClient" +import * as CH from "@maple-dev/effect-orm/clickhouse" +import * as Db from "@maple-dev/effect-orm/database" +import { ClickhouseHttpClient, layer } from "@maple-dev/effect-orm/clickhouse-http" +import { endpoint } from "./clickhouse-support" + +const Client = layer({ + url: endpoint ?? "http://127.0.0.1:8123", + username: process.env.EFFECT_ORM_CLICKHOUSE_USER ?? "default", + password: process.env.EFFECT_ORM_CLICKHOUSE_PASSWORD ?? "", + settings: { session_timezone: "UTC" }, +}).pipe(Layer.provide(FetchHttpClient.layer)) + +const Live = Layer.effect( + Db.Database, + Effect.gen(function* () { + return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: CH.clickhouseDialect }) + }), +).pipe(Layer.provideMerge(Client)) + +const Numbers = CH.table("system.numbers", { external: true, columns: { number: CH.uint64 } }) + +describe.skipIf(!endpoint)("ClickhouseHttpClient on ClickHouse", () => { + it.layer(Live)((it) => { + it.effect("runs a builder query and decodes it", () => + Effect.gen(function* () { + const rows = yield* Db.run( + CH.from(Numbers) + .select(($) => ({ n: $.number })) + .where(($) => [$.number.lt(CH.param.int("limit"))]) + .limit(3), + { limit: 3 }, + ) + expect(rows).toEqual([{ n: 0 }, { n: 1 }, { n: 2 }]) + }), + ) + + it.effect("binds parameters on the server", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const rows = yield* sql`SELECT ${"o'brien"} AS name, ${41} + 1 AS n` + expect(rows).toEqual([{ name: "o'brien", n: 42 }]) + }), + ) + + it.effect("streams rows", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const rows = yield* Stream.runCollect(sql.unsafe("SELECT toUInt8(number) AS n FROM numbers(4)").stream) + expect(rows).toEqual([{ n: 0 }, { n: 1 }, { n: 2 }, { n: 3 }]) + }), + ) + + it.effect("classifies a syntax error", () => + Effect.gen(function* () { + const exit = yield* Effect.exit(Db.query({ sql: "SELEKT 1" })) + assert(Exit.isFailure(exit)) + const error = exit.cause.reasons.find((r) => r._tag === "Fail") + assert(error?._tag === "Fail" && error.error._tag === "@maple-dev/effect-orm/DatabaseError") + expect(error.error.reason).toBe("SqlSyntaxError") + }), + ) + + // On the real clock: the timeout below has to fire, which a TestClock never does. + it.effect("stops the query on the server when the fiber is interrupted", () => + TestClock.withLive(Effect.gen(function* () { + const client = yield* ClickhouseHttpClient + const queryId = `effect-orm-cancel-${crypto.randomUUID()}` + const slow = client + .unsafe("SELECT count() FROM numbers(1e12) WHERE NOT ignore(sleepEachRow(0))") + .pipe(client.withQueryId(queryId), client.withSettings({ max_block_size: 1 })) + yield* Effect.timeout(slow, Duration.millis(500)).pipe(Effect.ignore) + // Aborting the request is not enough on its own; the client kills the query by id. + const running = yield* client + .unsafe<{ readonly n: number }>(`SELECT count() AS n FROM system.processes WHERE query_id = '${queryId}'`) + .pipe( + Effect.map((rows) => Number(rows[0]?.n)), + Effect.repeat({ until: (n) => n === 0, schedule: Schedule.spaced(Duration.millis(100)), times: 30 }), + ) + expect(running).toBe(0) + })), + ) + }) +}) diff --git a/tsdown.config.ts b/tsdown.config.ts index eddaa4f..7e1f75c 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -3,6 +3,7 @@ import { defineConfig } from "tsdown" export default defineConfig({ entry: { clickhouse: "./src/clickhouse.ts", + "clickhouse-http": "./src/clickhouse-http.ts", expr: "./src/expr.ts", sql: "./src/sql/index.ts", postgres: "./src/postgres.ts",