backbone_pos/infrastructure/persistence/

pos_invoice_repository.rs

//! Repository for PosInvoice entities
//!
//! Originally generated by metaphor-schema; now **user-owned** — this exact path is declared under
//! `user_owned` in `metaphor.codegen.yaml`, so the generator skips it wholesale. The custom methods
//! below hold the hand-written counter-ticket SQL (4-layer rule: services orchestrate, repos hold SQL).
//!
//! Thin newtype over `backbone_orm::GenericCrudRepository<PosInvoice, backbone_orm::SoftDelete>`.
//! All standard CRUD methods are available via `Deref`.
//!
//! Tenancy is composition-installed (ADR-0029): pool reads ride the ambient org scope's
//! request-dedicated connection (fail-narrow, never leaky — a row the scope cannot see is simply
//! absent); connection-taking methods run on the CALLER'S open, scope-relayed transaction.

use anyhow::Result;
use rust_decimal::Decimal;
use sqlx::{PgPool, Row};
use uuid::Uuid;

use backbone_orm::org_scope;
// The multi-row and scalar read twins live only in the legacy `company_scope` module. Their
// connection discipline is what this repository needs — request-dedicated connection when the
// composing service bound one, plain pool otherwise. The helper's legacy task-local branch is
// never taken: this module sets no legacy scope of its own (ADR-0029).
use backbone_orm::company_scope::{
    fetch_one_row_scoped, fetch_one_scalar_scoped, fetch_optional_scalar_scoped,
};

use crate::domain::entity::PosInvoice;

/// Table name for PosInvoice entities
pub const TABLE_NAME: &str = "pos.pos_invoices";

/// Repository for PosInvoice entities.
///
/// All standard CRUD, soft-delete, pagination, and bulk methods are
/// provided automatically via `Deref` to `backbone_orm::GenericCrudRepository`.
pub struct PosInvoiceRepository(
    backbone_orm::GenericCrudRepository<PosInvoice, backbone_orm::SoftDelete>,
);

impl std::ops::Deref for PosInvoiceRepository {
    type Target = backbone_orm::GenericCrudRepository<PosInvoice, backbone_orm::SoftDelete>;
    fn deref(&self) -> &Self::Target { &self.0 }
}

impl PosInvoiceRepository {
    /// Create a new repository instance.
    pub fn new(pool: PgPool) -> Self {
        Self(backbone_orm::GenericCrudRepository::new(pool, TABLE_NAME))
    }
}

/// The exact row a rung ticket writes. Mirrors the raw column shape rather than the `PosInvoice`
/// entity: the money fields are already computed by the service, and `paid_total`/`change_due`/
/// `is_return`/`status` are fixed by the SQL (`0,0,false,'draft'`). `client_uuid` carries the offline
/// sync identity when the ticket is being replayed from a client (server-originated rings pass None).
/// `pos_table_id` seats the ticket at a dining table (restaurant lanes); counter sales pass None.
pub struct NewDraftInvoiceRow<'a> {
    pub id: Uuid,
    pub client_uuid: Option<Uuid>,
    pub pos_profile_id: Uuid,
    pub opening_entry_id: Uuid,
    pub branch_id: Option<Uuid>,
    pub customer_id: Option<Uuid>,
    pub pos_table_id: Option<Uuid>,
    pub receipt_number: &'a str,
    pub posting_at: chrono::NaiveDateTime,
    pub net_total: Decimal,
    pub tax_total: Decimal,
    pub grand_total: Decimal,
    pub rounding_adjustment: Decimal,
    pub rounded_total: Decimal,
}

/// A ticket looked up by its offline-sync identity — everything `sync_from_ui` needs to decide
/// create-vs-update-vs-replay: identity anchors (session, partner), the refund lineage, and the
/// persisted money (returned as the replay's totals when the ticket is already finalized).
pub struct SyncLookupRow {
    pub id: Uuid,
    pub status: String,
    pub pos_profile_id: Uuid,
    pub opening_entry_id: Uuid,
    pub branch_id: Option<Uuid>,
    pub customer_id: Option<Uuid>,
    /// The table the draft is currently seated at (for transfer validation on replay).
    pub pos_table_id: Option<Uuid>,
    pub is_return: bool,
    /// For a refund ticket: the server id of the parent it was recorded against.
    pub return_against: Option<Uuid>,
    pub net_total: Decimal,
    pub tax_total: Decimal,
    pub grand_total: Decimal,
    pub rounding_adjustment: Decimal,
    pub rounded_total: Decimal,
    pub paid_total: Decimal,
    pub change_due: Decimal,
}

/// The exact row a return ticket writes. Mirrors the raw column shape: the SQL fixes
/// `rounding_adjustment` to 0, binds `rounded_total` to both the rounded and paid totals ($12), fixes
/// `change_due` to 0, and fixes `is_return`/`status` to `true`/`'returned'`.
pub struct NewReturnInvoiceRow {
    pub id: Uuid,
    pub pos_profile_id: Uuid,
    pub opening_entry_id: Uuid,
    pub branch_id: Option<Uuid>,
    pub customer_id: Option<Uuid>,
    pub receipt_number: String,
    pub posting_at: chrono::DateTime<chrono::Utc>,
    pub net_total: Decimal,
    pub tax_total: Decimal,
    pub grand_total: Decimal,
    pub rounded_total: Decimal,
    pub billing_invoice_id: Uuid,
    pub return_against: Uuid,
}

/// A ticket header as `add_tender` reads it: the draft gate and the target total.
pub struct TenderHeaderRow {
    pub status: String,
    pub rounded_total: Decimal,
}

/// A ticket's already-recognised state, as the recognise short-circuit reads it.
pub struct PaidStateRow {
    pub status: String,
    pub billing_invoice_id: Option<Uuid>,
    pub payment_entry_id: Option<Uuid>,
}

/// Everything recognition needs about a draft ticket + its register, in one read: the money to settle,
/// the accounts to post to, and the stock dimensions to relieve.
pub struct RecognitionRow {
    pub customer_id: Option<Uuid>,
    pub branch_id: Option<Uuid>,
    pub posting_at: chrono::DateTime<chrono::Utc>,
    pub rounded_total: Decimal,
    pub grand_total: Decimal,
    pub tax_total: Decimal,
    pub paid_total: Decimal,
    pub billing_invoice_id: Option<Uuid>,
    pub receivable_account_id: Option<Uuid>,
    pub income_account_id: Option<Uuid>,
    pub cash_account_id: Option<Uuid>,
    pub tax_account_id: Option<Uuid>,
    pub tax_rate: Decimal,
    pub warehouse_id: Option<Uuid>,
    pub cogs_account_id: Option<Uuid>,
    pub inventory_account_id: Option<Uuid>,
    pub currency: String,
    pub default_customer_id: Option<Uuid>,
}

/// The original ticket as the returns path reads it — the source of every field the return ticket
/// copies, plus the tender/invoice refs to reverse.
pub struct ReturnSourceRow {
    pub pos_profile_id: Uuid,
    pub opening_entry_id: Uuid,
    pub branch_id: Option<Uuid>,
    pub customer_id: Option<Uuid>,
    pub billing_invoice_id: Option<Uuid>,
    pub payment_entry_id: Option<Uuid>,
    pub rounded_total: Decimal,
    pub net_total: Decimal,
    pub tax_total: Decimal,
    pub grand_total: Decimal,
    pub posting_at: chrono::DateTime<chrono::Utc>,
    pub status: String,
}

/// A ticket header as the printed receipt reads it (register name/currency joined in). `tax_rate` is
/// the EFFECTIVE rate derived from the ticket's own totals — the register's flat `tax_rate` column is
/// retired from the compute path, so the slip reports what the document-grade compute actually
/// produced. `is_invoiced` derives from the billing link (never a stored state column).
pub struct ReceiptHeaderRow {
    pub receipt_number: String,
    pub posting_at: chrono::DateTime<chrono::Utc>,
    pub net_total: Decimal,
    pub tax_total: Decimal,
    pub rounding_adjustment: Decimal,
    pub rounded_total: Decimal,
    pub paid_total: Decimal,
    pub change_due: Decimal,
    pub status: String,
    pub register_name: String,
    pub currency: String,
    pub tax_rate: Decimal,
    pub is_invoiced: bool,
}

/// Hand-written counter-ticket SQL. Lives here (not in the write service) per the module's 4-layer
/// rule: services orchestrate and own the unit of work, repositories hold the SQL.
impl PosInvoiceRepository {
    /// Insert a rung ticket in `draft`.
    ///
    /// Takes the CALLER'S connection so the header and its lines commit as one unit. The caller has
    /// already relayed the ambient org scope onto it — don't re-bind here.
    ///
    /// Returns the raw `sqlx::Error` deliberately: the caller inspects it for a unique violation to turn
    /// a re-used receipt number (or a re-used sync `client_uuid`) into a domain error.
    pub async fn insert_draft(
        &self,
        conn: &mut sqlx::PgConnection,
        i: &NewDraftInvoiceRow<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"INSERT INTO pos.pos_invoices
                (id, client_uuid, pos_profile_id, opening_entry_id, branch_id, customer_id, pos_table_id, receipt_number,
                 posting_at, net_total, tax_total, grand_total, rounding_adjustment, rounded_total,
                 paid_total, change_due, is_return, status)
               VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,0,0,false,'draft'::pos_invoice_status)"#,
        )
        .bind(i.id).bind(i.client_uuid).bind(i.pos_profile_id).bind(i.opening_entry_id).bind(i.branch_id)
        .bind(i.customer_id).bind(i.pos_table_id).bind(i.receipt_number).bind(i.posting_at).bind(i.net_total).bind(i.tax_total)
        .bind(i.grand_total).bind(i.rounding_adjustment).bind(i.rounded_total)
        .execute(conn)
        .await?;
        Ok(())
    }

    /// Find the live ticket carrying an offline client uuid. The uuid only namespaces inside the
    /// caller's scope: the ambient org scope's fence is the whole guard — another unit's ticket
    /// carrying the same uuid is simply not found.
    pub async fn find_by_client_uuid(
        &self,
        pool: &PgPool,
        client_uuid: Uuid,
    ) -> Result<Option<SyncLookupRow>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT id, status::text AS st, pos_profile_id, opening_entry_id, branch_id, customer_id,
                          pos_table_id, is_return, return_against, net_total, tax_total, grand_total, rounding_adjustment,
                          rounded_total, paid_total, change_due
                   FROM pos.pos_invoices WHERE client_uuid=$1 AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(client_uuid),
        )
        .await?;
        Ok(row.map(|r| SyncLookupRow {
            id: r.get("id"), status: r.get("st"), pos_profile_id: r.get("pos_profile_id"),
            opening_entry_id: r.get("opening_entry_id"), branch_id: r.get("branch_id"),
            customer_id: r.get("customer_id"), pos_table_id: r.get("pos_table_id"), is_return: r.get("is_return"),
            return_against: r.get("return_against"), net_total: r.get("net_total"),
            tax_total: r.get("tax_total"), grand_total: r.get("grand_total"),
            rounding_adjustment: r.get("rounding_adjustment"), rounded_total: r.get("rounded_total"),
            paid_total: r.get("paid_total"), change_due: r.get("change_due"),
        }))
    }

    /// Rewrite a DRAFT ticket's header for an offline replay: re-point it at the (possibly rescued)
    /// session, re-seat it at the replay's table (a changed `pos_table_id` is a table TRANSFER), and
    /// replace its server-derived money + partner with the replay's re-derived values.
    /// State-guarded on `draft` — a finalized ticket is never rewritten (the replay short-circuits
    /// before this). Takes the CALLER'S connection; the caller has already relayed the ambient scope.
    /// Returns the raw `sqlx::Error` so the caller can map the one-draft-per-table partial unique.
    pub async fn update_draft_from_sync(
        &self,
        conn: &mut sqlx::PgConnection,
        pos_invoice_id: Uuid,
        opening_entry_id: Uuid,
        branch_id: Option<Uuid>,
        customer_id: Option<Uuid>,
        pos_table_id: Option<Uuid>,
        net_total: Decimal,
        tax_total: Decimal,
        grand_total: Decimal,
        rounding_adjustment: Decimal,
        rounded_total: Decimal,
        paid_total: Decimal,
        change_due: Decimal,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"UPDATE pos.pos_invoices SET
                 opening_entry_id=$2, branch_id=$3, customer_id=$4, pos_table_id=$5,
                 net_total=$6, tax_total=$7, grand_total=$8, rounding_adjustment=$9,
                 rounded_total=$10, paid_total=$11, change_due=$12
               WHERE id=$1 AND status='draft'::pos_invoice_status"#,
        )
        .bind(pos_invoice_id).bind(opening_entry_id).bind(branch_id).bind(customer_id).bind(pos_table_id)
        .bind(net_total).bind(tax_total).bind(grand_total).bind(rounding_adjustment)
        .bind(rounded_total).bind(paid_total).bind(change_due)
        .execute(conn)
        .await?;
        Ok(())
    }

    /// The draft ticket currently seated at a dining table — the friendly half of the
    /// one-draft-per-table guard (the DB partial unique `idx_pos_invoices_pos_table_id` is the
    /// backstop; this read turns the collision into a typed refusal naming the occupying ticket).
    /// `exclude_invoice_id` skips the ticket being re-seated (a replay that keeps its own table is
    /// not a collision). `Ok(None)` = the table is free. Takes the CALLER'S scope-relayed
    /// connection — the fence is the whole guard.
    pub async fn find_draft_on_table(
        &self,
        conn: &mut sqlx::PgConnection,
        pos_table_id: Uuid,
        exclude_invoice_id: Option<Uuid>,
    ) -> Result<Option<Uuid>, sqlx::Error> {
        let row = sqlx::query(
            r#"SELECT id FROM pos.pos_invoices
               WHERE pos_table_id=$1 AND status='draft'::pos_invoice_status
                 AND (metadata->>'deleted_at') IS NULL AND ($2::uuid IS NULL OR id <> $2)
               LIMIT 1"#,
        )
        .bind(pos_table_id)
        .bind(exclude_invoice_id)
        .fetch_optional(conn)
        .await?;
        Ok(row.map(|r| r.get("id")))
    }

    /// The pool twin of [`Self::find_draft_on_table`] for the duplicate-mapping enrichment: after a
    /// failed insert the caller's transaction is ABORTED, so the occupying ticket must be read on a
    /// fresh connection. Rides the ambient org scope (ADR-0029) — under a composed fence the read
    /// stays inside the acting unit, and a fence-blind read would enrich the 409 with a nil id.
    pub async fn find_draft_on_table_scoped(
        &self,
        pool: &PgPool,
        pos_table_id: Uuid,
        exclude_invoice_id: Option<Uuid>,
    ) -> Result<Option<Uuid>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT id FROM pos.pos_invoices
                   WHERE pos_table_id=$1 AND status='draft'::pos_invoice_status
                     AND (metadata->>'deleted_at') IS NULL AND ($2::uuid IS NULL OR id <> $2)
                   LIMIT 1"#,
            )
            .bind(pos_table_id)
            .bind(exclude_invoice_id),
        )
        .await?;
        Ok(row.map(|r| r.get("id")))
    }

    /// Soft-delete a DRAFT ticket's lines so an offline replay can rewrite the basket (full-snapshot
    /// semantics: the replay carries the complete ticket). Soft-delete keeps the audit trail and frees
    /// the partial-unique on `client_uuid`. Takes the CALLER'S connection.
    pub async fn soft_delete_lines_for_ticket(
        &self,
        conn: &mut sqlx::PgConnection,
        pos_invoice_id: Uuid,
        now: chrono::DateTime<chrono::Utc>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"UPDATE pos.pos_invoice_items SET metadata = jsonb_set(metadata, '{deleted_at}', to_jsonb($2::timestamptz))
               WHERE pos_invoice_id=$1 AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(pos_invoice_id).bind(now)
        .execute(conn)
        .await?;
        Ok(())
    }

    /// Count a session's DRAFT tickets — the close guard. ID-only: rides the ambient org scope.
    pub async fn count_draft_orders_for_session(
        &self,
        pool: &PgPool,
        opening_entry_id: Uuid,
    ) -> Result<i64, sqlx::Error> {
        fetch_one_scalar_scoped(
            pool,
            sqlx::query_scalar(
                "SELECT COUNT(*) FROM pos.pos_invoices WHERE opening_entry_id=$1 AND status='draft'::pos_invoice_status AND is_return=false AND (metadata->>'deleted_at') IS NULL",
            )
            .bind(opening_entry_id),
        )
        .await
    }

    /// Count a session's half-recognised tickets — billing linked but the draft→paid flip never
    /// landed (a settle failed after the invoice was raised). The close guard refuses on these: they
    /// need recognition retried, not skipped. ID-only scope as above.
    pub async fn count_linked_unposted_for_session(
        &self,
        pool: &PgPool,
        opening_entry_id: Uuid,
    ) -> Result<i64, sqlx::Error> {
        fetch_one_scalar_scoped(
            pool,
            sqlx::query_scalar(
                "SELECT COUNT(*) FROM pos.pos_invoices WHERE opening_entry_id=$1 AND status='draft'::pos_invoice_status AND billing_invoice_id IS NOT NULL AND (metadata->>'deleted_at') IS NULL",
            )
            .bind(opening_entry_id),
        )
        .await
    }

    /// Read the header `add_tender` gates on.
    ///
    /// ID-only: rides the ambient org scope's request-dedicated connection, so a ticket the scope
    /// cannot see simply is not found. The caller runs this on its own scope-relayed transaction
    /// (`fetch_tender_header_on` shape inlined: same statement, caller's connection).
    pub async fn fetch_tender_header(
        &self,
        conn: &mut sqlx::PgConnection,
        pos_invoice_id: Uuid,
    ) -> Result<Option<TenderHeaderRow>, sqlx::Error> {
        let row = sqlx::query(
            "SELECT status::text AS st, rounded_total FROM pos.pos_invoices WHERE id=$1 AND (metadata->>'deleted_at') IS NULL",
        )
        .bind(pos_invoice_id)
        .fetch_optional(conn)
        .await?;
        Ok(row.map(|r| TenderHeaderRow {
            status: r.get("st"), rounded_total: r.get("rounded_total"),
        }))
    }

    /// Write back a ticket's recomputed tender totals. Takes the CALLER'S connection so this and the
    /// tender insert commit as one unit; the caller has already relayed the ambient scope on it.
    pub async fn update_tender_totals(
        &self,
        conn: &mut sqlx::PgConnection,
        pos_invoice_id: Uuid,
        paid_total: Decimal,
        change_due: Decimal,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("UPDATE pos.pos_invoices SET paid_total=$2, change_due=$3 WHERE id=$1")
            .bind(pos_invoice_id).bind(paid_total).bind(change_due)
            .execute(conn)
            .await?;
        Ok(())
    }

    /// Read a ticket's recognised state, for the recognise short-circuit. Same ID-only scope contract
    /// as [`Self::fetch_tender_header`].
    pub async fn fetch_paid_state(
        &self,
        pool: &PgPool,
        pos_invoice_id: Uuid,
    ) -> Result<Option<PaidStateRow>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                "SELECT status::text AS st, billing_invoice_id, payment_entry_id FROM pos.pos_invoices WHERE id=$1 AND (metadata->>'deleted_at') IS NULL",
            )
            .bind(pos_invoice_id),
        )
        .await?;
        Ok(row.map(|r| PaidStateRow {
            status: r.get("st"), billing_invoice_id: r.get("billing_invoice_id"),
            payment_entry_id: r.get("payment_entry_id"),
        }))
    }

    /// Read a DRAFT ticket joined to its register — everything recognition needs to drive billing,
    /// payment, and the optional stock issue. `Ok(None)` = absent, already recognised, or outside the
    /// caller's scope.
    ///
    /// ID-only: rides the ambient org scope. Under HTTP the request-dedicated connection carries the
    /// scope. When driven by an EVENT (the recognition sink subscribing to `PosTenderCompleted`), the
    /// caller MUST wrap this in the composing service's org request scope — bound from the event's
    /// `company_id` twin (the acting unit) — otherwise this read fails closed.
    pub async fn fetch_for_recognition(
        &self,
        pool: &PgPool,
        pos_invoice_id: Uuid,
    ) -> Result<Option<RecognitionRow>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT i.customer_id, i.pos_profile_id, i.branch_id, i.posting_at, i.rounded_total, i.grand_total,
                          i.tax_total, i.paid_total, i.billing_invoice_id,
                          p.receivable_account_id, p.income_account_id, p.cash_account_id, p.tax_account_id,
                          ROUND(COALESCE(i.tax_total / NULLIF(i.net_total, 0), 0), 6) AS tax_rate,
                          p.warehouse_id, p.cogs_account_id, p.inventory_account_id,
                          p.currency, p.default_customer_id
                   FROM pos.pos_invoices i JOIN pos.pos_profiles p ON p.id = i.pos_profile_id
                   WHERE i.id=$1 AND i.status='draft'::pos_invoice_status AND (i.metadata->>'deleted_at') IS NULL"#,
            )
            .bind(pos_invoice_id),
        )
        .await?;
        Ok(row.map(|r| RecognitionRow {
            customer_id: r.get("customer_id"), branch_id: r.get("branch_id"),
            posting_at: r.get("posting_at"), rounded_total: r.get("rounded_total"), grand_total: r.get("grand_total"),
            tax_total: r.get("tax_total"), paid_total: r.get("paid_total"),
            billing_invoice_id: r.get("billing_invoice_id"),
            receivable_account_id: r.get("receivable_account_id"), income_account_id: r.get("income_account_id"),
            cash_account_id: r.get("cash_account_id"), tax_account_id: r.get("tax_account_id"),
            tax_rate: r.get("tax_rate"), warehouse_id: r.get("warehouse_id"), cogs_account_id: r.get("cogs_account_id"),
            inventory_account_id: r.get("inventory_account_id"), currency: r.get("currency"),
            default_customer_id: r.get("default_customer_id"),
        }))
    }

    /// Persist the raised Sales Invoice onto the STILL-DRAFT ticket, before settle.
    ///
    /// This is billing's at-most-once gate (council 2026-07-05): writing the link the instant
    /// `raise_and_post` returns makes a retry after a settle failure (declined card / closed period)
    /// REUSE the invoice instead of raising a second one — otherwise the counter's ordinary
    /// decline-then-retry doubles the revenue journal. State-guarded on `draft`. Rides the ambient
    /// org scope.
    pub async fn link_billing_invoice(
        &self,
        pool: &PgPool,
        pos_invoice_id: Uuid,
        billing_invoice_id: Uuid,
    ) -> Result<(), sqlx::Error> {
        org_scope::execute_scoped(
            pool,
            sqlx::query("UPDATE pos.pos_invoices SET billing_invoice_id=$2 WHERE id=$1 AND status='draft'::pos_invoice_status")
                .bind(pos_invoice_id).bind(billing_invoice_id),
        )
        .await?;
        Ok(())
    }

    /// Flip a ticket draft→paid, persisting both downstream refs. Returns the rows affected: the caller
    /// gates the `PosInvoicePaid` event and the stock issue on this being 1, so each happens exactly
    /// once even if recognise is retried.
    ///
    /// Persisting `payment_entry_id` in the same flip is what lets a later return refund the tender from
    /// the ticket (no cross-schema lookup) and gives a crash-retry a durable skip-gate (ADR-001
    /// `settle`-idempotency). Rides the ambient org scope.
    pub async fn mark_paid(
        &self,
        pool: &PgPool,
        pos_invoice_id: Uuid,
        billing_invoice_id: Uuid,
        payment_entry_id: Uuid,
    ) -> Result<u64, sqlx::Error> {
        let res = org_scope::execute_scoped(
            pool,
            sqlx::query(
                "UPDATE pos.pos_invoices SET status='paid'::pos_invoice_status, billing_invoice_id=$2, payment_entry_id=$3 WHERE id=$1 AND status='draft'::pos_invoice_status",
            )
            .bind(pos_invoice_id).bind(billing_invoice_id).bind(payment_entry_id),
        )
        .await?;
        Ok(res.rows_affected())
    }

    /// Read the original ticket the returns path reverses and copies. Same ID-only scope contract as
    /// [`Self::fetch_tender_header`].
    pub async fn fetch_return_source(
        &self,
        pool: &PgPool,
        pos_invoice_id: Uuid,
    ) -> Result<Option<ReturnSourceRow>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT pos_profile_id, opening_entry_id, branch_id, customer_id, billing_invoice_id,
                          payment_entry_id, rounded_total, net_total, tax_total, grand_total, posting_at, status::text AS st
                   FROM pos.pos_invoices WHERE id=$1 AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(pos_invoice_id),
        )
        .await?;
        Ok(row.map(|r| ReturnSourceRow {
            pos_profile_id: r.get("pos_profile_id"),
            opening_entry_id: r.get("opening_entry_id"), branch_id: r.get("branch_id"),
            customer_id: r.get("customer_id"), billing_invoice_id: r.get("billing_invoice_id"),
            payment_entry_id: r.get("payment_entry_id"), rounded_total: r.get("rounded_total"),
            net_total: r.get("net_total"), tax_total: r.get("tax_total"), grand_total: r.get("grand_total"),
            posting_at: r.get("posting_at"), status: r.get("st"),
        }))
    }

    /// Flip the original ticket paid→returned. Returns the rows affected: the caller gates the return
    /// ticket and the `PosInvoiceReturned` event on this being 1, so a repeat return records at most
    /// once. Rides the ambient org scope.
    pub async fn mark_returned(
        &self,
        pool: &PgPool,
        pos_invoice_id: Uuid,
    ) -> Result<u64, sqlx::Error> {
        let res = org_scope::execute_scoped(
            pool,
            sqlx::query("UPDATE pos.pos_invoices SET status='returned'::pos_invoice_status WHERE id=$1 AND status='paid'::pos_invoice_status")
                .bind(pos_invoice_id),
        )
        .await?;
        Ok(res.rows_affected())
    }

    /// Insert the mirror return ticket, linked to the original via `return_against`. Rides the ambient
    /// org scope.
    pub async fn insert_return_ticket(
        &self,
        pool: &PgPool,
        r: &NewReturnInvoiceRow,
    ) -> Result<(), sqlx::Error> {
        org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"INSERT INTO pos.pos_invoices
                    (id, pos_profile_id, opening_entry_id, branch_id, customer_id, receipt_number,
                     posting_at, net_total, tax_total, grand_total, rounding_adjustment, rounded_total,
                     paid_total, change_due, billing_invoice_id, is_return, return_against, status)
                   VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,0,$11,$11,0,$12,true,$13,'returned'::pos_invoice_status)"#,
            )
            .bind(r.id).bind(r.pos_profile_id).bind(r.opening_entry_id)
            .bind(r.branch_id).bind(r.customer_id).bind(&r.receipt_number)
            .bind(r.posting_at).bind(r.net_total).bind(r.tax_total).bind(r.grand_total).bind(r.rounded_total)
            .bind(r.billing_invoice_id).bind(r.return_against),
        )
        .await?;
        Ok(())
    }

    /// Find the return ticket already recorded against an original — the idempotent-replay path when the
    /// paid→returned flip found nothing to do. Rides the ambient org scope.
    pub async fn find_return_ticket(
        &self,
        pool: &PgPool,
        original_pos_invoice_id: Uuid,
    ) -> Result<Option<Uuid>, sqlx::Error> {
        fetch_optional_scalar_scoped(
            pool,
            sqlx::query_scalar(
                "SELECT id FROM pos.pos_invoices WHERE return_against=$1 AND is_return=true AND (metadata->>'deleted_at') IS NULL LIMIT 1",
            )
            .bind(original_pos_invoice_id),
        )
        .await
    }

    /// Sum the cash handed back as change across a session's recognised tickets — it reduces the drawer.
    /// ID-only: rides the ambient org scope.
    pub async fn sum_change_due_for_session(
        &self,
        pool: &PgPool,
        opening_entry_id: Uuid,
    ) -> Result<Decimal, sqlx::Error> {
        fetch_one_scalar_scoped(
            pool,
            sqlx::query_scalar(
                "SELECT COALESCE(SUM(change_due),0) FROM pos.pos_invoices WHERE opening_entry_id=$1 AND status='paid'::pos_invoice_status AND (metadata->>'deleted_at') IS NULL",
            )
            .bind(opening_entry_id),
        )
        .await
    }

    /// A session's recognised turnover + paid-ticket count, for the X/Z report. ID-only: rides the
    /// ambient org scope.
    pub async fn session_totals(
        &self,
        pool: &PgPool,
        opening_entry_id: Uuid,
    ) -> Result<(Decimal, i64), sqlx::Error> {
        let row = fetch_one_row_scoped(
            pool,
            sqlx::query(
                "SELECT COALESCE(SUM(rounded_total),0), COUNT(*) FROM pos.pos_invoices WHERE opening_entry_id=$1 AND status='paid'::pos_invoice_status AND (metadata->>'deleted_at') IS NULL",
            )
            .bind(opening_entry_id),
        )
        .await?;
        Ok((row.get(0), row.get(1)))
    }

    /// Read a ticket header joined to its register for the printed receipt. `Ok(None)` = no such
    /// ticket in the caller's scope. Rides the ambient org scope — the fence is the whole guard.
    pub async fn fetch_receipt_header(
        &self,
        pool: &PgPool,
        pos_invoice_id: Uuid,
    ) -> Result<Option<ReceiptHeaderRow>, sqlx::Error> {
        let row = org_scope::fetch_optional_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT i.receipt_number, i.posting_at, i.net_total, i.tax_total, i.rounding_adjustment,
                          i.rounded_total, i.paid_total, i.change_due, i.status::text AS status,
                          p.name AS register_name, p.currency,
                          ROUND(COALESCE(i.tax_total / NULLIF(i.net_total, 0), 0), 6) AS tax_rate,
                          (i.billing_invoice_id IS NOT NULL) AS is_invoiced
                   FROM pos.pos_invoices i JOIN pos.pos_profiles p ON p.id = i.pos_profile_id
                   WHERE i.id=$1 AND (i.metadata->>'deleted_at') IS NULL"#,
            )
            .bind(pos_invoice_id),
        )
        .await?;
        Ok(row.map(|r| ReceiptHeaderRow {
            receipt_number: r.get("receipt_number"), posting_at: r.get("posting_at"),
            net_total: r.get("net_total"), tax_total: r.get("tax_total"),
            rounding_adjustment: r.get("rounding_adjustment"), rounded_total: r.get("rounded_total"),
            paid_total: r.get("paid_total"), change_due: r.get("change_due"), status: r.get("status"),
            register_name: r.get("register_name"), currency: r.get("currency"), tax_rate: r.get("tax_rate"),
            is_invoiced: r.get("is_invoiced"),
        }))
    }
}

backbone_core::impl_crud_repository!(PosInvoiceRepository, PosInvoice, soft_delete);