Class ConnectionImpl

packages/sql-clickhouse/src/ClickhouseClient.ts:118–231

    )

    class ConnectionImpl implements Connection {
      constructor(private readonly conn: Clickhouse.ClickHouseClient) {}

      private runRaw(sql: string, params: ReadonlyArray<unknown>, format: Clickhouse.DataFormat = "JSON") {
        const paramsObj: Record<string, unknown> = {}
        for (let i = 0; i < params.length; i++) {
          paramsObj[`p${i + 1}`] = params[i]
        }
        return Effect.withFiberRuntime<Clickhouse.ResultSet<"JSON"> | Clickhouse.CommandResult, SqlError>((fiber) => {
          const method = fiber.getFiberRef(currentClientMethod)
          return Effect.async<Clickhouse.ResultSet<"JSON"> | Clickhouse.CommandResult, SqlError>((resume) => {
            const queryId = fiber.getFiberRef(currentQueryId) ?? Crypto.randomUUID()
            const settings = fiber.getFiberRef(currentClickhouseSettings) ?? {}
            const controller = new AbortController()
            if (method === "command") {
              this.conn.command({
                query: sql,
                query_params: paramsObj,
                abort_signal: controller.signal,
                query_id: queryId,
                clickhouse_settings: settings
              }).then(
                (result) => resume(Effect.succeed(result)),
                (cause) => resume(Effect.fail(new SqlError({ cause, message: "Failed to execute statement" })))
              )
            } else {
              this.conn.query({
                query: sql,
                query_params: paramsObj,
                abort_signal: controller.signal,
                query_id: queryId,
                clickhouse_settings: settings,
                format
              }).then(
                (result) => resume(Effect.succeed(result)),
                (cause) => resume(Effect.fail(new SqlError({ cause, message: "Failed to execute statement" })))
              )
            }
            return Effect.suspend(() => {
              controller.abort()
              return Effect.promise(() => this.conn.command({ query: `KILL QUERY WHERE query_id = '${queryId}'` }))
            })
          })
        })
      }

      private run(sql: string, params: ReadonlyArray<unknown>, format?: Clickhouse.DataFormat) {
        return this.runRaw(sql, params, format).pipe(
          Effect.flatMap((result) => {
            if ("json" in result) {
              return Effect.promise(() =>
                result.json().then(
                  (result) => "data" in result ? result.data : result as any,
                  () => []
                )
              )
            }
            return Effect.succeed([])