Skip to main content

backbone_orm/
org_scope.rs

1//! Org-tree request scope for the entitlement-union RLS fence (ADR-0028/0029).
2//!
3//! ADR-0028 replaces the single `company_id` scoping key with `org_unit_id`: one org tree per
4//! tenant database (root / company / branch nodes), and a session sees the UNION of the subtrees
5//! under every node it is entitled to — always including the root node, which owns tenant-wide
6//! shared rows. The database half is the policy
7//! `org_unit_id = ANY(string_to_array(current_setting('app.scope_unit_ids', true), ',')::uuid[])`;
8//! this module is the application half that resolves a session's entitled ids and carries them
9//! for the duration of a request.
10//!
11//! Session variables set by an org request scope, in full:
12//! - `app.scope_unit_ids` — the entitlement-union fence (read by org policies).
13//! - `app.company_id` — the legacy equality fence during the re-key transition, resolved from
14//!   the acting node's company ancestry.
15//! - `app.acting_unit_id` — where new records land: the column DEFAULT
16//!   `nullif(current_setting('app.acting_unit_id', true), '')::uuid` on decorated tables
17//!   (ADR-0029) resolves INSERTs that omit `org_unit_id`. Unset/empty → NULL → NOT NULL
18//!   violation: an insert outside a scope fails loud, never silently unscoped.
19//! - the six `app.*` audit variables of [`crate::audit_context`] (actor, correlation id,
20//!   request facts) when the request carries a
21//!   [`RequestAuditContext`](crate::audit_context::RequestAuditContext) — set by the audited
22//!   twin [`with_org_request_scope_and_audit`], on the same request-dedicated connection, so
23//!   the auditlog capture function's triggers read attribution off every write of the request.
24//!
25//! During the module-by-module re-key both fences are live at once: some tables still read
26//! `app.company_id` (ADR-0008 equality fence), org-re-keyed tables read `app.scope_unit_ids`.
27//! [`with_org_request_scope`] therefore sets ALL THREE session variables on one request-dedicated
28//! connection. When the last company-fenced table is re-keyed, the legacy bridge retires with
29//! the old fence.
30//!
31//! The scope binds the same `REQUEST_CONN` task-local as
32//! [`with_request_scope`](crate::company_scope::with_request_scope): every scoped execute helper
33//! in this crate — and therefore every generated repository — runs on that connection and
34//! inherits the fence variables without a single call-site change.
35//!
36//! **The task-local is not the fence.** RLS is. Unscoped statements see the variables unset and
37//! match zero rows — fail-closed, identical to the ADR-0008 contract.
38
39use crate::audit_context::{AUDIT_CONTEXT_VARS, RequestAuditContext};
40use sqlx::PgPool;
41use std::fmt;
42use std::future::Future;
43use std::sync::Arc;
44use tokio::sync::Mutex;
45use uuid::Uuid;
46
47/// The fence variables an org request scope sets — the reset inventory's first half. Private:
48/// callers reset through the scope wrappers, which clear fence AND audit variables together.
49const ORG_FENCE_VARS: [&str; 3] = ["app.scope_unit_ids", "app.company_id", "app.acting_unit_id"];
50
51tokio::task_local! {
52    /// Identity of the pool the ambient request connection was drawn from. Kept beside the scope so
53    /// a nested call can prove it is asking for the SAME database before reusing that connection —
54    /// with a database per tenant, reusing across pools would run the query on the wrong one.
55    static ORG_SCOPE_POOL: std::sync::Arc<sqlx::postgres::PgConnectOptions>;
56    /// The resolved org scope of the current request: entitled unit ids (union of subtrees,
57    /// root included) and the acting node. Unset for platform callers and non-request code.
58    static ORG_SCOPE: Arc<OrgScope>;
59    /// The connection the ambient org scope bound its variables on. A nested scope of another
60    /// kind (the legacy company scope) can make a different connection the current request
61    /// connection, and that one carries none of the org variables: reuse is safe only while the
62    /// current request connection is still this one.
63    static ORG_SCOPE_CONN: Arc<Mutex<sqlx::pool::PoolConnection<sqlx::Postgres>>>;
64}
65
66/// A resolved session scope over the org tree (ADR-0028).
67///
68/// Built by [`resolve_org_scope`]; carried by [`with_org_request_scope`].
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub struct OrgScope {
71    scope_unit_ids: Vec<Uuid>,
72    acting_unit_id: Uuid,
73    /// The company node governing the acting node (itself for a company, nearest company
74    /// ancestor for a branch). `None` when the chain has no company node — then no legacy
75    /// `app.company_id` is set and company-fenced tables fail closed for this session.
76    legacy_company_id: Option<Uuid>,
77}
78
79impl OrgScope {
80    /// A single-company scope for paths that cannot run the resolver — chiefly composition
81    /// seams minting records on a node the request already pinned (ADR-0029's decorator makes
82    /// those inserts resolve their `org_unit_id` from `app.acting_unit_id`).
83    ///
84    /// **Precondition: `unit` must be a COMPANY node** (or whatever node kind the caller's
85    /// rows anchor on, with `legacy_company_id` semantics in mind). [`resolve_org_scope`]
86    /// derives the legacy `app.company_id` by walking ancestors from the acting node; this
87    /// constructor sets it verbatim, so a BRANCH handed here would bind a legacy variable
88    /// matching no company-fenced row and fail closed on every not-yet-stripped table.
89    ///
90    /// The scope ids are exactly `[unit]` — no root-shared rows, no sibling subtrees. That is
91    /// fail-narrow (the same property as [`execute_unit_scoped`]): inserts land on the unit
92    /// and reads see only the unit's own rows. Paths needing the session's entitlement union
93    /// must resolve a real scope instead.
94    pub fn for_company_unit(unit: Uuid) -> Self {
95        Self {
96            scope_unit_ids: vec![unit],
97            acting_unit_id: unit,
98            legacy_company_id: Some(unit),
99        }
100    }
101
102    /// A scope for work done on behalf of one event or record (a workflow step, a job): it acts at
103    /// `unit` and sees only `unit`'s own rows, for `company`'s legacy fence. Unlike
104    /// [`for_company_unit`](Self::for_company_unit), the unit need not be the company node: a
105    /// branch's event is handled under the branch while the company fence stays the company's.
106    pub fn for_unit_of_company(unit: Uuid, company: Uuid) -> Self {
107        Self {
108            scope_unit_ids: vec![unit],
109            acting_unit_id: unit,
110            legacy_company_id: Some(company),
111        }
112    }
113
114    /// Every unit id this session may see rows for: the union of the entitled subtrees plus the
115    /// root node. Order is stable (sorted) so the serialized form is deterministic.
116    pub fn scope_unit_ids(&self) -> &[Uuid] {
117        &self.scope_unit_ids
118    }
119
120    /// The node the session acts at — the default `org_unit_id` for new records and the UI's
121    /// context node.
122    pub fn acting_unit_id(&self) -> Uuid {
123        self.acting_unit_id
124    }
125
126    /// The company node for the legacy `app.company_id` fence during the re-key transition.
127    pub fn legacy_company_id(&self) -> Option<Uuid> {
128        self.legacy_company_id
129    }
130
131    /// The scope ids in the exact wire form the fence policy parses: comma-joined, no spaces.
132    fn scope_unit_ids_csv(&self) -> String {
133        self.scope_unit_ids
134            .iter()
135            .map(|id| id.to_string())
136            .collect::<Vec<_>>()
137            .join(",")
138    }
139}
140
141/// Failures of [`resolve_org_scope`] — all of them mean "do NOT open a scoped session".
142#[derive(Debug)]
143pub enum OrgScopeError {
144    /// The acting node does not exist in `organization.org_units` (wrong id, or the org spine
145    /// migration has not run in this database).
146    UnknownActingUnit(Uuid),
147    /// The database lacks the spine helpers (`organization.org_unit_subtree` / `org_unit_root`).
148    MissingSpineHelpers,
149    Database(sqlx::Error),
150}
151
152impl fmt::Display for OrgScopeError {
153    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
154        match self {
155            Self::UnknownActingUnit(id) => {
156                write!(f, "org scope: acting unit {id} is not an organization.org_units node")
157            }
158            Self::MissingSpineHelpers => write!(
159                f,
160                "org scope: organization.org_unit_subtree / org_unit_root are missing — \
161                 has the org spine migration run in this database?"
162            ),
163            Self::Database(e) => write!(f, "org scope: resolution query failed: {e}"),
164        }
165    }
166}
167
168impl std::error::Error for OrgScopeError {}
169
170impl From<sqlx::Error> for OrgScopeError {
171    fn from(e: sqlx::Error) -> Self {
172        Self::Database(e)
173    }
174}
175
176/// The company node governing a unit: the unit itself when it is a company, otherwise its
177/// NEAREST company ancestor. The walk carries each step's depth and the pick orders by it, so
178/// with nested companies (a subsidiary under a parent company) a subsidiary's branch acts as
179/// the subsidiary — never as whichever company in the chain happens to have the lowest id.
180const NEAREST_COMPANY_SQL: &str = "WITH RECURSIVE up AS ( \
181        SELECT id, parent_id, kind, 0 AS depth FROM organization.org_units WHERE id = $1 \
182        UNION ALL \
183        SELECT o.id, o.parent_id, o.kind, up.depth + 1 FROM organization.org_units o \
184          JOIN up ON o.id = up.parent_id \
185     ) SELECT up.id FROM up WHERE up.kind::text = 'company' ORDER BY up.depth LIMIT 1";
186
187/// Resolve a session's scope from the org tree.
188///
189/// `acting_unit` is the node the session acts at (from the signed token / tenant registry).
190/// `additional_entitled` are further nodes the session holds entitlements for (e.g. a group
191/// admin entitled to a sister company) — each contributes its whole subtree. The resolved scope
192/// is the union of those subtrees plus the root node.
193///
194/// `org_units` is unfenced by design (it IS the scoping substrate), so this runs on any
195/// connection of the tenant database regardless of fence state.
196pub async fn resolve_org_scope(
197    conn: &mut sqlx::PgConnection,
198    acting_unit: Uuid,
199    additional_entitled: &[Uuid],
200) -> Result<OrgScope, OrgScopeError> {
201    let mut roots = vec![acting_unit];
202    roots.extend_from_slice(additional_entitled);
203
204    let scope_unit_ids: Vec<Uuid> = sqlx::query_scalar(
205        "SELECT u.id FROM organization.org_unit_subtree($1::uuid[]) AS u(id) \
206         UNION SELECT organization.org_unit_root()",
207    )
208    .bind(&roots)
209    .fetch_all(&mut *conn)
210    .await
211    .map_err(|e| {
212        // 42883 undefined_function / 42P01 undefined_table both mean the spine is absent.
213        match &e {
214            sqlx::Error::Database(db) if db.code().as_deref() == Some("42883")
215                || db.code().as_deref() == Some("42P01") =>
216            {
217                OrgScopeError::MissingSpineHelpers
218            }
219            _ => OrgScopeError::Database(e),
220        }
221    })?;
222
223    if scope_unit_ids.is_empty() {
224        return Err(OrgScopeError::MissingSpineHelpers);
225    }
226    if !scope_unit_ids.contains(&acting_unit) {
227        // subtree() silently drops unknown roots; surface that instead of opening a
228        // narrower-than-asked scope (a wrong-but-working session is worse than a loud error).
229        return Err(OrgScopeError::UnknownActingUnit(acting_unit));
230    }
231
232    // Walk up from the acting node to the governing company node for the legacy fence.
233    let legacy_company_id: Option<Uuid> = sqlx::query_scalar(NEAREST_COMPANY_SQL)
234    .bind(acting_unit)
235    .fetch_optional(&mut *conn)
236    .await?;
237
238    let mut sorted = scope_unit_ids;
239    sorted.sort_unstable();
240    sorted.dedup();
241
242    Ok(OrgScope {
243        scope_unit_ids: sorted,
244        acting_unit_id: acting_unit,
245        legacy_company_id,
246    })
247}
248
249/// Run `f` with a request-dedicated connection carrying the session's org scope.
250///
251/// Sets `app.scope_unit_ids` (the entitlement-union fence), the legacy `app.company_id`
252/// (equality fence, resolved from the acting node's company ancestry), and `app.acting_unit_id`
253/// (the acting-unit DEFAULT source for inserts on decorated tables, ADR-0029) at the session
254/// level, so org-re-keyed and not-yet-re-keyed tables are both fenced correctly for the whole
255/// request — including ID-only lookups, which ride the connection rather than the query text.
256///
257/// Mirrors [`with_request_scope`](crate::company_scope::with_request_scope)'s reset discipline:
258/// all three variables are cleared unconditionally before the connection returns to the pool,
259/// even when a `REQUEST_CONN` clone outlives the scope — a clone that runs queries after the
260/// reset does so unscoped (fail-closed), never with the previous session's scope.
261pub async fn with_org_request_scope<F, R>(pool: &PgPool, scope: OrgScope, f: F) -> Result<R, sqlx::Error>
262where
263    F: Future<Output = R>,
264{
265    with_org_request_scope_internal(pool, scope, None, f).await
266}
267
268/// [`with_org_request_scope`] plus the request's audit attribution (ADR-0025): the six
269/// [`RequestAuditContext`] variables are bound on the same request-dedicated connection, so
270/// every write of the request — including the ones that fire the auditlog capture function's
271/// triggers — reads the same actor and request facts off the connection it rides.
272///
273/// This is the wrapper a guarded route runs: the guard resolves the scope off the token, builds
274/// the audit context off the token's `sub` and the request itself, and the two channels travel
275/// together for the whole request. The reset discipline clears fence AND audit variables
276/// unconditionally: a pooled connection must never carry the previous request's attribution
277/// into the next one's audit rows.
278pub async fn with_org_request_scope_and_audit<F, R>(
279    pool: &PgPool,
280    scope: OrgScope,
281    audit: RequestAuditContext,
282    f: F,
283) -> Result<R, sqlx::Error>
284where
285    F: Future<Output = R>,
286{
287    with_org_request_scope_internal(pool, scope, Some(&audit), f).await
288}
289
290async fn with_org_request_scope_internal<F, R>(
291    pool: &PgPool,
292    scope: OrgScope,
293    audit: Option<&RequestAuditContext>,
294    f: F,
295) -> Result<R, sqlx::Error>
296where
297    F: Future<Output = R>,
298{
299    // Everything below the fast path lives in `open_org_request_scope`, behind a Box::pin. That is
300    // what keeps a nest cheap: this function's future would otherwise carry the whole slow path's
301    // locals — a pool connection, the Arc/Mutex holder, the bind statements and the task-local
302    // scope futures — and a three-level nest would stack three of those frames whether or not the
303    // slow path ever runs. Boxing moves them to the heap, so a nesting level costs a pointer.
304    // Reuse an ambient scope that already matches, instead of nesting a second one.
305    //
306    // A nested wrapper used to acquire a second connection, re-run the fence binds, and lay another
307    // prologue on the stack — roughly 600 KB per level in a debug build, so two levels plus the
308    // caller's own depth overflowed a default 2 MB worker thread and killed the task. It also held
309    // two pool connections for one logical request, which self-deadlocks a small pool.
310    //
311    // Three things must all hold before reuse is safe: a request connection is bound, the incoming
312    // scope is exactly the ambient one (a different scope needs its own binds), and the pool is the
313    // same (with a database per tenant, the ambient connection may be another tenant's entirely).
314    // A call carrying audit attribution always takes the slow path — its variables are not part of
315    // the scope comparison, so matching scopes say nothing about matching attribution.
316    if audit.is_none()
317        && crate::company_scope::current_request_conn()
318            .is_some_and(|conn| ORG_SCOPE_CONN.try_with(|own| Arc::ptr_eq(own, &conn)).unwrap_or(false))
319        && ORG_SCOPE.try_with(|ambient| **ambient == scope).unwrap_or(false)
320        && ORG_SCOPE_POOL
321            .try_with(|ambient| std::sync::Arc::ptr_eq(ambient, &pool.connect_options()))
322            .unwrap_or(false)
323    {
324        // The ambient scope owns the connection and will reset it; this call adds nothing to undo.
325        return Ok(f.await);
326    }
327
328    Box::pin(open_org_request_scope(pool, scope, audit, f)).await
329}
330
331/// The slow path: acquire a request-dedicated connection, bind the fence (and audit) variables on
332/// it, run `f` inside the task-locals, then reset every variable it may have set.
333async fn open_org_request_scope<F, R>(
334    pool: &PgPool,
335    scope: OrgScope,
336    audit: Option<&RequestAuditContext>,
337    f: F,
338) -> Result<R, sqlx::Error>
339where
340    F: Future<Output = R>,
341{
342    let conn = pool.acquire().await?;
343    let holder = Arc::new(Mutex::new(conn));
344    // Armed before the first variable is set, so a failed bind halfway through, a panic, or the
345    // whole future being dropped (a client disconnect) still clears every fence and audit
346    // variable before the connection can return to the pool. Resetting a variable this scope
347    // did not set costs one cheap set_config and keeps the inventory one list.
348    let reset = crate::company_scope::ScopeResetGuard::new(
349        holder.clone(),
350        ORG_FENCE_VARS
351            .into_iter()
352            .chain(AUDIT_CONTEXT_VARS)
353            .collect(),
354        "backbone_orm::org_scope",
355    );
356    {
357        let mut conn = holder.lock().await;
358        sqlx::query("SELECT set_config('app.scope_unit_ids', $1, false)")
359            .bind(scope.scope_unit_ids_csv())
360            .execute(&mut **conn)
361            .await?;
362        sqlx::query("SELECT set_config('app.company_id', $1, false)")
363            .bind(
364                scope
365                    .legacy_company_id
366                    .map(|id| id.to_string())
367                    .unwrap_or_default(),
368            )
369            .execute(&mut **conn)
370            .await?;
371        sqlx::query("SELECT set_config('app.acting_unit_id', $1, false)")
372            .bind(scope.acting_unit_id.to_string())
373            .execute(&mut **conn)
374            .await?;
375        if let Some(audit) = audit {
376            audit.bind_on(&mut conn, false).await?;
377        }
378    }
379
380    let scope_arc = Arc::new(scope);
381    let pool_identity = pool.connect_options();
382    // The audit attribution rides as a task-local TOO (not only as variables on
383    // the dedicated connection): write services below open their own pool
384    // transactions, whose connections the session-level binding never touches,
385    // and they relay it onto those transactions with
386    // `relay_ambient_audit_on` — the same discipline as the fence relay.
387    let audit_owned = audit.cloned();
388    let result = ORG_SCOPE_POOL
389        .scope(
390            pool_identity,
391            ORG_SCOPE_CONN.scope(holder.clone(), ORG_SCOPE.scope(scope_arc.clone(), async {
392                crate::company_scope::with_company_scope_internal(scope_arc.legacy_company_id, async {
393                    let inner = crate::company_scope::with_request_conn_internal(holder.clone(), f);
394                    match audit_owned {
395                        Some(audit) => {
396                            crate::audit_context::with_request_audit(audit, inner).await
397                        }
398                        None => inner.await,
399                    }
400                })
401                .await
402            })),
403        )
404        .await;
405
406    // The normal exit resets in line (see with_request_scope for the lingering-clone
407    // reasoning — the same contract applies to every variable); every other exit is the
408    // guard's Drop.
409    reset.reset().await;
410    drop(holder);
411    Ok(result)
412}
413
414/// The resolved org scope of the current request, if one is bound.
415///
416/// Services use the acting node as the default `org_unit_id` for new records; `None` means no
417/// org scope (platform caller or non-request code) — do not guess a node then.
418pub fn current_org_scope() -> Option<OrgScope> {
419    ORG_SCOPE.try_with(|s| s.as_ref().clone()).ok()
420}
421
422/// Bind an explicit org scope onto an already-open transaction/connection, transaction-locally.
423///
424/// The transaction twin of [`with_org_request_scope`], for hand-written write services and jobs
425/// that manage their own transaction (mirrors
426/// [`bind_company_on`](crate::company_scope::bind_company_on)). Binds all three session
427/// variables, `app.acting_unit_id` included, so an insert inside the transaction can rely on
428/// the acting-unit DEFAULT — and relays the request's audit attribution, so the transaction's
429/// changes are recorded as the caller's, not `'system'`.
430pub async fn bind_org_scope_on(
431    conn: &mut sqlx::PgConnection,
432    scope: &OrgScope,
433) -> Result<(), sqlx::Error> {
434    sqlx::query("SELECT set_config('app.scope_unit_ids', $1, true)")
435        .bind(scope.scope_unit_ids_csv())
436        .execute(&mut *conn)
437        .await?;
438    sqlx::query("SELECT set_config('app.company_id', $1, true)")
439        .bind(scope.legacy_company_id.map(|id| id.to_string()).unwrap_or_default())
440        .execute(&mut *conn)
441        .await?;
442    sqlx::query("SELECT set_config('app.acting_unit_id', $1, true)")
443        .bind(scope.acting_unit_id.to_string())
444        .execute(&mut *conn)
445        .await?;
446    // The request's actor, so the transaction's audit rows name who made the change.
447    crate::audit_context::relay_ambient_audit_on(conn).await
448}
449
450/// Statement-level org fence for a write whose row belongs to one org node — the hand-written
451/// repository path for callers outside a request scope.
452///
453/// Inside a request scope ([`with_org_request_scope`] or the legacy
454/// [`with_request_scope`](crate::company_scope::with_request_scope)) the query runs on the
455/// request connection, which already carries both fence variables. Outside one, it opens a
456/// short transaction and sets `app.scope_unit_ids` to exactly `unit`: the INSERT's
457/// `WITH CHECK` only tests the row's own `org_unit_id`, so the row's node alone satisfies it —
458/// and a read that accidentally rides this helper sees only that node (fail-narrow, never
459/// leaky). Reads that need the session's whole entitlement union must run inside a resolved
460/// request scope, not this helper.
461pub async fn execute_unit_scoped<'q>(
462    pool: &PgPool,
463    unit: Uuid,
464    query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
465) -> Result<sqlx::postgres::PgQueryResult, sqlx::Error> {
466    if let Some(conn) = crate::company_scope::current_request_conn() {
467        let mut g = conn.lock().await;
468        return query.execute(&mut **g).await;
469    }
470    let mut tx = pool.begin().await?;
471    sqlx::query("SELECT set_config('app.scope_unit_ids', $1, true)")
472        .bind(unit.to_string())
473        .execute(&mut *tx)
474        .await?;
475    crate::audit_context::relay_ambient_audit_on(&mut tx).await?;
476    let res = query.execute(&mut *tx).await?;
477    tx.commit().await?;
478    Ok(res)
479}
480
481/// Tenant-agnostic `execute` for hand-written module SQL (ADR-0029): ride the request-dedicated
482/// connection when one is bound — carrying whatever fence variables the COMPOSING service's scope
483/// set (`with_org_request_scope` / `with_request_scope`) — otherwise execute plainly on the pool.
484///
485/// Unlike [`execute_unit_scoped`] this invents no scope of its own: a module that knows nothing
486/// about tenancy must not fabricate a unit or a company. Under a composer's request scope the
487/// database fence owns isolation; with no scope bound this is a plain unfenced execute.
488pub async fn execute_scoped<'q>(
489    pool: &PgPool,
490    query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
491) -> Result<sqlx::postgres::PgQueryResult, sqlx::Error> {
492    if let Some(conn) = crate::company_scope::current_request_conn() {
493        let mut g = conn.lock().await;
494        return query.execute(&mut **g).await;
495    }
496    query.execute(pool).await
497}
498
499/// Tenant-agnostic `fetch_optional` for an untyped row query — the read twin of
500/// [`execute_scoped`], same connection discipline: request-dedicated connection when bound,
501/// plain pool otherwise, no scope invented.
502pub async fn fetch_optional_row_scoped<'q>(
503    pool: &PgPool,
504    query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
505) -> Result<Option<sqlx::postgres::PgRow>, sqlx::Error> {
506    if let Some(conn) = crate::company_scope::current_request_conn() {
507        let mut g = conn.lock().await;
508        return query.fetch_optional(&mut **g).await;
509    }
510    query.fetch_optional(pool).await
511}
512
513/// Tenant-agnostic `fetch_all` for an untyped row query — the set-returning sibling of
514/// [`fetch_optional_row_scoped`], same connection discipline: request-dedicated connection
515/// when bound, plain pool otherwise, no scope invented. Reads that must ride the request
516/// connection (a bare-pool read under a decorated deployment lands on a fresh connection
517/// with no fence variables and returns nothing) but owe no company predicate reach for
518/// this, not the company-scoped helpers.
519pub async fn fetch_all_rows_scoped<'q>(
520    pool: &PgPool,
521    query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
522) -> Result<Vec<sqlx::postgres::PgRow>, sqlx::Error> {
523    if let Some(conn) = crate::company_scope::current_request_conn() {
524        let mut g = conn.lock().await;
525        return query.fetch_all(&mut **g).await;
526    }
527    query.fetch_all(pool).await
528}
529
530/// Tenant-agnostic `fetch_one` for an untyped row query — the single-row sibling of
531/// [`fetch_optional_row_scoped`], same connection discipline: request-dedicated connection
532/// when bound, plain pool otherwise, no scope invented.
533pub async fn fetch_one_row_scoped<'q>(
534    pool: &PgPool,
535    query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
536) -> Result<sqlx::postgres::PgRow, sqlx::Error> {
537    if let Some(conn) = crate::company_scope::current_request_conn() {
538        let mut g = conn.lock().await;
539        return query.fetch_one(&mut **g).await;
540    }
541    query.fetch_one(pool).await
542}
543
544#[cfg(test)]
545mod tests {
546    //! Gated on `BACKBONE_ORM_RLS_DSN` (a superuser DSN). Self-contained: builds a minimal org
547    //! spine + an org-fenced table + an app role, then proves the resolver and the request
548    //! scope against them — the same shapes the organization/inventory migrations emit.
549    use super::{resolve_org_scope, with_org_request_scope, with_org_request_scope_and_audit};
550    use crate::audit_context::RequestAuditContext;
551    use sqlx::postgres::PgPoolOptions;
552    use sqlx::PgPool;
553    use sqlx::Row;
554    use uuid::Uuid;
555
556    fn dsn() -> Option<String> {
557        std::env::var("BACKBONE_ORM_RLS_DSN").ok()
558    }
559
560    async fn admin_pool(dsn: &str) -> PgPool {
561        PgPoolOptions::new().max_connections(4).connect(dsn).await.unwrap()
562    }
563
564    async fn app_pool(dsn: &str, role: &str) -> PgPool {
565        let after_at = dsn.rsplit('@').next().unwrap();
566        let url = format!("postgresql://{role}:orgpw@{after_at}");
567        PgPoolOptions::new().max_connections(1).connect(&url).await.unwrap()
568    }
569
570    /// Mint the per-run role name — a fixed name breaks on shared dev clusters (`DROP ROLE`
571    /// fails while the role holds grants in another database), a fresh one can never collide.
572    fn role_name() -> String {
573        format!("org_scope_app_{}", &Uuid::new_v4().simple().to_string()[..8])
574    }
575
576    async fn setup(admin: &PgPool, role: &str, root: Uuid, company: Uuid, branch: Uuid, other: Uuid) {
577        sqlx::raw_sql(&format!(
578            "DROP SCHEMA IF EXISTS organization CASCADE; \
579             DROP SCHEMA IF EXISTS org_scope_test CASCADE; \
580             CREATE SCHEMA organization; \
581             CREATE TABLE organization.org_units ( \
582                 id uuid PRIMARY KEY, kind text NOT NULL, parent_id uuid, \
583                 code text, name text NOT NULL, metadata jsonb NOT NULL DEFAULT '{{}}' \
584             ); \
585             CREATE UNIQUE INDEX one_root ON organization.org_units (kind) WHERE kind = 'root'; \
586             CREATE OR REPLACE FUNCTION organization.org_unit_subtree(p_roots uuid[]) \
587                 RETURNS SETOF uuid LANGUAGE sql STABLE AS $$ \
588                 WITH RECURSIVE tree AS ( \
589                     SELECT o.id FROM organization.org_units o WHERE o.id = ANY(p_roots) \
590                     UNION ALL \
591                     SELECT o.id FROM organization.org_units o JOIN tree t ON o.parent_id = t.id \
592                 ) SELECT id FROM tree $$; \
593             CREATE OR REPLACE FUNCTION organization.org_unit_root() RETURNS uuid \
594                 LANGUAGE sql STABLE AS $$ \
595                 SELECT id FROM organization.org_units WHERE kind = 'root' $$; \
596             CREATE SCHEMA org_scope_test; \
597             CREATE TABLE org_scope_test.t (id uuid PRIMARY KEY, org_unit_id uuid NOT NULL, code text); \
598             ALTER TABLE org_scope_test.t ENABLE ROW LEVEL SECURITY; \
599             ALTER TABLE org_scope_test.t FORCE ROW LEVEL SECURITY; \
600             CREATE POLICY t_org_isolation ON org_scope_test.t FOR ALL \
601                 USING (org_unit_id = ANY(string_to_array(current_setting('app.scope_unit_ids', true), ',')::uuid[])) \
602                 WITH CHECK (org_unit_id = ANY(string_to_array(current_setting('app.scope_unit_ids', true), ',')::uuid[])); \
603             CREATE ROLE {role} LOGIN PASSWORD 'orgpw'; \
604             GRANT USAGE ON SCHEMA organization, org_scope_test TO {role}; \
605             GRANT SELECT ON organization.org_units TO {role}; \
606             GRANT SELECT, INSERT, UPDATE, DELETE ON org_scope_test.t TO {role};",
607        ))
608        .execute(admin)
609        .await
610        .unwrap();
611
612        sqlx::query(
613            "INSERT INTO organization.org_units (id, kind, parent_id, code, name) VALUES \
614                 ($1, 'root',   NULL, 'root',  'Tenant root'), \
615                 ($2, 'company', $1,   'CO',    'Company'), \
616                 ($3, 'branch',  $2,   'BR',    'Branch'), \
617                 ($4, 'company', $1,   'OTHER', 'Other Company')",
618        )
619        .bind(root)
620        .bind(company)
621        .bind(branch)
622        .bind(other)
623        .execute(admin)
624        .await
625        .unwrap();
626    }
627
628    /// A legacy company scope nested inside an org scope makes its own connection the request
629    /// connection, and that connection carries no org variables. A second request for the same
630    /// org scope inside it must bind a connection of its own, so a write it makes reads the acting
631    /// unit, rather than reuse the legacy one and insert with no org unit.
632    #[tokio::test]
633    async fn a_nested_company_scope_never_lends_its_connection_to_an_org_scope() {
634        let Some(dsn) = dsn() else { eprintln!("skipping: set BACKBONE_ORM_RLS_DSN"); return };
635        let pool = admin_pool(&dsn).await;
636        let unit = Uuid::new_v4();
637        let scope = super::OrgScope::for_company_unit(unit);
638        let acting = with_org_request_scope(&pool, scope.clone(), async {
639            crate::company_scope::with_request_scope(&pool, unit, async {
640                with_org_request_scope(&pool, scope.clone(), async {
641                    let row = super::fetch_optional_row_scoped(
642                        &pool,
643                        sqlx::query("SELECT current_setting('app.acting_unit_id', true) AS acting"),
644                    )
645                    .await
646                    .unwrap()
647                    .unwrap();
648                    row.get::<Option<String>, _>("acting")
649                })
650                .await
651                .unwrap()
652            })
653            .await
654            .unwrap()
655        })
656        .await
657        .unwrap();
658        assert_eq!(acting.as_deref(), Some(unit.to_string().as_str()));
659    }
660
661    #[tokio::test]
662    async fn resolver_and_fence_shape() {
663        let Some(dsn) = dsn() else { eprintln!("skipping: set BACKBONE_ORM_RLS_DSN"); return };
664        let root = Uuid::new_v4();
665        let company = Uuid::new_v4();
666        let branch = Uuid::new_v4();
667        let other = Uuid::new_v4();
668        let role = role_name();
669        let admin = admin_pool(&dsn).await;
670        setup(&admin, &role, root, company, branch, other).await;
671
672        // Seed as superuser (bypasses RLS).
673        sqlx::query("INSERT INTO org_scope_test.t (id, org_unit_id, code) VALUES ($1,$2,'CO-WH'), ($3,$4,'BR-WH'), ($5,$6,'OTHER-WH')")
674            .bind(Uuid::new_v4()).bind(company)
675            .bind(Uuid::new_v4()).bind(branch)
676            .bind(Uuid::new_v4()).bind(other)
677            .execute(&admin).await.unwrap();
678
679        // Resolve from the branch node: subtree() DESCENDS only, so the scope is the branch's
680        // own subtree (itself) plus the root node — the parent company is NOT auto-included;
681        // seeing it requires entitlement. Legacy bridge = the governing company.
682        let mut conn = admin.acquire().await.unwrap();
683        let scope = resolve_org_scope(&mut conn, branch, &[]).await.unwrap();
684        let mut got = scope.scope_unit_ids().to_vec();
685        got.sort_unstable();
686        let mut want = vec![branch, root];
687        want.sort_unstable();
688        assert_eq!(got, want);
689        assert_eq!(scope.acting_unit_id(), branch);
690        assert_eq!(scope.legacy_company_id(), Some(company));
691
692        // Union entitlement: branch acting + sister company entitled.
693        let scope_union = resolve_org_scope(&mut conn, branch, &[other]).await.unwrap();
694        assert!(scope_union.scope_unit_ids().contains(&other));
695
696        // Unknown acting node is a loud error, not a narrower scope.
697        let ghost = resolve_org_scope(&mut conn, Uuid::new_v4(), &[]).await;
698        assert!(ghost.is_err());
699
700        // The request scope fences queries on the app role's pool. The scoped helper (not a raw
701        // pool fetch) is the interesting path: it must route onto the request connection the
702        // scope bound, which carries the fence variables.
703        let pool = app_pool(&dsn, &role).await;
704        let (codes, twin_codes): (Vec<String>, Vec<String>) =
705            with_org_request_scope(&pool, scope, async {
706                let codes: Vec<String> = crate::company_scope::fetch_all_scoped(
707                    &pool,
708                    sqlx::query_as::<_, (String,)>("SELECT code FROM org_scope_test.t ORDER BY code"),
709                )
710                .await
711                .unwrap()
712                .into_iter()
713                .map(|r| r.0)
714                .collect();
715                // The tenant-agnostic fetch-all twin rides the same request
716                // connection and therefore the same fence — the honest helper
717                // for reads that owe no company predicate.
718                let twin_codes: Vec<String> = super::fetch_all_rows_scoped(
719                    &pool,
720                    sqlx::query("SELECT code FROM org_scope_test.t ORDER BY code"),
721                )
722                .await
723                .unwrap()
724                .into_iter()
725                .map(|r| r.get::<String, &str>("code"))
726                .collect();
727                (codes, twin_codes)
728            })
729            .await
730            .unwrap();
731        assert_eq!(codes, ["BR-WH"], "branch session sees only its subtree (+ shared root rows), not the company or sister-company rows");
732        assert_eq!(
733            twin_codes, codes,
734            "the org-scope fetch-all twin fences identically to the scoped helper"
735        );
736
737        // A bare pool read (no request connection, no fence variables on the
738        // freshly acquired connection) sees nothing under the decorated fence —
739        // the failure mode that makes riding the request connection mandatory.
740        let bare: Vec<String> = sqlx::query("SELECT code FROM org_scope_test.t")
741            .fetch_all(&pool)
742            .await
743            .unwrap()
744            .into_iter()
745            .map(|r| r.get::<String, usize>(0))
746            .collect();
747        assert!(
748            bare.is_empty(),
749            "a bare pool read must not see fenced rows"
750        );
751
752        // After the scope, the pooled connection is clean (fail-closed for the next acquire).
753        let mut after = pool.acquire().await.unwrap();
754        let setting: String =
755            sqlx::query_scalar("SELECT current_setting('app.scope_unit_ids', true)")
756                .fetch_one(&mut *after)
757                .await
758                .unwrap();
759        assert_eq!(setting, "", "scope leaked onto the pooled connection");
760    }
761
762    /// The audit-context channel of the request scope (ADR-0025): the audited twin binds the
763    /// six `app.*` attribution variables on the request-dedicated connection, a row-level
764    /// trigger on a write through the scoped helpers reads them off that connection, and every
765    /// variable — fence AND audit — is cleared before the connection returns to the pool. The
766    /// unaudited twin leaves the audit variables unset (the capture function's `'system'`
767    /// fallback reads exactly that: empty).
768    ///
769    /// Runs in its own disposable database (dropped at the end), because the spine-building
770    /// sibling test above recreates the `organization` schema and would race this one's if they
771    /// shared a database — cargo runs test fns concurrently.
772    #[tokio::test]
773    async fn audit_context_binds_rides_and_clears_with_the_scope() {
774        let Some(maintenance_dsn) = dsn() else {
775            eprintln!("skipping: set BACKBONE_ORM_RLS_DSN");
776            return;
777        };
778        const PROOF_DB: &str = "backbone_orm_audit_ctx_probe";
779        let proof_dsn = {
780            let (before_db, ..) = maintenance_dsn.rsplit_once('/').unwrap();
781            format!("{before_db}/{PROOF_DB}")
782        };
783        let maintenance = admin_pool(&maintenance_dsn).await;
784        // Each statement rides its own simple-protocol query — CREATE/DROP DATABASE may not run
785        // inside a transaction block.
786        sqlx::raw_sql(&format!("DROP DATABASE IF EXISTS {PROOF_DB} WITH (FORCE)"))
787            .execute(&maintenance)
788            .await
789            .unwrap();
790        sqlx::raw_sql(&format!("CREATE DATABASE {PROOF_DB}"))
791            .execute(&maintenance)
792            .await
793            .unwrap();
794        let admin = admin_pool(&proof_dsn).await;
795
796        let root = Uuid::new_v4();
797        let company = Uuid::new_v4();
798        let role = role_name();
799        // The spine + fenced table of the sibling test, plus the audit probe: a table and an
800        // AFTER INSERT trigger capturing two of the attribution variables — the minimal shape
801        // of the auditlog module's capture function, whose contract this pins without the
802        // framework crate depending on a domain module.
803        sqlx::raw_sql(&format!(
804            "CREATE SCHEMA organization; \
805             CREATE TABLE organization.org_units ( \
806                 id uuid PRIMARY KEY, kind text NOT NULL, parent_id uuid, \
807                 code text, name text NOT NULL, metadata jsonb NOT NULL DEFAULT '{{}}' \
808             ); \
809             CREATE OR REPLACE FUNCTION organization.org_unit_subtree(p_roots uuid[]) \
810                 RETURNS SETOF uuid LANGUAGE sql STABLE AS $$ \
811                 WITH RECURSIVE tree AS ( \
812                     SELECT o.id FROM organization.org_units o WHERE o.id = ANY(p_roots) \
813                     UNION ALL \
814                     SELECT o.id FROM organization.org_units o JOIN tree t ON o.parent_id = t.id \
815                 ) SELECT id FROM tree $$; \
816             CREATE OR REPLACE FUNCTION organization.org_unit_root() RETURNS uuid \
817                 LANGUAGE sql STABLE AS $$ \
818                 SELECT id FROM organization.org_units WHERE kind = 'root' $$; \
819             CREATE SCHEMA org_scope_test; \
820             CREATE TABLE org_scope_test.t (id uuid PRIMARY KEY, org_unit_id uuid NOT NULL \
821                 DEFAULT nullif(current_setting('app.acting_unit_id', true), '')::uuid, code text); \
822             ALTER TABLE org_scope_test.t ENABLE ROW LEVEL SECURITY; \
823             ALTER TABLE org_scope_test.t FORCE ROW LEVEL SECURITY; \
824             CREATE POLICY t_org_isolation ON org_scope_test.t FOR ALL \
825                 USING (org_unit_id = ANY(string_to_array(current_setting('app.scope_unit_ids', true), ',')::uuid[])) \
826                 WITH CHECK (org_unit_id = ANY(string_to_array(current_setting('app.scope_unit_ids', true), ',')::uuid[])); \
827             CREATE TABLE org_scope_test.audit_probe ( \
828                 id bigserial PRIMARY KEY, actor text NOT NULL, correlation_id text NOT NULL \
829             ); \
830             CREATE FUNCTION org_scope_test.capture_probe() RETURNS trigger LANGUAGE plpgsql AS $$ \
831                 BEGIN \
832                     INSERT INTO org_scope_test.audit_probe (actor, correlation_id) \
833                     VALUES (current_setting('app.actor', true), \
834                             current_setting('app.correlation_id', true)); \
835                     RETURN NULL; \
836                 END $$; \
837             CREATE TRIGGER t_audit_probe AFTER INSERT ON org_scope_test.t \
838                 FOR EACH ROW EXECUTE FUNCTION org_scope_test.capture_probe(); \
839             CREATE ROLE {role} LOGIN PASSWORD 'orgpw'; \
840             GRANT USAGE ON SCHEMA organization, org_scope_test TO {role}; \
841             GRANT SELECT ON organization.org_units TO {role}; \
842             GRANT SELECT, INSERT ON org_scope_test.t TO {role}; \
843             GRANT INSERT ON org_scope_test.audit_probe TO {role}; \
844             GRANT USAGE, SELECT ON SEQUENCE org_scope_test.audit_probe_id_seq TO {role};",
845        ))
846        .execute(&admin)
847        .await
848        .unwrap();
849        sqlx::query(
850            "INSERT INTO organization.org_units (id, kind, parent_id, code, name) VALUES \
851                 ($1, 'root',    NULL, 'ROOT', 'Tenant root'), \
852                 ($2, 'company', $1,   'CO',   'Company')",
853        )
854        .bind(root)
855        .bind(company)
856        .execute(&admin)
857        .await
858        .unwrap();
859
860        let pool = app_pool(&proof_dsn, &role).await;
861        let scope = {
862            let mut conn = admin.acquire().await.unwrap();
863            resolve_org_scope(&mut conn, company, &[]).await.unwrap()
864        };
865
866        // 1. The audited twin: every attribution variable is set on the connection writes ride
867        //    (inventory proof, same style as the fence-variable inventory above), and a trigger
868        //    on a real insert reads them — the seam the auditlog capture function depends on.
869        let audit = RequestAuditContext {
870            actor: "user-77".to_string(),
871            correlation_id: "corr-77".to_string(),
872            client_ip: "203.0.113.7".to_string(),
873            user_agent: "probe-agent/1.0".to_string(),
874            http_method: "POST".to_string(),
875            resource_path: "/api/v1/probe/widgets".to_string(),
876        };
877        let inserted = Uuid::new_v4();
878        let settings: Vec<(String, String)> = with_org_request_scope_and_audit(
879            &pool,
880            scope.clone(),
881            audit.clone(),
882            async {
883                crate::company_scope::execute_scoped(
884                    &pool,
885                    sqlx::query("INSERT INTO org_scope_test.t (id, code) VALUES ($1, 'AUDITED')")
886                        .bind(inserted),
887                )
888                .await
889                .unwrap();
890                crate::company_scope::fetch_all_scoped(
891                    &pool,
892                    sqlx::query_as::<_, (String, String)>(
893                        "SELECT s.name, current_setting(s.name, true) FROM unnest(ARRAY[\
894                         'app.actor','app.correlation_id','app.client_ip','app.user_agent',\
895                         'app.http_method','app.resource_path']) AS s(name)",
896                    ),
897                )
898                .await
899                .unwrap()
900            },
901        )
902        .await
903        .unwrap();
904        for (var, want) in audit.pairs() {
905            let got = settings
906                .iter()
907                .find(|(n, _)| n == var)
908                .map(|(_, v)| v.as_str())
909                .unwrap_or_else(|| panic!("{var} missing from the inventory query"));
910            assert_eq!(got, want, "{var} must ride the request connection");
911        }
912        // The fence still works under the audited twin: the acting-unit DEFAULT filled the row.
913        let landed: Uuid = sqlx::query_scalar("SELECT org_unit_id FROM org_scope_test.t WHERE id = $1")
914            .bind(inserted)
915            .fetch_one(&admin)
916            .await
917            .unwrap();
918        assert_eq!(landed, company, "audit channel must not disturb the fence");
919        let probe: (String, String) =
920            sqlx::query_as("SELECT actor, correlation_id FROM org_scope_test.audit_probe ORDER BY id DESC LIMIT 1")
921                .fetch_one(&admin)
922                .await
923                .unwrap();
924        assert_eq!(probe, ("user-77".to_string(), "corr-77".to_string()),
925            "a row-level trigger on the insert must read the bound attribution");
926
927        // 2. Pool hygiene: after the scope, no audit variable survives on the pooled connection.
928        let mut after = pool.acquire().await.unwrap();
929        for var in crate::audit_context::AUDIT_CONTEXT_VARS {
930            let v: String = sqlx::query_scalar("SELECT current_setting($1, true)")
931                .bind(var)
932                .fetch_one(&mut *after)
933                .await
934                .unwrap();
935            assert_eq!(v, "", "{var} leaked onto the pooled connection");
936        }
937        drop(after);
938
939        // 3. The unaudited twin leaves the channel unset — the capture function's `'system'`
940        //    fallback and NULL columns read exactly this empty wire form.
941        let plain = Uuid::new_v4();
942        with_org_request_scope(&pool, scope, async {
943            crate::company_scope::execute_scoped(
944                &pool,
945                sqlx::query("INSERT INTO org_scope_test.t (id, code) VALUES ($1, 'PLAIN')")
946                    .bind(plain),
947            )
948            .await
949            .unwrap();
950        })
951        .await
952        .unwrap();
953        let probe: (String, String) =
954            sqlx::query_as("SELECT actor, correlation_id FROM org_scope_test.audit_probe ORDER BY id DESC LIMIT 1")
955                .fetch_one(&admin)
956                .await
957                .unwrap();
958        assert_eq!(probe, (String::new(), String::new()),
959            "without an audit context the channel reads empty, never the previous request's");
960
961        // Zero residue: schemas + role in the proof database, then the database itself.
962        sqlx::raw_sql(&format!(
963            "DROP SCHEMA organization CASCADE; DROP SCHEMA org_scope_test CASCADE; DROP ROLE {role};"
964        ))
965        .execute(&admin)
966        .await
967        .unwrap();
968        pool.close().await;
969        admin.close().await;
970        sqlx::raw_sql(&format!("DROP DATABASE IF EXISTS {PROOF_DB} WITH (FORCE);"))
971            .execute(&maintenance)
972            .await
973            .unwrap();
974        maintenance.close().await;
975    }
976}