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