backbone-pos 0.7.0

//! Repository for PosOpeningEntry 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 cashier-session SQL (4-layer rule: services orchestrate, repos hold SQL).
//!
//! Thin newtype over `backbone_orm::GenericCrudRepository<PosOpeningEntry, 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 (a session another unit owns reads as plain absence); the
//! connection-taking methods run on the CALLER'S open, scope-relayed transaction so a session's
//! guards and the writes they gate see the same world.

use anyhow::Result;
use sqlx::PgPool;
use uuid::Uuid;

use backbone_orm::org_scope;
// The scalar read twin lives only in the legacy `company_scope` module; what this repository needs
// from it is the connection discipline — request-dedicated connection when the composing service
// bound one, plain pool otherwise. Its legacy task-local branch never fires: this module sets no
// legacy scope of its own (ADR-0029).
use backbone_orm::company_scope::fetch_optional_scalar_scoped;

use crate::domain::entity::PosOpeningEntry;

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

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

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

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

/// The exact row an `open_session` writes.
///
/// Mirrors the raw column shape rather than the `PosOpeningEntry` entity: `opening_balances` is the
/// already-serialised per-method float array, and `status` is fixed to `'open'` by the SQL (cast
/// DB-side as `pos_session_status`). No tenancy column — the composing decorator's fill trigger
/// stamps the acting unit (ADR-0029).
pub struct NewOpeningEntryRow {
    pub id: Uuid,
    pub pos_profile_id: Uuid,
    pub branch_id: Option<Uuid>,
    pub cashier_party_id: Uuid,
    pub opened_at: chrono::NaiveDateTime,
    pub opening_balances: serde_json::Value,
}

/// An open session claimed by the old-session alert scheduler: everything the alert consumer (the
/// host wiring a mail activity onto the cashier) needs to act, plus the anchors the emitted
/// `PosStaleSessionAlerted` event carries.
pub struct StaleSessionRow {
    pub opening_entry_id: Uuid,
    pub pos_profile_id: Uuid,
    pub cashier_party_id: Uuid,
    pub opened_at: chrono::NaiveDateTime,
}

/// Hand-written cashier-session 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 PosOpeningEntryRepository {
    /// Open a cashier session.
    ///
    /// Takes the CALLER'S connection so the register/one-open-session guards and the insert commit
    /// as one unit; the caller has already relayed the ambient org scope onto it, which is what
    /// satisfies the fence's WITH CHECK under a composed host (and leaves the insert plain on a
    /// standalone deployment).
    pub async fn insert_opening_entry(
        &self,
        conn: &mut sqlx::PgConnection,
        s: &NewOpeningEntryRow,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"INSERT INTO pos.pos_opening_entries
                (id, pos_profile_id, branch_id, cashier_party_id, opened_at, opening_balances, status)
               VALUES ($1,$2,$3,$4,$5,$6,'open'::pos_session_status)"#,
        )
        .bind(s.id).bind(s.pos_profile_id).bind(s.branch_id).bind(s.cashier_party_id)
        .bind(s.opened_at).bind(sqlx::types::Json(&s.opening_balances))
        .execute(conn)
        .await?;
        Ok(())
    }

    /// Read a session's status (pool variant, for guards outside a unit of work). ID-only: rides the
    /// ambient org scope — `Ok(None)` covers "no such session" and "another unit's session" alike,
    /// and deliberately cannot tell them apart.
    pub async fn fetch_status(
        &self,
        pool: &PgPool,
        opening_entry_id: Uuid,
    ) -> Result<Option<String>, sqlx::Error> {
        fetch_optional_scalar_scoped(
            pool,
            sqlx::query_scalar(
                "SELECT status::text FROM pos.pos_opening_entries WHERE id=$1 AND (metadata->>'deleted_at') IS NULL",
            )
            .bind(opening_entry_id),
        )
        .await
    }

    /// Read a session's status on the CALLER'S scope-relayed transaction — the guard inside a unit
    /// of work (ring/movement/sync), so the read sees exactly the fence the paired write writes under.
    pub async fn fetch_status_on(
        &self,
        conn: &mut sqlx::PgConnection,
        opening_entry_id: Uuid,
    ) -> Result<Option<String>, sqlx::Error> {
        sqlx::query_scalar(
            "SELECT status::text FROM pos.pos_opening_entries WHERE id=$1 AND (metadata->>'deleted_at') IS NULL",
        )
        .bind(opening_entry_id)
        .fetch_optional(conn)
        .await
    }

    /// Read a session's opening float array (the per-method drawer starting balance). Same ID-only
    /// scope contract as [`Self::fetch_status`].
    ///
    /// The `Option` nests deliberately, and the caller distinguishes the two levels: the OUTER `None`
    /// means no such session (the caller raises `SessionNotFound`), while an inner `None` means the
    /// session exists with a NULL `opening_balances` (a drawer that opened with no float — expected
    /// starts empty, which is not an error).
    #[allow(clippy::type_complexity)]
    pub async fn fetch_opening_balances(
        &self,
        pool: &PgPool,
        opening_entry_id: Uuid,
    ) -> Result<Option<Option<sqlx::types::Json<serde_json::Value>>>, sqlx::Error> {
        fetch_optional_scalar_scoped(
            pool,
            sqlx::query_scalar(
                "SELECT opening_balances FROM pos.pos_opening_entries WHERE id=$1 AND (metadata->>'deleted_at') IS NULL",
            )
            .bind(opening_entry_id),
        )
        .await
    }

    /// Whether the register already holds an OPEN session — the service-side arm of the
    /// one-open-session rule. Runs on the CALLER'S scope-relayed transaction so the check and the
    /// open insert it guards see the same world (the composition-installed per-(unit, register)
    /// partial unique is the race backstop).
    pub async fn has_open_session(
        &self,
        conn: &mut sqlx::PgConnection,
        pos_profile_id: Uuid,
    ) -> Result<bool, sqlx::Error> {
        let found: Option<i32> = sqlx::query_scalar(
            r#"SELECT 1 FROM pos.pos_opening_entries
               WHERE pos_profile_id=$1 AND status='open'::pos_session_status
                 AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(pos_profile_id)
        .fetch_optional(conn)
        .await?;
        Ok(found.is_some())
    }

    /// Read the session's register (pool variant, for lookups outside a unit of work). ID-only:
    /// rides the ambient org scope; `Ok(None)` = no such session in the caller's scope.
    pub async fn fetch_profile_id(
        &self,
        pool: &PgPool,
        opening_entry_id: Uuid,
    ) -> Result<Option<Uuid>, sqlx::Error> {
        fetch_optional_scalar_scoped(
            pool,
            sqlx::query_scalar(
                "SELECT pos_profile_id FROM pos.pos_opening_entries WHERE id=$1 AND (metadata->>'deleted_at') IS NULL",
            )
            .bind(opening_entry_id),
        )
        .await
    }

    /// Read the session's register, on the CALLER'S connection so it reads the same snapshot the
    /// closing-entry insert writes. `Ok(None)` = no such session — the caller maps that refusal.
    /// The caller has already relayed the ambient org scope onto it — don't re-bind here.
    pub async fn fetch_profile_id_on(
        &self,
        conn: &mut sqlx::PgConnection,
        opening_entry_id: Uuid,
    ) -> Result<Option<Uuid>, sqlx::Error> {
        sqlx::query_scalar("SELECT pos_profile_id FROM pos.pos_opening_entries WHERE id=$1")
            .bind(opening_entry_id)
            .fetch_optional(conn)
            .await
    }

    /// The variance-booking accounts of a session's register: the cash account (the drawer side) and
    /// the difference/write-off account (the absorption side), plus the currency the correction
    /// posts in. Read on the CALLER'S connection so the close sees the same snapshot it writes.
    /// `Ok(None)` on either account means the register is not configured for variance booking — the
    /// caller decides whether that refuses the close (non-zero variance) or skips it (balanced).
    pub async fn fetch_variance_accounts_on(
        &self,
        conn: &mut sqlx::PgConnection,
        opening_entry_id: Uuid,
    ) -> Result<(Option<Uuid>, Option<Uuid>, String), sqlx::Error> {
        sqlx::query_as(
            r#"SELECT p.cash_account_id, p.write_off_account_id, p.currency
               FROM pos.pos_opening_entries o JOIN pos.pos_profiles p ON p.id = o.pos_profile_id
               WHERE o.id=$1"#,
        )
        .bind(opening_entry_id)
        .fetch_one(conn)
        .await
    }

    /// Flip the session to `closed`. Takes the CALLER'S connection so this and the closing-entry insert
    /// commit as one unit; the caller has already relayed the ambient scope onto it.
    pub async fn mark_closed(
        &self,
        conn: &mut sqlx::PgConnection,
        opening_entry_id: Uuid,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("UPDATE pos.pos_opening_entries SET status='closed'::pos_session_status WHERE id=$1")
            .bind(opening_entry_id)
            .execute(conn)
            .await?;
        Ok(())
    }

    /// Claim the caller's scope's stale open sessions — the intake half of the old-session alert
    /// scheduler. Runs on the CALLER'S scope-relayed transaction: the fence the caller bound is what
    /// bounds the claim to the acting unit (ADR-0029 per-unit handler).
    ///
    /// `FOR UPDATE SKIP LOCKED` is the pickup lock (ADR-0020 §4): two concurrent runs of the drainer
    /// never claim the same session. A session is stale when it is still `open`, opened before
    /// `cutoff`, and has not been alerted yet (the `stale_session_alerted_at` marker in `metadata` is
    /// the once-only latch — it is what keeps a daily scheduler from re-nagging the same session
    /// every run). `opened_at` is cast to a wall-clock `timestamp` in the session zone — the exact
    /// inverse of the cast the open insert applies — so the Rust-side `NaiveDateTime` round-trips
    /// whatever the server's zone is.
    pub async fn claim_stale_sessions(
        &self,
        conn: &mut sqlx::PgConnection,
        cutoff: chrono::NaiveDateTime,
        limit: i64,
    ) -> Result<Vec<StaleSessionRow>, sqlx::Error> {
        let rows = sqlx::query_as::<_, (Uuid, Uuid, Uuid, chrono::NaiveDateTime)>(
            r#"SELECT id, pos_profile_id, cashier_party_id, opened_at::timestamp
               FROM pos.pos_opening_entries
               WHERE status='open'::pos_session_status
                 AND opened_at < $1
                 AND (metadata->>'stale_session_alerted_at') IS NULL
                 AND (metadata->>'deleted_at') IS NULL
               ORDER BY opened_at
               LIMIT $2
               FOR UPDATE SKIP LOCKED"#,
        )
        .bind(cutoff)
        .bind(limit)
        .fetch_all(conn)
        .await?;
        Ok(rows
            .into_iter()
            .map(|(opening_entry_id, pos_profile_id, cashier_party_id, opened_at)| StaleSessionRow {
                opening_entry_id,
                pos_profile_id,
                cashier_party_id,
                opened_at,
            })
            .collect())
    }

    /// Stamp the once-only alert marker onto the claimed sessions (same transaction as the claim —
    /// `commit_policy: single_transaction` per the scheduler declaration). Takes the CALLER'S
    /// connection.
    pub async fn mark_stale_alerted(
        &self,
        conn: &mut sqlx::PgConnection,
        opening_entry_ids: &[Uuid],
        at: chrono::DateTime<chrono::Utc>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"UPDATE pos.pos_opening_entries
               SET metadata = jsonb_set(COALESCE(metadata, '{}'::jsonb), '{stale_session_alerted_at}', to_jsonb($2::timestamptz))
               WHERE id = ANY($1)"#,
        )
        .bind(opening_entry_ids)
        .bind(at)
        .execute(conn)
        .await?;
        Ok(())
    }
}

backbone_core::impl_crud_repository!(PosOpeningEntryRepository, PosOpeningEntry, soft_delete);