トランザクション

/// Published once the money has moved.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Transferred {
    pub from: i64,
    pub to: i64,
    pub amount: i64,
}

/// Why the transfer was rolled back.
#[derive(Debug)]
pub enum TransferFailed {
    InsufficientFunds,
    UnknownAccount,
    /// Serialization failure or deadlock: safe to run the transfer again.
    Conflict,
    Database(sqlx::Error),
}

impl TransferFailed {
    fn from_sqlx(error: sqlx::Error) -> Self {
        if is_conflict(&error) {
            TransferFailed::Conflict
        } else {
            TransferFailed::Database(error)
        }
    }
}

tx_effect! {
    /// Everything a transfer touches.
    pub enum TransferEffect { write Account }
}

/// The state of a transfer.
#[derive(Debug)]
pub enum Transfer {
    Start { from: i64, to: i64, amount: i64 },
    End(TxEnd<Transfer>),
}

impl Transfer {
    pub fn new(from: i64, to: i64, amount: i64) -> Self {
        Transfer::Start { from, to, amount }
    }

    fn commit(from: i64, to: i64, amount: i64) -> Self {
        Transfer::End(TxEnd::Commit {
            output: (),
            report: Some(Transferred { from, to, amount }),
        })
    }

    fn rollback(why: TransferFailed) -> Self {
        Transfer::End(TxEnd::Rollback { output: why, report: None })
    }

    /// Whether the debit took the money.
    fn check_debit(debited: Result<u64, sqlx::Error>) -> Result<(), TransferFailed> {
        match debited {
            Ok(1) => Ok(()),
            Ok(_) => Err(TransferFailed::InsufficientFunds),
            Err(error) => Err(TransferFailed::from_sqlx(error)),
        }
    }

    /// Whether the credit reached an account.
    fn check_credit(credited: Result<u64, sqlx::Error>) -> Result<(), TransferFailed> {
        match credited {
            Ok(1) => Ok(()),
            Ok(_) => Err(TransferFailed::UnknownAccount),
            Err(error) => Err(TransferFailed::from_sqlx(error)),
        }
    }
}

impl From<TxEnd<Transfer>> for Transfer {
    fn from(end: TxEnd<Transfer>) -> Self {
        Transfer::End(end)
    }
}

impl TxFlow for Transfer {
    type Capability = TransferEffect;
    type Database = Postgres;
    type Commit = ();
    type Abort = TransferFailed;
    type Report = Option<Transferred>;

    const ISOLATION: Option<IsolationLevel> = Some(IsolationLevel::Serializable);

    async fn step(tx: Tx<'_, Self>, state: Self) -> TxState<'_, Self> {
        let Transfer::Start { from, to, amount } = state else {
            // Never stepped: the driver ends a terminal state first.
            return tx.end(state).await;
        };
        let (tx, debited) = tx.run(Debit { from, amount }).await;
        if let Err(why) = Self::check_debit(debited) {
            return tx.end(Self::rollback(why)).await;
        }
        let (tx, credited) = tx.run(Credit { to, amount }).await;
        match Self::check_credit(credited) {
            Ok(()) => tx.end(Self::commit(from, to, amount)).await,
            Err(why) => tx.end(Self::rollback(why)).await,
        }
    }

    fn finished(self) -> Either<Self, TxEnd<Self>> {
        match self {
            Transfer::End(end) => Either::Right(end),
            state => Either::Left(state),
        }
    }
}

use crate::transfer::{Transfer, TransferFailed};
use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::deps::sqlx;
use wakuwaku::entity::sqlx::backend::PgSource;
use wakuwaku::entity::sqlx::transaction::{Transaction, TxError, TxOutput};
use wakuwaku::entity::{Db, ReadWrite};

/// Why a payment did not go through.
#[derive(Debug)]
pub enum PayError {
    /// The transfer rolled back, and this is why.
    Refused(TransferFailed),
    /// `COMMIT` was sent but its answer was lost: the money may have moved.
    OutcomeUnknown(sqlx::Error),
    /// The transaction did not end as the flow planned.
    Transaction(TxError),
}

pub async fn pay(
    db: &Db<PgSource, ReadWrite>,
    from: i64,
    to: i64,
    amount: i64,
) -> Result<(), PayError> {
    match db.process(Transaction(Transfer::new(from, to, amount))).await {
        Ok(TxOutput::Committed(())) => Ok(()),
        Ok(TxOutput::RolledBack(why)) => Err(PayError::Refused(why)),
        Err(TxError::CommitUnknown(error)) => Err(PayError::OutcomeUnknown(error)),
        Err(error) => Err(PayError::Transaction(error)),
    }
}

tx_effect! {
    /// Moves money between accounts, after checking who owns them.
    pub enum CheckedTransferEffect { write Account, read User }
}

async fn on_replica(replica: &Db<PgSource, ReadOnly>) {
    // error[E0277]: the trait bound
    // `Db<SqlxSource<Postgres>, ReadOnly>: Processor<Transaction<Transfer>>` is not satisfied
    let _ = replica.process(Transaction(Transfer::new(1, 2, 10))).await;
}

async fn step(tx: Tx<'_, Self>, state: Self) -> TxState<'_, Self> {
    // Does not compile: `Debit` writes `Account`, and this flow declares `read Account`.
    let (tx, _) = tx.run(Debit { from: 1, amount: 10 }).await;
    // …
}

use wakuwaku::deps::sqlx;
use wakuwaku::entity::sqlx::transaction::AsSqlxError;

#[derive(Debug)]
pub enum SaveError {
    /// Someone else saved first: the version did not match.
    Stale,
    Database(sqlx::Error),
}

impl From<sqlx::Error> for SaveError {
    fn from(error: sqlx::Error) -> Self {
        SaveError::Database(error)
    }
}

impl AsSqlxError for SaveError {
    fn as_sqlx_error(&self) -> Option<&sqlx::Error> {
        match self {
            SaveError::Database(error) => Some(error),
            SaveError::Stale => None,
        }
    }
}

use crate::transfer::{Transfer, TransferFailed, is_conflict};
use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::entity::sqlx::transaction::{Transaction, TxError, TxOutput};

const MAX_ATTEMPTS: usize = 3;

/// Run the transfer, running it again from the start after a conflict.
pub async fn transfer_with_retry<D>(
    db: &D,
    from: i64,
    to: i64,
    amount: i64,
) -> Result<TxOutput<(), TransferFailed>, TxError>
where
    D: Processor<Transaction<Transfer>, Output = TxOutput<(), TransferFailed>, Error = TxError>,
{
    for _ in 1..MAX_ATTEMPTS {
        match db.process(Transaction(Transfer::new(from, to, amount))).await {
            // Aborted by a serialization failure or a deadlock.
            Ok(TxOutput::RolledBack(TransferFailed::Conflict)) => continue,
            // SERIALIZABLE can also reject the COMMIT itself.
            Err(TxError::CommitRejected(error)) if is_conflict(&error) => continue,
            // Never retry `CommitUnknown`: the transfer may have happened.
            outcome => return outcome,
        }
    }
    db.process(Transaction(Transfer::new(from, to, amount))).await
}

use crate::transfer::{Transfer, Transferred};
use std::convert::Infallible;
use std::sync::mpsc::Sender;
use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::deps::sqlx::PgPool;
use wakuwaku::entity::sqlx::backend::{PgSource, SqlxSource};
use wakuwaku::entity::sqlx::transaction::hook::AfterTransaction;
use wakuwaku::entity::sqlx::transaction::{Transaction, TxError, TxFinalState};
use wakuwaku::entity::{Db, ReadWrite};

/// Hands each committed transfer to the event publisher.
pub struct PublishTransfers {
    events: Sender<Transferred>,
}

impl Processor<TxFinalState<Option<Transferred>>> for PublishTransfers {
    type Output = ();
    type Error = String;

    async fn process(&self, state: TxFinalState<Option<Transferred>>) -> Result<(), String> {
        match state {
            TxFinalState::Committed(Some(event)) => {
                self.events.send(event).map_err(|error| error.to_string())
            }
            TxFinalState::CommitUnknown(Some(event)) => {
                Err(format!("{event:?} may or may not have been committed"))
            }
            // Rolled back: nothing happened, nothing to publish.
            _ => Ok(()),
        }
    }
}

/// Hook errors end here; they never change what the caller gets.
pub struct LogHookError;

impl Processor<String> for LogHookError {
    type Output = ();
    type Error = Infallible;

    async fn process(&self, error: String) -> Result<(), Infallible> {
        eprintln!("transfer hook failed: {error}");
        Ok(())
    }
}

pub type Accounts = AfterTransaction<Db<PgSource, ReadWrite>, PublishTransfers, LogHookError>;

pub fn accounts(pool: PgPool, events: Sender<Transferred>) -> Accounts {
    AfterTransaction::new(
        Db::new(SqlxSource::new(pool)),
        PublishTransfers { events },
        LogHookError,
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use wakuwaku::entity::sqlx::transaction::Either;

    #[test]
    fn no_row_debited_is_insufficient_funds() {
        assert!(matches!(Transfer::check_debit(Ok(0)), Err(TransferFailed::InsufficientFunds)));
    }

    #[test]
    fn no_row_credited_is_an_unknown_account() {
        assert!(matches!(Transfer::check_credit(Ok(0)), Err(TransferFailed::UnknownAccount)));
    }

    #[test]
    fn a_failed_query_is_a_database_error() {
        let lost = sqlx::Error::Io(std::io::Error::other("connection reset"));
        assert!(matches!(Transfer::check_credit(Err(lost)), Err(TransferFailed::Database(_))));
    }

    #[test]
    fn a_commit_reports_the_transfer() {
        let Either::Right(TxEnd::Commit { report, .. }) = Transfer::commit(1, 2, 10).finished() else {
            panic!("`commit` is a terminal commit");
        };
        assert_eq!(report, Some(Transferred { from: 1, to: 2, amount: 10 }));
    }

    #[test]
    fn a_rollback_reports_nothing() {
        assert!(matches!(
            Transfer::rollback(TransferFailed::InsufficientFunds).finished(),
            Either::Right(TxEnd::Rollback { output: TransferFailed::InsufficientFunds, report: None })
        ));
    }

    #[test]
    fn the_start_is_not_terminal() {
        assert!(Transfer::new(1, 2, 10).finished().is_left());
    }
}