| 358 | return Effect.provideService(effect, ClientMethod, "command") |
| 359 | }, |
| 360 | insertQuery<T = unknown>(options: { |
| 361 | readonly table: string |
| 362 | readonly values: Clickhouse.InsertValues<Readable, T> |
| 363 | readonly format?: Clickhouse.DataFormat |
| 364 | }) { |
| 365 | return Effect.callback<Clickhouse.InsertResult, SqlError>((resume) => { |
| 366 | const fiber = Fiber.getCurrent()! |
| 367 | const queryId = fiber.getRef(QueryId) ?? Crypto.randomUUID() |
| 368 | const settings = fiber.getRef(ClickhouseSettings) |
| 369 | const controller = new AbortController() |
| 370 | client.insert({ |
| 371 | format: "JSONEachRow", |
| 372 | ...options, |
| 373 | abort_signal: controller.signal, |
| 374 | query_id: queryId, |
| 375 | clickhouse_settings: settings |
| 376 | }).then( |
| 377 | (result) => resume(Effect.succeed(result)), |
| 378 | (cause) => |
| 379 | resume(Effect.fail(new SqlError({ reason: classifyError(cause, "Failed to insert data", "insert") }))) |
| 380 | ) |
| 381 | return Effect.suspend(() => { |
| 382 | controller.abort() |
| 383 | return Effect.promise(() => |
| 384 | client.command({ |
| 385 | query: "KILL QUERY WHERE query_id = {queryId:String}", |
| 386 | query_params: { queryId } |
| 387 | }) |
| 388 | ) |
| 389 | }) |
| 390 | }) |
| 391 | }, |
| 392 | withQueryId: dual( |
| 393 | 2, |
| 394 | <A, E, R>(effect: Effect.Effect<A, E, R>, queryId: string) => Effect.provideService(effect, QueryId, queryId) |