Function makeWithTransaction

packages/effect/src/unstable/sql/SqlClient.ts:224–302

(options: {
  readonly transactionService: Context.Key<I, readonly [conn: S, counter: number]>
  readonly spanAttributes: ReadonlyArray<readonly [string, unknown]>
  readonly acquireConnection: Effect.Effect<readonly [Scope.Closeable | undefined, S], SqlError>
  readonly begin: (conn: NoInfer<S>) => Effect.Effect<void, SqlError>
  readonly savepoint: (conn: NoInfer<S>, id: number) => Effect.Effect<void, SqlError>
  readonly commit: (conn: NoInfer<S>) => Effect.Effect<void, SqlError>
  readonly rollback: (conn: NoInfer<S>) => Effect.Effect<void, SqlError>
  readonly rollbackSavepoint: (conn: NoInfer<S>, id: number) => Effect.Effect<void, SqlError>
})

 * @since 4.0.0
 */
export const makeWithTransaction = <I, S>(options: {
  readonly transactionService: Context.Key<I, readonly [conn: S, counter: number]>
  readonly spanAttributes: ReadonlyArray<readonly [string, unknown]>
  readonly acquireConnection: Effect.Effect<readonly [Scope.Closeable | undefined, S], SqlError>
  readonly begin: (conn: NoInfer<S>) => Effect.Effect<void, SqlError>
  readonly savepoint: (conn: NoInfer<S>, id: number) => Effect.Effect<void, SqlError>
  readonly commit: (conn: NoInfer<S>) => Effect.Effect<void, SqlError>
  readonly rollback: (conn: NoInfer<S>) => Effect.Effect<void, SqlError>
  readonly rollbackSavepoint: (conn: NoInfer<S>, id: number) => Effect.Effect<void, SqlError>
}) => {
  const transactionSemaphore = Context.Service<Semaphore.Semaphore>(
    `effect/sql/SqlClient/TransactionSemaphore/${transactionSemaphoreIdCounter++}`
  )
  return <R, E, A>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E | SqlError, R> =>
    Effect.uninterruptibleMask((restore) =>
      Effect.useSpan(
        "sql.transaction",
        { kind: "client" },
        (span) =>
          Effect.withFiber<A, E | SqlError, R>((fiber) => {
            for (const [key, value] of options.spanAttributes) {
              span.attribute(key, value)
            }
            const services = fiber.context
            const clock = fiber.getRef(Clock)
            const connOption = Context.getOption(services, options.transactionService)
            const conn = connOption._tag === "Some"
              ? Effect.succeed([undefined, connOption.value[0]] as const)
              : options.acquireConnection
            const id = connOption._tag === "Some" ? connOption.value[1] + 1 : 0
            const transaction = Effect.flatMap(
              conn,
              (
                [scope, conn]
              ) =>
                (id === 0 ? options.begin(conn) : options.savepoint(conn, id)).pipe(
                  Effect.flatMap(() =>
                    Effect.onExitPrimitive(
                      Effect.provideContext(
                        restore(effect),
                        services.pipe(
                          Context.add(options.transactionService, [conn, id]),
                          Context.add(transactionSemaphore, Semaphore.makeUnsafe(1)),
                          Context.add(Tracer.ParentSpan, span)
                        )
                      ),
                      (exit) => {
                        let effect: Effect.Effect<void>
                        if (Exit.isSuccess(exit)) {
                          if (id === 0) {
                            span.event("db.transaction.commit", clock.currentTimeNanosUnsafe())
                            effect = Effect.orDie(options.commit(conn))
                          } else {
                            span.event("db.transaction.savepoint", clock.currentTimeNanosUnsafe())
                            effect = Effect.void
                          }
                        } else {
                          span.event("db.transaction.rollback", clock.currentTimeNanosUnsafe())