Function makeConection

repos/effect/packages/sql/pg/src/PgClient.ts:290–355

(client?: Pg.PoolClient)

    readonly transformQueryNames?: ((str: string) => string) | undefined
    readonly transformJson?: boolean | undefined
    readonly types?: Pg.CustomTypesConfig | undefined
  }
): Effect.fn.Return<PgClient, SqlError, Scope.Scope | Reactivity.Reactivity> {
  const pool = yield* options.acquire

  const makeConection = (client?: Pg.PoolClient) =>
    new ConnectionImpl(
      function runWithClient<A>(f: (client: Pg.ClientBase, resume: (_: Effect.Effect<A, SqlError>) => void) => void) {
        if (client !== undefined) {
          return Effect.callback<A, SqlError>((resume) => {
            f(client!, resume)
            return makeCancel(pool, client!)
          })
        }
        return Effect.callback<A, SqlError>((resume) => {
          let done = false
          let cancel: Effect.Effect<void> | undefined = undefined
          let client: Pg.PoolClient | undefined = undefined
          function onError(cause: Error) {
            cleanup(cause)
            resume(Effect.fail(new SqlError({ reason: classifyError(cause, "Connection error", "acquireConnection") })))
          }
          function cleanup(cause?: Error) {
            if (!done) client?.release(cause)
            done = true
            client?.off("error", onError)
          }
          pool.connect((cause, client_) => {
            if (cause) {
              return resume(
                Effect.fail(
                  new SqlError({
                    reason: classifyError(cause, "Failed to acquire connection", "acquireConnection")
                  })
                )
              )
            } else if (!client_) {
              return resume(
                Effect.fail(
                  new SqlError({
                    reason: new ConnectionError({
                      message: "Failed to acquire connection",
                      cause: new Error("No client returned"),
                      operation: "acquireConnection"
                    })
                  })
                )
              )
            } else if (done) {
              client_.release()
              return
            }
            client = client_
            client.once("error", onError)
            cancel = makeCancel(pool, client)
            f(client, (eff) => {
              cleanup()
              resume(eff)