backbone_pos/infrastructure/persistence/
pos_opening_entry_repository.rs

1//! Repository for PosOpeningEntry entities
2//!
3//! Originally generated by metaphor-schema; now **user-owned** — this exact path is declared under
4//! `user_owned` in `metaphor.codegen.yaml`, so the generator skips it wholesale. The custom methods
5//! below hold the hand-written cashier-session SQL (4-layer rule: services orchestrate, repos hold SQL).
6//!
7//! Thin newtype over `backbone_orm::GenericCrudRepository<PosOpeningEntry, backbone_orm::SoftDelete>`.
8//! All standard CRUD methods are available via `Deref`.
9//!
10//! Tenancy is composition-installed (ADR-0029): pool reads ride the ambient org scope's
11//! request-dedicated connection (a session another unit owns reads as plain absence); the
12//! connection-taking methods run on the CALLER'S open, scope-relayed transaction so a session's
13//! guards and the writes they gate see the same world.
14
15use anyhow::Result;
16use sqlx::PgPool;
17use uuid::Uuid;
18
19use backbone_orm::org_scope;
20// The scalar read twin lives only in the legacy `company_scope` module; what this repository needs
21// from it is the connection discipline — request-dedicated connection when the composing service
22// bound one, plain pool otherwise. Its legacy task-local branch never fires: this module sets no
23// legacy scope of its own (ADR-0029).
24use backbone_orm::company_scope::fetch_optional_scalar_scoped;
25
26use crate::domain::entity::PosOpeningEntry;
27
28/// Table name for PosOpeningEntry entities
29pub const TABLE_NAME: &str = "pos.pos_opening_entries";
30
31/// Repository for PosOpeningEntry entities.
32///
33/// All standard CRUD, soft-delete, pagination, and bulk methods are
34/// provided automatically via `Deref` to `backbone_orm::GenericCrudRepository`.
35pub struct PosOpeningEntryRepository(
36    backbone_orm::GenericCrudRepository<PosOpeningEntry, backbone_orm::SoftDelete>,
37);
38
39impl std::ops::Deref for PosOpeningEntryRepository {
40    type Target = backbone_orm::GenericCrudRepository<PosOpeningEntry, backbone_orm::SoftDelete>;
41    fn deref(&self) -> &Self::Target { &self.0 }
42}
43
44impl PosOpeningEntryRepository {
45    /// Create a new repository instance.
46    pub fn new(pool: PgPool) -> Self {
47        Self(backbone_orm::GenericCrudRepository::new(pool, TABLE_NAME))
48    }
49}
50
51/// The exact row an `open_session` writes.
52///
53/// Mirrors the raw column shape rather than the `PosOpeningEntry` entity: `opening_balances` is the
54/// already-serialised per-method float array, and `status` is fixed to `'open'` by the SQL (cast
55/// DB-side as `pos_session_status`). No tenancy column — the composing decorator's fill trigger
56/// stamps the acting unit (ADR-0029).
57pub struct NewOpeningEntryRow {
58    pub id: Uuid,
59    pub pos_profile_id: Uuid,
60    pub branch_id: Option<Uuid>,
61    pub cashier_party_id: Uuid,
62    pub opened_at: chrono::NaiveDateTime,
63    pub opening_balances: serde_json::Value,
64}
65
66/// An open session claimed by the old-session alert scheduler: everything the alert consumer (the
67/// host wiring a mail activity onto the cashier) needs to act, plus the anchors the emitted
68/// `PosStaleSessionAlerted` event carries.
69pub struct StaleSessionRow {
70    pub opening_entry_id: Uuid,
71    pub pos_profile_id: Uuid,
72    pub cashier_party_id: Uuid,
73    pub opened_at: chrono::NaiveDateTime,
74}
75
76/// Hand-written cashier-session SQL. Lives here (not in the write service) per the module's 4-layer
77/// rule: services orchestrate and own the unit of work, repositories hold the SQL.
78impl PosOpeningEntryRepository {
79    /// Open a cashier session.
80    ///
81    /// Takes the CALLER'S connection so the register/one-open-session guards and the insert commit
82    /// as one unit; the caller has already relayed the ambient org scope onto it, which is what
83    /// satisfies the fence's WITH CHECK under a composed host (and leaves the insert plain on a
84    /// standalone deployment).
85    pub async fn insert_opening_entry(
86        &self,
87        conn: &mut sqlx::PgConnection,
88        s: &NewOpeningEntryRow,
89    ) -> Result<(), sqlx::Error> {
90        sqlx::query(
91            r#"INSERT INTO pos.pos_opening_entries
92                (id, pos_profile_id, branch_id, cashier_party_id, opened_at, opening_balances, status)
93               VALUES ($1,$2,$3,$4,$5,$6,'open'::pos_session_status)"#,
94        )
95        .bind(s.id).bind(s.pos_profile_id).bind(s.branch_id).bind(s.cashier_party_id)
96        .bind(s.opened_at).bind(sqlx::types::Json(&s.opening_balances))
97        .execute(conn)
98        .await?;
99        Ok(())
100    }
101
102    /// Read a session's status (pool variant, for guards outside a unit of work). ID-only: rides the
103    /// ambient org scope — `Ok(None)` covers "no such session" and "another unit's session" alike,
104    /// and deliberately cannot tell them apart.
105    pub async fn fetch_status(
106        &self,
107        pool: &PgPool,
108        opening_entry_id: Uuid,
109    ) -> Result<Option<String>, sqlx::Error> {
110        fetch_optional_scalar_scoped(
111            pool,
112            sqlx::query_scalar(
113                "SELECT status::text FROM pos.pos_opening_entries WHERE id=$1 AND (metadata->>'deleted_at') IS NULL",
114            )
115            .bind(opening_entry_id),
116        )
117        .await
118    }
119
120    /// Read a session's status on the CALLER'S scope-relayed transaction — the guard inside a unit
121    /// of work (ring/movement/sync), so the read sees exactly the fence the paired write writes under.
122    pub async fn fetch_status_on(
123        &self,
124        conn: &mut sqlx::PgConnection,
125        opening_entry_id: Uuid,
126    ) -> Result<Option<String>, sqlx::Error> {
127        sqlx::query_scalar(
128            "SELECT status::text FROM pos.pos_opening_entries WHERE id=$1 AND (metadata->>'deleted_at') IS NULL",
129        )
130        .bind(opening_entry_id)
131        .fetch_optional(conn)
132        .await
133    }
134
135    /// Read a session's opening float array (the per-method drawer starting balance). Same ID-only
136    /// scope contract as [`Self::fetch_status`].
137    ///
138    /// The `Option` nests deliberately, and the caller distinguishes the two levels: the OUTER `None`
139    /// means no such session (the caller raises `SessionNotFound`), while an inner `None` means the
140    /// session exists with a NULL `opening_balances` (a drawer that opened with no float — expected
141    /// starts empty, which is not an error).
142    #[allow(clippy::type_complexity)]
143    pub async fn fetch_opening_balances(
144        &self,
145        pool: &PgPool,
146        opening_entry_id: Uuid,
147    ) -> Result<Option<Option<sqlx::types::Json<serde_json::Value>>>, sqlx::Error> {
148        fetch_optional_scalar_scoped(
149            pool,
150            sqlx::query_scalar(
151                "SELECT opening_balances FROM pos.pos_opening_entries WHERE id=$1 AND (metadata->>'deleted_at') IS NULL",
152            )
153            .bind(opening_entry_id),
154        )
155        .await
156    }
157
158    /// Whether the register already holds an OPEN session — the service-side arm of the
159    /// one-open-session rule. Runs on the CALLER'S scope-relayed transaction so the check and the
160    /// open insert it guards see the same world (the composition-installed per-(unit, register)
161    /// partial unique is the race backstop).
162    pub async fn has_open_session(
163        &self,
164        conn: &mut sqlx::PgConnection,
165        pos_profile_id: Uuid,
166    ) -> Result<bool, sqlx::Error> {
167        let found: Option<i32> = sqlx::query_scalar(
168            r#"SELECT 1 FROM pos.pos_opening_entries
169               WHERE pos_profile_id=$1 AND status='open'::pos_session_status
170                 AND (metadata->>'deleted_at') IS NULL"#,
171        )
172        .bind(pos_profile_id)
173        .fetch_optional(conn)
174        .await?;
175        Ok(found.is_some())
176    }
177
178    /// Read the session's register (pool variant, for lookups outside a unit of work). ID-only:
179    /// rides the ambient org scope; `Ok(None)` = no such session in the caller's scope.
180    pub async fn fetch_profile_id(
181        &self,
182        pool: &PgPool,
183        opening_entry_id: Uuid,
184    ) -> Result<Option<Uuid>, sqlx::Error> {
185        fetch_optional_scalar_scoped(
186            pool,
187            sqlx::query_scalar(
188                "SELECT pos_profile_id FROM pos.pos_opening_entries WHERE id=$1 AND (metadata->>'deleted_at') IS NULL",
189            )
190            .bind(opening_entry_id),
191        )
192        .await
193    }
194
195    /// Read the session's register, on the CALLER'S connection so it reads the same snapshot the
196    /// closing-entry insert writes. `Ok(None)` = no such session — the caller maps that refusal.
197    /// The caller has already relayed the ambient org scope onto it — don't re-bind here.
198    pub async fn fetch_profile_id_on(
199        &self,
200        conn: &mut sqlx::PgConnection,
201        opening_entry_id: Uuid,
202    ) -> Result<Option<Uuid>, sqlx::Error> {
203        sqlx::query_scalar("SELECT pos_profile_id FROM pos.pos_opening_entries WHERE id=$1")
204            .bind(opening_entry_id)
205            .fetch_optional(conn)
206            .await
207    }
208
209    /// The variance-booking accounts of a session's register: the cash account (the drawer side) and
210    /// the difference/write-off account (the absorption side), plus the currency the correction
211    /// posts in. Read on the CALLER'S connection so the close sees the same snapshot it writes.
212    /// `Ok(None)` on either account means the register is not configured for variance booking — the
213    /// caller decides whether that refuses the close (non-zero variance) or skips it (balanced).
214    pub async fn fetch_variance_accounts_on(
215        &self,
216        conn: &mut sqlx::PgConnection,
217        opening_entry_id: Uuid,
218    ) -> Result<(Option<Uuid>, Option<Uuid>, String), sqlx::Error> {
219        sqlx::query_as(
220            r#"SELECT p.cash_account_id, p.write_off_account_id, p.currency
221               FROM pos.pos_opening_entries o JOIN pos.pos_profiles p ON p.id = o.pos_profile_id
222               WHERE o.id=$1"#,
223        )
224        .bind(opening_entry_id)
225        .fetch_one(conn)
226        .await
227    }
228
229    /// Flip the session to `closed`. Takes the CALLER'S connection so this and the closing-entry insert
230    /// commit as one unit; the caller has already relayed the ambient scope onto it.
231    pub async fn mark_closed(
232        &self,
233        conn: &mut sqlx::PgConnection,
234        opening_entry_id: Uuid,
235    ) -> Result<(), sqlx::Error> {
236        sqlx::query("UPDATE pos.pos_opening_entries SET status='closed'::pos_session_status WHERE id=$1")
237            .bind(opening_entry_id)
238            .execute(conn)
239            .await?;
240        Ok(())
241    }
242
243    /// Claim the caller's scope's stale open sessions — the intake half of the old-session alert
244    /// scheduler. Runs on the CALLER'S scope-relayed transaction: the fence the caller bound is what
245    /// bounds the claim to the acting unit (ADR-0029 per-unit handler).
246    ///
247    /// `FOR UPDATE SKIP LOCKED` is the pickup lock (ADR-0020 §4): two concurrent runs of the drainer
248    /// never claim the same session. A session is stale when it is still `open`, opened before
249    /// `cutoff`, and has not been alerted yet (the `stale_session_alerted_at` marker in `metadata` is
250    /// the once-only latch — it is what keeps a daily scheduler from re-nagging the same session
251    /// every run). `opened_at` is cast to a wall-clock `timestamp` in the session zone — the exact
252    /// inverse of the cast the open insert applies — so the Rust-side `NaiveDateTime` round-trips
253    /// whatever the server's zone is.
254    pub async fn claim_stale_sessions(
255        &self,
256        conn: &mut sqlx::PgConnection,
257        cutoff: chrono::NaiveDateTime,
258        limit: i64,
259    ) -> Result<Vec<StaleSessionRow>, sqlx::Error> {
260        let rows = sqlx::query_as::<_, (Uuid, Uuid, Uuid, chrono::NaiveDateTime)>(
261            r#"SELECT id, pos_profile_id, cashier_party_id, opened_at::timestamp
262               FROM pos.pos_opening_entries
263               WHERE status='open'::pos_session_status
264                 AND opened_at < $1
265                 AND (metadata->>'stale_session_alerted_at') IS NULL
266                 AND (metadata->>'deleted_at') IS NULL
267               ORDER BY opened_at
268               LIMIT $2
269               FOR UPDATE SKIP LOCKED"#,
270        )
271        .bind(cutoff)
272        .bind(limit)
273        .fetch_all(conn)
274        .await?;
275        Ok(rows
276            .into_iter()
277            .map(|(opening_entry_id, pos_profile_id, cashier_party_id, opened_at)| StaleSessionRow {
278                opening_entry_id,
279                pos_profile_id,
280                cashier_party_id,
281                opened_at,
282            })
283            .collect())
284    }
285
286    /// Stamp the once-only alert marker onto the claimed sessions (same transaction as the claim —
287    /// `commit_policy: single_transaction` per the scheduler declaration). Takes the CALLER'S
288    /// connection.
289    pub async fn mark_stale_alerted(
290        &self,
291        conn: &mut sqlx::PgConnection,
292        opening_entry_ids: &[Uuid],
293        at: chrono::DateTime<chrono::Utc>,
294    ) -> Result<(), sqlx::Error> {
295        sqlx::query(
296            r#"UPDATE pos.pos_opening_entries
297               SET metadata = jsonb_set(COALESCE(metadata, '{}'::jsonb), '{stale_session_alerted_at}', to_jsonb($2::timestamptz))
298               WHERE id = ANY($1)"#,
299        )
300        .bind(opening_entry_ids)
301        .bind(at)
302        .execute(conn)
303        .await?;
304        Ok(())
305    }
306}
307
308backbone_core::impl_crud_repository!(PosOpeningEntryRepository, PosOpeningEntry, soft_delete);
backbone_pos/infrastructure/persistence/pos_opening_entry_repository.rs

backbone_pos/infrastructure/persistence/
pos_opening_entry_repository.rs
