// Pooled transaction APIs.

///|
/// Convert one pooled isolation level into the SQL clause expected by `BEGIN`.
fn isolation_level_sql(level : IsolationLevel) -> String {
  match level {
    ReadUncommitted => "READ UNCOMMITTED"
    ReadCommitted => "READ COMMITTED"
    RepeatableRead => "REPEATABLE READ"
    Serializable => "SERIALIZABLE"
  }
}

///|
/// Start one transaction using PostgreSQL's default `BEGIN` settings or the
/// given `BEGIN` options.
///
/// Preconditions: no exclusive scope may already be active on this lease. The
/// returned `Transaction` keeps the connection checked out until it is
/// committed or rolled back.
pub async fn Client::transaction(
  self : Client,
  options? : TransactionOptions = TransactionOptions::new(),
) -> Transaction {
  let (client, connection) = self.state.begin_transaction_scope()
  errdefer self.state.finish_transaction_scope()
  let client_options = @client.TransactionOptions::new(
    isolation_level?=options.isolation_level.map(isolation_level_sql),
    read_only?=options.read_only,
    deferrable?=options.deferrable,
  )
  let transaction = client.transaction(options=client_options)
  make_pooled_transaction(connection, transaction, ClientLease(self.state))
}

///|
/// Run one callback inside a transaction and auto-complete it.
///
/// On normal return, the transaction is committed if the callback did not
/// already finish it explicitly. If the callback raises or is cancelled, the
/// transaction is rolled back best-effort unless it was already finished.
pub async fn[T] Client::with_transaction(
  self : Client,
  f : async (Transaction) -> T,
  options? : TransactionOptions = TransactionOptions::new(),
) -> T {
  let transaction = self.transaction(options~)
  errdefer (if !transaction.state.finished.val {
    transaction.rollback_best_effort()
  })
  let result = f(transaction)
  if !transaction.state.finished.val {
    transaction.commit_internal()
  }
  result
}

///|
/// Release the outer scope holding a pooled transaction's connection lease.
fn TransactionOwner::finish_scope(self : TransactionOwner) -> Unit {
  match self {
    ClientLease(state) => state.finish_transaction_scope()
    ParentTransaction(state) => state.finish_nested_scope()
  }
}

///|
/// Start one transaction operation or reject work after completion begins.
fn TransactionState::begin_operation(
  self : TransactionState,
) -> @client.Transaction raise {
  if self.closing.val || self.finished.val {
    raise PoolError::LeaseReleased
  }
  if self.scope_active.val {
    raise PoolError::OperationInProgress
  }
  match self.transaction.val {
    Some(transaction) => {
      self.active_ops.val += 1
      transaction
    }
    None => raise PoolError::LeaseReleased
  }
}

///|
/// Finish one transaction operation.
fn TransactionState::finish_operation(self : TransactionState) -> Unit {
  if self.active_ops.val > 0 {
    self.active_ops.val -= 1
  }
}

///|
/// Start one nested transaction or savepoint scope on this transaction.
fn TransactionState::begin_nested_scope(
  self : TransactionState,
) -> (@client.Transaction, ConnectionState) raise {
  if self.closing.val || self.finished.val {
    raise PoolError::LeaseReleased
  }
  if self.scope_active.val || self.active_ops.val > 0 {
    raise PoolError::OperationInProgress
  }
  match self.transaction.val {
    Some(transaction) => {
      self.scope_active.val = true
      self.active_ops.val += 1
      (transaction, self.connection)
    }
    None => raise PoolError::LeaseReleased
  }
}

///|
/// Finish one nested transaction or savepoint scope on this transaction.
fn TransactionState::finish_nested_scope(self : TransactionState) -> Unit {
  self.scope_active.val = false
  self.finish_operation()
}

///|
/// Release the outer client or parent transaction once this transaction completes.
fn TransactionState::release_owner_scope(self : TransactionState) -> Unit {
  if self.owner_released.val {
    return
  }
  self.owner_released.val = true
  self.owner.finish_scope()
}

///|
/// Wait for every active operation to finish and take ownership of the transaction.
async fn Transaction::take_for_completion(
  self : Transaction,
) -> @client.Transaction {
  if self.state.finished.val {
    raise PoolError::LeaseReleased
  }
  if self.state.scope_active.val {
    raise PoolError::OperationInProgress
  }
  self.state.closing.val = true
  while self.state.active_ops.val > 0 {
    @async.pause()
  }
  self.state.finished.val = true
  match self.state.transaction.val {
    Some(transaction) => {
      self.state.transaction.val = None
      transaction
    }
    None => raise PoolError::LeaseReleased
  }
}

///|
/// Commit one pooled transaction after all in-flight operations finish.
async fn Transaction::commit_internal(self : Transaction) -> Unit {
  let transaction = self.take_for_completion()
  defer self.state.release_owner_scope()
  @async.protect_from_cancel(() => transaction.commit())
}

///|
/// Roll back one pooled transaction while ignoring rollback failures.
async fn Transaction::rollback_best_effort(self : Transaction) -> Unit {
  try
    @async.protect_from_cancel(() => {
      let transaction = self.take_for_completion()
      defer self.state.release_owner_scope()
      transaction.rollback()
    })
  catch {
    _ => ()
  } noraise {
    _ => ()
  }
}

///|
/// Run one async operation against the live transaction.
async fn[T] Transaction::run_operation(
  self : Transaction,
  op : async (@client.Transaction) -> T,
) -> T {
  let transaction = self.state.begin_operation()
  defer self.state.finish_operation()
  @async.protect_from_cancel(() => op(transaction))
}

///|
/// Commit this pooled transaction explicitly.
///
/// Preconditions: the transaction must not already be finished, and no nested
/// transaction/savepoint scope may still be active. The call waits for any
/// in-flight non-exclusive operation to finish first.
pub async fn Transaction::commit(self : Transaction) -> Unit {
  self.commit_internal()
}

///|
/// Roll back this pooled transaction explicitly.
///
/// Preconditions and completion behavior match `commit()`, except the final SQL
/// action is `ROLLBACK`.
pub async fn Transaction::rollback(self : Transaction) -> Unit {
  let transaction = self.take_for_completion()
  defer self.state.release_owner_scope()
  @async.protect_from_cancel(() => transaction.rollback())
}

///|
/// Start one nested transaction backed by an unnamed savepoint.
///
/// Preconditions: the parent transaction must still be active and must not
/// already be inside another nested transaction scope. The returned nested
/// transaction must also be committed or rolled back.
pub async fn Transaction::transaction(self : Transaction) -> Transaction {
  let (transaction, connection) = self.state.begin_nested_scope()
  errdefer self.state.finish_nested_scope()
  let nested = transaction.transaction()
  make_pooled_transaction(connection, nested, ParentTransaction(self.state))
}

///|
/// Start one nested transaction backed by a named savepoint.
///
/// `name` is passed directly to the underlying client savepoint API.
pub async fn Transaction::savepoint(
  self : Transaction,
  name : String,
) -> Transaction {
  let (transaction, connection) = self.state.begin_nested_scope()
  errdefer self.state.finish_nested_scope()
  let nested = transaction.savepoint(name)
  make_pooled_transaction(connection, nested, ParentTransaction(self.state))
}

///|
/// Run one callback inside an unnamed savepoint-backed nested transaction.
///
/// On success the nested transaction is committed if still unfinished; on error
/// or cancellation it is rolled back best-effort.
pub async fn[T] Transaction::with_transaction(
  self : Transaction,
  f : async (Transaction) -> T,
) -> T {
  let nested = self.transaction()
  errdefer (if !nested.state.finished.val { nested.rollback_best_effort() })
  let result = f(nested)
  if !nested.state.finished.val {
    nested.commit_internal()
  }
  result
}

///|
/// Run one callback inside a named savepoint-backed nested transaction.
///
/// On success the nested transaction is committed if still unfinished; on error
/// or cancellation it is rolled back best-effort.
pub async fn[T] Transaction::with_savepoint(
  self : Transaction,
  name : String,
  f : async (Transaction) -> T,
) -> T {
  let nested = self.savepoint(name)
  errdefer (if !nested.state.finished.val { nested.rollback_best_effort() })
  let result = f(nested)
  if !nested.state.finished.val {
    nested.commit_internal()
  }
  result
}

///|
/// Run one query inside this transaction and collect all rows.
pub async fn Transaction::query_all(
  self : Transaction,
  sql : String,
  params? : Array[&@client.ToSql] = [],
) -> Array[@client.Row] {
  self.run_operation(txn => txn.query(sql, params~).collect())
}

///|
/// Run one query inside this transaction and require exactly one row.
///
/// Edge behavior: this helper materializes all rows first and raises
/// `PoolError::RowCount` when PostgreSQL returned anything other than exactly
/// one row.
pub async fn Transaction::query_one(
  self : Transaction,
  sql : String,
  params? : Array[&@client.ToSql] = [],
) -> @client.Row {
  let rows = self.query_all(sql, params~)
  guard rows.length() == 1 else {
    raise PoolError::RowCount(
      "expected exactly one row, got \{rows.length().to_string()}",
    )
  }
  rows[0]
}

///|
/// Run one query inside this transaction and allow zero or one rows.
///
/// Edge behavior: raises `PoolError::RowCount` when more than one row is
/// returned.
pub async fn Transaction::query_opt(
  self : Transaction,
  sql : String,
  params? : Array[&@client.ToSql] = [],
) -> @client.Row? {
  let rows = self.query_all(sql, params~)
  guard rows.length() <= 1 else {
    raise PoolError::RowCount(
      "expected zero or one row, got \{rows.length().to_string()}",
    )
  }
  if rows.length() == 0 {
    None
  } else {
    Some(rows[0])
  }
}

///|
/// Run one typed query inside the transaction and collect all rows.
///
/// Side effects: prepares a temporary typed statement for this call and closes
/// it before returning.
pub async fn Transaction::query_typed_all(
  self : Transaction,
  sql : String,
  param_types : Array[@client.Type],
  params? : Array[&@client.ToSql] = [],
) -> Array[@client.Row] {
  self.with_prepared_typed(sql, param_types, prepared => {
    prepared.query_all(params~)
  })
}

///|
/// Execute one command inside this transaction and return its affected row count.
pub async fn Transaction::execute(
  self : Transaction,
  sql : String,
  params? : Array[&@client.ToSql] = [],
) -> Int {
  self.run_operation(txn => txn.execute(sql, params~))
}

///|
/// Execute one or more SQL commands inside this transaction.
pub async fn Transaction::batch_execute(
  self : Transaction,
  sql : String,
) -> Unit {
  self.run_operation(txn => txn.batch_execute(sql))
}

///|
/// Perform a lightweight health check while the transaction is open.
///
/// Side effects: executes `select 1::int4 as value` inside the current
/// transaction instead of using a transaction-independent probe.
pub async fn Transaction::check_connection(self : Transaction) -> Unit {
  self.run_operation(txn => {
    ignore(txn.query("select 1::int4 as value").finish())
  })
}