backbone_banking/application/service/

banking_clearance.rs

//! Clearing engine — the bank-side GL leg (hand-authored, user-owned).
//!
//! An `impl BankingWriteService` chunk over the vocabulary in [`super::banking_write_service`]. Two
//! symmetric seams that CLEAR a matched statement line through the GL:
//!   - **`clear_transaction`** (received/paid): `Dr Bank · Cr Clearing` (or its mirror), bounded by
//!     BOTH the line's un-allocated remainder AND the settlement's un-cleared remainder (a payment
//!     cannot be cleared twice).
//!   - **`recognize_bank_charge`**: `Dr Bank Charges · Cr Bank`.
//! Both post through the supplied `GlPostSink`, write a `BankClearance` row, advance the line's
//! `allocated_amount` + status, and emit the corresponding event.
//!
//! Per the module's 4-layer rule this file holds no SQL — the line reads, the settlement advisory
//! lock + already-cleared SUM, the clearance inserts, and the allocation watermark live on
//! `BankTransactionRepository` / `BankClearanceRepository`, whose methods take this service's
//! transaction so the post + clearance + allocation advance commit as one unit.

use backbone_orm::org_scope;
use rust_decimal::Decimal;
use uuid::Uuid;

use crate::domain::entity::TxnStatus;
use crate::infrastructure::persistence::{NewChargeClearanceRow, NewClearanceRow};

use super::banking_events::{
    BankChargeRecognized, BankTransactionCleared, BankTransactionMatched, BankingEvent,
};
use super::banking_gl::{
    AccountingPostEnvelope, GlPostLine, GlPostSink, ReconcileLine, ReconcileOrigin,
    ReconcilePairRequest, ReconcileSink,
};
use super::banking_write_service::{
    money, BankingError, BankingWriteService, ClearOutcome, NewCharge, NewClearance,
};

/// The legacy tenancy twin (ADR-0029): the cleared tables hold no company column, but the GL
/// envelope, the reconcile-graph pair request, the durable outbox record, and the published
/// events still carry a `company_id` lane for unstripped consumers. Echo the ambient org
/// scope's legacy company id when the composing service bound one; nil otherwise. Nothing in
/// this module keys a statement on it.
fn legacy_company_echo() -> Uuid {
    org_scope::current_org_scope()
        .and_then(|s| s.legacy_company_id())
        .unwrap_or(Uuid::nil())
}

impl BankingWriteService {
    /// Clear a matched statement line through the GL — the bank-side leg. received (deposit):
    /// `Dr Bank · Cr Clearing`; paid (withdrawal): `Dr Clearing · Cr Bank`. Bounded: the clearance
    /// cannot exceed the line's un-allocated remainder. Records a `BankClearance`, advances the line's
    /// `allocated_amount` + status, and emits `BankTransactionMatched` + `BankTransactionCleared`.
    ///
    /// When the line was matched to a **payment**, the same unit of work also writes the
    /// reconciliation-graph edge on the clearing account: the payment's clearing leg meets the
    /// clearance's clearing leg (the pair that nets the undeposited-funds position to zero), through
    /// the shared [`ReconcileSink`] port with origin `Clearing`. Receipt: the payment DEPOSITED into
    /// clearing (`Dr Clearing`) and the clearance withdraws it (`Cr Clearing`); paid-out mirrors.
    /// **Fail-closed:** if the sink refuses the pair (unposted payment journal, a clearing account
    /// that is not flagged reconcilable, a clamp disagreement) the clearance row and the allocation
    /// advance roll back — banking refuses to record a clear whose ledger-side edge cannot exist.
    /// The clearing post itself is committed by its own sink before the edge; a refusal therefore
    /// strands that journal. The stranded post is NOT re-posted on retry: the clearance identity and
    /// its posting idempotency key are derived deterministically from the fenced cumulative-cleared
    /// position, so the retry reuses the stranded journal (the accounting sink dedups the key), the
    /// edge locator resolves against its lines, and the clear completes — the strand heals itself
    /// once the operator fixes the refusal's cause. (Re-clearing a DIFFERENT amount after a
    /// stranded attempt still lands as a residual mismatch — an integrity-probe finding, not silent
    /// money.) Matched-source kinds other than `payment` keep the bounded clearance only — their
    /// graph convergence is the statement-side reconciliation surface.
    pub async fn clear_transaction(
        &self,
        c: NewClearance,
        sink: &dyn GlPostSink,
        reconcile: &dyn ReconcileSink,
    ) -> Result<ClearOutcome, BankingError> {
        if c.matched_amount <= Decimal::ZERO {
            return Err(BankingError::NonPositiveAmount);
        }
        let matched = money(c.matched_amount);
        // Load the line + its account's GL/clearing accounts.
        // Tenancy (ADR-0029), ID-only pattern — see `propose_match`. The read rides the ambient
        // request org scope; having read the line we relay that same scope onto the clearing
        // transaction below.
        let row = self
            .repos
            .transactions
            .fetch_clearing_line(&self.rpool(), c.bank_transaction_id)
            .await?
            .ok_or(BankingError::TransactionNotFound(c.bank_transaction_id))?;
        let currency: String = row.currency;
        if currency != "IDR" {
            return Err(BankingError::UnsupportedCurrency(currency));
        }
        let company_id = legacy_company_echo();
        let deposit = row.deposit;
        let withdrawal = row.withdrawal;
        let allocated = row.allocated_amount;
        let bank_acct = row.gl_account_id;
        let clearing = row.clearing_account_id;
        let line_net = deposit + withdrawal;
        if matched > line_net - allocated {
            return Err(BankingError::OverAllocated {
                line_net,
                already: allocated,
                attempted: matched,
            });
        }
        let is_receipt = deposit > Decimal::ZERO;

        // Settlement-dimension bound (council 2026-07-05): the SUM of all clearances against this
        // settlement cannot exceed the settled document's amount — so one payment cannot be cleared
        // twice (a re-imported line, a retry, two operators). The line bound above only limits the
        // LINE; without this, two lines each matching one payment both pass and strand the clearing
        // account. Serialize per settlement with an advisory lock so concurrent first-clears can't race
        // the phantom-insert, and hold the tx across the post so the check + write are one unit.
        let mut tx = self.rpool().begin().await?;
        // Tenancy (ADR-0029): relay the AMBIENT request org scope onto this transaction when the
        // composing service bound one, so the already-cleared SUM and the clearance insert evaluate
        // under the decorator's row-level fences. An undecorated deployment has no ambient scope
        // and skips this entirely (unfenced by design).
        if let Some(scope) = org_scope::current_org_scope() {
            org_scope::bind_org_scope_on(&mut tx, &scope).await?;
            backbone_orm::audit_context::relay_ambient_audit_on(&mut tx).await?;
        }
        self.repos
            .clearances
            .lock_settlement(&mut tx, c.matched_source_id)
            .await?;
        let already_cleared = self
            .repos
            .clearances
            .sum_cleared_against_settlement(&mut tx, &c.matched_source_type, c.matched_source_id)
            .await?;
        if already_cleared + matched > money(c.matched_source_amount) {
            return Err(BankingError::SettlementOverCleared {
                settlement_amount: money(c.matched_source_amount),
                already_cleared,
                attempted: matched,
            });
        }

        // Deterministic clearance identity, derived from (line, settlement, cumulative-cleared) read
        // UNDER the settlement lock above. A refused clear rolls the whole unit back — the failed
        // attempt contributes nothing to `already_cleared` — so the operator's retry derives the SAME
        // id and the SAME posting idempotency key: the clearing post dedups at the accounting sink
        // instead of committing a second `Dr Bank · Cr Clearing`, and the edge locator resolves
        // against the first attempt's (stranded) journal lines. Legitimate sequential clears of one
        // settlement see a strictly increasing cumulative sum under the lock, so their identities
        // never collide.
        let identity = format!(
            "bankclr:{}:{}:{}:{}",
            c.bank_transaction_id, c.matched_source_type, c.matched_source_id, already_cleared
        );
        let clearance_id = Uuid::new_v5(&Uuid::NAMESPACE_URL, identity.as_bytes());
        let lines = if is_receipt {
            vec![
                GlPostLine::debit(bank_acct, matched).with_description("Bank clearing (received)"),
                GlPostLine::credit(clearing, matched).with_description("Clear undeposited funds"),
            ]
        } else {
            vec![
                GlPostLine::debit(clearing, matched).with_description("Clear undeposited funds"),
                GlPostLine::credit(bank_acct, matched).with_description("Bank clearing (paid)"),
            ]
        };
        let env = AccountingPostEnvelope {
            idempotency_key: identity,
            // The legacy tenancy twin (ADR-0029) — the composing host's GL ACL maps it onto
            // accounting's `PostingRequest.company_id`, which keeps the same legacy lane.
            company_id,
            branch_id: None,
            source_type: "settlement".into(),
            source_id: clearance_id,
            source_reference: Some(format!("clear {}", c.bank_transaction_id)),
            posting_date: c.clearance_date,
            currency,
            posting_type: "original".into(),
            reverses_post_id: None,
            description: Some("Bank clearing".into()),
            lines,
        };
        if !env.is_balanced() {
            return Err(BankingError::UnbalancedPost);
        }

        match sink.post(&env).await {
            Ok(ack) => {
                // The clearing edge (payment-matched only). The post above has committed the
                // clearance's clearing leg into the ledger; this pairs it with the payment's own
                // clearing leg on the same account, on the SAME transaction as the clearance row +
                // allocation advance below — so the graph edge and the banking bookkeeping commit
                // or roll back together. Direction follows the flow: a receipt deposits INTO
                // clearing on the payment (its leg is the debit) and the clearance withdraws
                // (credit); a payout mirrors.
                if c.matched_source_type == "payment" {
                    let payment_leg = ReconcileLine::new("payment", c.matched_source_id, clearing);
                    let clearance_leg = ReconcileLine::new("settlement", clearance_id, clearing);
                    let (debit, credit) = if is_receipt {
                        (payment_leg, clearance_leg)
                    } else {
                        (clearance_leg, payment_leg)
                    };
                    let edge = reconcile
                        .reconcile_pair_on(
                            &mut *tx,
                            &ReconcilePairRequest {
                                // The legacy tenancy twin (ADR-0029) — the shared
                                // `backbone_gl_posting` wire shape keeps its `company_id` lane
                                // for unstripped producers; nothing keys the edge on it.
                                company_id,
                                debit,
                                credit,
                                amount: matched,
                                origin: ReconcileOrigin::Clearing,
                            },
                        )
                        .await
                        .map_err(|r| BankingError::ReconcileRefused {
                            code: r.code,
                            message: r.message,
                        })?;
                    if edge.applied != matched {
                        // The graph clamped to a different amount than banking's own bound — the
                        // bank bookkeeping and the ledger disagree about what is clearable. Refuse
                        // rather than drift.
                        return Err(BankingError::ReconcileRefused {
                            code: "reconcile_clamp_mismatch".into(),
                            message: format!(
                                "banking cleared {matched} but the graph applied {}",
                                edge.applied
                            ),
                        });
                    }
                }
                let match_method = c.match_method.clone().unwrap_or_else(|| "manual".into());
                self.repos
                    .clearances
                    .insert_clearance(
                        &mut tx,
                        &NewClearanceRow {
                            id: clearance_id,
                            bank_transaction_id: c.bank_transaction_id,
                            matched_source_type: &c.matched_source_type,
                            matched_source_id: c.matched_source_id,
                            matched_amount: matched,
                            match_method: &match_method,
                            clearance_date: c.clearance_date,
                            accounting_post_id: ack.post_id,
                            journal_id: ack.journal_id,
                        },
                    )
                    .await?;
                let new_alloc = allocated + matched;
                let fully = new_alloc >= line_net;
                let status = if fully {
                    TxnStatus::Reconciled
                } else {
                    TxnStatus::PartlyReconciled
                };
                self.repos
                    .transactions
                    .set_allocation(&mut tx, c.bank_transaction_id, new_alloc, status)
                    .await?;
                // Durable `BankClearanceRecorded` staging (payment-matched only — the event's only
                // consumer is payment's bank-confirmation drift). Inside THIS transaction: a crash
                // after the clearance can never lose the confirmation event, and a rolled-back clear
                // never emits one. The relay later delivers it; payment's `confirm_cash_once`
                // consumer dedups on the event id.
                if let Some(schema) = self.outbox_schema.clone() {
                    if c.matched_source_type == "payment" {
                        // `company_id` in the payload and on the outbox record is the legacy
                        // tenancy twin (ADR-0029): the consumer (payment's bank-confirmation
                        // drift) keeps the lane; nothing keys delivery on it.
                        let payload = serde_json::json!({
                            "payment_id": c.matched_source_id.to_string(),
                            "company_id": company_id.to_string(),
                            "bank_transaction_id": c.bank_transaction_id.to_string(),
                            "matched_source_type": c.matched_source_type,
                            "matched_source_id": c.matched_source_id.to_string(),
                            "amount": matched.to_string(),
                            "journal_id": ack.journal_id.to_string(),
                            "post_id": ack.post_id.to_string(),
                        });
                        let rec = backbone_outbox::OutboxRecord::new(
                            "BankClearanceRecorded",
                            "BankClearance",
                            clearance_id.to_string(),
                            company_id,
                            payload,
                            chrono::Utc::now(),
                        );
                        backbone_outbox::outbox::stage(&mut *tx, &schema, &rec)
                            .await
                            .map_err(|e| BankingError::Db(sqlx::Error::Protocol(e.to_string())))?;
                    }
                }
                tx.commit().await?;

                self.sink.publish(BankingEvent::BankTransactionMatched(
                    BankTransactionMatched {
                        bank_transaction_id: c.bank_transaction_id,
                        matched_source_type: c.matched_source_type.clone(),
                        matched_source_id: c.matched_source_id,
                        amount: matched,
                    },
                ));
                self.sink.publish(BankingEvent::BankTransactionCleared(
                    BankTransactionCleared {
                        bank_transaction_id: c.bank_transaction_id,
                        matched_source_type: c.matched_source_type,
                        matched_source_id: c.matched_source_id,
                        company_id,
                        journal_id: ack.journal_id,
                        post_id: ack.post_id,
                        amount: matched,
                    },
                ));
                Ok(ClearOutcome {
                    clearance_id,
                    post_id: ack.post_id,
                    journal_id: ack.journal_id,
                    fully_reconciled: fully,
                })
            }
            Err(rej) => Err(BankingError::GlRejected {
                code: rej.code,
                message: rej.message,
            }),
        }
    }

    /// Recognise an outflow line as a bank charge: `Dr Bank Charges · Cr Bank`. Marks the line
    /// reconciled and emits `BankChargeRecognized`.
    pub async fn recognize_bank_charge(
        &self,
        ch: NewCharge,
        sink: &dyn GlPostSink,
    ) -> Result<ClearOutcome, BankingError> {
        if ch.amount <= Decimal::ZERO {
            return Err(BankingError::NonPositiveAmount);
        }
        let amount = money(ch.amount);
        // Tenancy (ADR-0029), ID-only pattern — see `propose_match`; the charge tx below relays
        // the ambient request org scope.
        let row = self
            .repos
            .transactions
            .fetch_charge_line(&self.rpool(), ch.bank_transaction_id)
            .await?
            .ok_or(BankingError::TransactionNotFound(ch.bank_transaction_id))?;
        let currency: String = row.currency;
        if currency != "IDR" {
            return Err(BankingError::UnsupportedCurrency(currency));
        }
        let company_id = legacy_company_echo();
        let allocated = row.allocated_amount;
        let line_net = row.deposit + row.withdrawal;
        let bank_acct = row.gl_account_id;
        if amount > line_net - allocated {
            return Err(BankingError::OverAllocated {
                line_net,
                already: allocated,
                attempted: amount,
            });
        }
        let clearance_id = Uuid::new_v4();
        let env = AccountingPostEnvelope {
            idempotency_key: format!("bankchg:{}:{}", ch.bank_transaction_id, clearance_id),
            // The legacy tenancy twin (ADR-0029) — see `clear_transaction`.
            company_id,
            branch_id: None,
            source_type: "settlement".into(),
            source_id: clearance_id,
            source_reference: Some(format!("charge {}", ch.bank_transaction_id)),
            posting_date: ch.clearance_date,
            currency,
            posting_type: "original".into(),
            reverses_post_id: None,
            description: Some("Bank charge".into()),
            lines: vec![
                GlPostLine::debit(ch.charge_account_id, amount).with_description("Bank charges"),
                GlPostLine::credit(bank_acct, amount).with_description("Bank"),
            ],
        };
        if !env.is_balanced() {
            return Err(BankingError::UnbalancedPost);
        }
        match sink.post(&env).await {
            Ok(ack) => {
                let mut tx = self.rpool().begin().await?;
                // Tenancy (ADR-0029): relay the AMBIENT request org scope — see `clear_transaction`.
                if let Some(scope) = org_scope::current_org_scope() {
                    org_scope::bind_org_scope_on(&mut tx, &scope).await?;
                    backbone_orm::audit_context::relay_ambient_audit_on(&mut tx).await?;
                }
                self.repos
                    .clearances
                    .insert_charge_clearance(
                        &mut tx,
                        &NewChargeClearanceRow {
                            id: clearance_id,
                            bank_transaction_id: ch.bank_transaction_id,
                            charge_account_id: ch.charge_account_id,
                            matched_amount: amount,
                            clearance_date: ch.clearance_date,
                            accounting_post_id: ack.post_id,
                            journal_id: ack.journal_id,
                        },
                    )
                    .await?;
                let new_alloc = allocated + amount;
                let fully = new_alloc >= line_net;
                self.repos
                    .transactions
                    .set_allocation(
                        &mut tx,
                        ch.bank_transaction_id,
                        new_alloc,
                        if fully {
                            TxnStatus::Reconciled
                        } else {
                            TxnStatus::PartlyReconciled
                        },
                    )
                    .await?;
                tx.commit().await?;
                self.sink
                    .publish(BankingEvent::BankChargeRecognized(BankChargeRecognized {
                        bank_transaction_id: ch.bank_transaction_id,
                        company_id,
                        amount,
                        journal_id: ack.journal_id,
                        post_id: ack.post_id,
                    }));
                Ok(ClearOutcome {
                    clearance_id,
                    post_id: ack.post_id,
                    journal_id: ack.journal_id,
                    fully_reconciled: fully,
                })
            }
            Err(rej) => Err(BankingError::GlRejected {
                code: rej.code,
                message: rej.message,
            }),
        }
    }
}