1pub mod channel;
30pub mod owner;
32pub mod tenant;
34
35#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub struct ScopeRegistryCounts {
38 pub tenant: usize,
40 pub owner: usize,
42}
43
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum ScopeRegistryState {
47 Uninitialized,
50 Initialized,
52 PolicyOnly,
55}
56
57#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59pub struct ScopeModeConflict {
60 pub current: ScopeRegistryState,
62 pub requested: ScopeRegistryState,
64}
65
66impl std::fmt::Display for ScopeModeConflict {
67 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
68 write!(
69 f,
70 "RLS isolation mode is sealed as {:?}; refusing transition to {:?}",
71 self.current, self.requested
72 )
73 }
74}
75
76impl std::error::Error for ScopeModeConflict {}
77
78#[derive(Debug)]
95pub(crate) struct ScopeModeCoordinator {
96 state: std::sync::atomic::AtomicU8,
97 policy_only_reason: std::sync::OnceLock<&'static str>,
98}
99
100const MODE_UNINITIALIZED: u8 = 0;
101const MODE_INITIALIZED: u8 = 1;
102const MODE_POLICY_ONLY: u8 = 2;
103
104impl ScopeModeCoordinator {
105 pub(crate) const fn new() -> Self {
107 Self {
108 state: std::sync::atomic::AtomicU8::new(MODE_UNINITIALIZED),
109 policy_only_reason: std::sync::OnceLock::new(),
110 }
111 }
112
113 fn decode(raw: u8) -> ScopeRegistryState {
114 match raw {
115 MODE_INITIALIZED => ScopeRegistryState::Initialized,
116 MODE_POLICY_ONLY => ScopeRegistryState::PolicyOnly,
117 _ => ScopeRegistryState::Uninitialized,
118 }
119 }
120
121 pub(crate) fn state(&self) -> ScopeRegistryState {
123 Self::decode(self.state.load(std::sync::atomic::Ordering::Acquire))
124 }
125
126 fn seal(&self, target: u8) -> Result<(), ScopeModeConflict> {
128 match self.state.compare_exchange(
129 MODE_UNINITIALIZED,
130 target,
131 std::sync::atomic::Ordering::AcqRel,
132 std::sync::atomic::Ordering::Acquire,
133 ) {
134 Ok(_) => Ok(()),
135 Err(current) if current == target => Ok(()),
136 Err(current) => Err(ScopeModeConflict {
137 current: Self::decode(current),
138 requested: Self::decode(target),
139 }),
140 }
141 }
142
143 pub(crate) fn declare_initialized(&self) -> Result<(), ScopeModeConflict> {
148 self.seal(MODE_INITIALIZED)
149 }
150
151 pub(crate) fn declare_policy_only(
154 &self,
155 reason: &'static str,
156 ) -> Result<(), ScopeModeConflict> {
157 self.seal(MODE_POLICY_ONLY)?;
158 self.policy_only_reason.get_or_init(|| reason);
159 Ok(())
160 }
161
162 pub(crate) fn policy_only_reason(&self) -> Option<&'static str> {
164 if self.state() == ScopeRegistryState::PolicyOnly {
165 self.policy_only_reason.get().copied()
166 } else {
167 None
168 }
169 }
170}
171
172impl Default for ScopeModeCoordinator {
173 fn default() -> Self {
174 Self::new()
175 }
176}
177
178static SCOPE_MODE: ScopeModeCoordinator = ScopeModeCoordinator::new();
181
182pub fn scope_registry_state() -> ScopeRegistryState {
184 SCOPE_MODE.state()
185}
186
187#[derive(Debug, Clone, PartialEq, Eq)]
189pub enum ScopeInitError {
190 ModeConflict(ScopeModeConflict),
192 NoScopedTables,
197 RegistryUnavailable(String),
200}
201
202impl std::fmt::Display for ScopeInitError {
203 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
204 match self {
205 Self::ModeConflict(conflict) => write!(f, "{conflict}"),
206 Self::NoScopedTables => write!(
207 f,
208 "refusing to seal RLS scope registries: no tenant- or owner-scoped tables were registered (declare_no_scoped_tables(reason) if that is intentional)"
209 ),
210 Self::RegistryUnavailable(why) => {
211 write!(f, "RLS scope registry unavailable: {why}")
212 }
213 }
214 }
215}
216
217impl std::error::Error for ScopeInitError {}
218
219impl From<ScopeModeConflict> for ScopeInitError {
220 fn from(conflict: ScopeModeConflict) -> Self {
221 Self::ModeConflict(conflict)
222 }
223}
224
225fn seal_initialized_if_populated(
231 counts: ScopeRegistryCounts,
232) -> Result<ScopeRegistryCounts, ScopeInitError> {
233 let live = live_registered_total()?;
234 seal_initialized_if_populated_with(live, counts)
235}
236
237fn live_registered_total() -> Result<usize, ScopeInitError> {
240 let tenant = tenant::try_tenant_table_count().map_err(ScopeInitError::RegistryUnavailable)?;
241 let owner = owner::try_owner_table_count().map_err(ScopeInitError::RegistryUnavailable)?;
242 Ok(tenant + owner)
243}
244
245fn seal_initialized_if_populated_with(
248 live_registered: usize,
249 counts: ScopeRegistryCounts,
250) -> Result<ScopeRegistryCounts, ScopeInitError> {
251 if live_registered == 0 {
252 return Err(ScopeInitError::NoScopedTables);
253 }
254 SCOPE_MODE.declare_initialized()?;
255 Ok(counts)
256}
257
258pub fn init_scope_registries(
275 schema: &crate::migrate::Schema,
276) -> Result<ScopeRegistryCounts, ScopeInitError> {
277 if SCOPE_MODE.state() == ScopeRegistryState::PolicyOnly {
278 return Err(ScopeModeConflict {
279 current: ScopeRegistryState::PolicyOnly,
280 requested: ScopeRegistryState::Initialized,
281 }
282 .into());
283 }
284 let tenant = tenant::register_from_migrate_schema(schema)
285 .map_err(ScopeInitError::RegistryUnavailable)?;
286 let owner =
287 owner::register_from_migrate_schema(schema).map_err(ScopeInitError::RegistryUnavailable)?;
288 seal_initialized_if_populated(ScopeRegistryCounts { tenant, owner })
289}
290
291pub fn init_scope_registries_from_tables(
295 tenant_tables: &[(&str, &str)],
296 owner_tables: &[(&str, &str)],
297) -> Result<ScopeRegistryCounts, ScopeInitError> {
298 if SCOPE_MODE.state() == ScopeRegistryState::PolicyOnly {
299 return Err(ScopeModeConflict {
300 current: ScopeRegistryState::PolicyOnly,
301 requested: ScopeRegistryState::Initialized,
302 }
303 .into());
304 }
305 let tenant = tenant::try_register_tenant_tables(tenant_tables)
309 .map_err(ScopeInitError::RegistryUnavailable)?;
310 let owner = owner::try_register_owner_tables(owner_tables)
311 .map_err(ScopeInitError::RegistryUnavailable)?;
312 seal_initialized_if_populated(ScopeRegistryCounts { tenant, owner })
313}
314
315pub fn declare_no_scoped_tables(reason: &'static str) -> Result<(), ScopeInitError> {
321 if live_registered_total()? != 0 {
325 return Err(ScopeInitError::RegistryUnavailable(
326 "registries are not empty; use init_scope_registries instead".to_string(),
327 ));
328 }
329 SCOPE_MODE.declare_initialized()?;
330 NO_SCOPED_TABLES_REASON.get_or_init(|| reason);
331 Ok(())
332}
333
334static NO_SCOPED_TABLES_REASON: std::sync::OnceLock<&'static str> = std::sync::OnceLock::new();
335
336pub fn no_scoped_tables_reason() -> Option<&'static str> {
338 NO_SCOPED_TABLES_REASON.get().copied()
339}
340
341pub fn declare_policy_only_isolation(reason: &'static str) -> Result<(), ScopeModeConflict> {
348 SCOPE_MODE.declare_policy_only(reason)
349}
350
351pub fn policy_only_reason() -> Option<&'static str> {
353 SCOPE_MODE.policy_only_reason()
354}
355
356#[derive(Debug, Clone, PartialEq, Eq)]
373pub struct SuperAdminToken {
374 _private: (),
375}
376
377impl SuperAdminToken {
378 pub fn for_system_process(_reason: &str) -> Self {
388 Self { _private: () }
389 }
390
391 pub fn for_webhook(_source: &str) -> Self {
397 Self { _private: () }
398 }
399
400 pub fn for_auth(_operation: &str) -> Self {
406 Self { _private: () }
407 }
408}
409
410#[derive(Debug, Clone, PartialEq, Eq)]
412pub struct RlsContext {
413 pub tenant_id: String,
416
417 is_super_admin: bool,
424
425 is_global: bool,
428
429 user_id: String,
432}
433
434impl RlsContext {
435 pub fn tenant(tenant_id: &str) -> Self {
437 Self {
438 tenant_id: tenant_id.to_string(),
439 is_super_admin: false,
440 is_global: false,
441 user_id: String::new(),
442 }
443 }
444
445 pub fn global() -> Self {
450 Self {
451 tenant_id: String::new(),
452 is_super_admin: false,
453 is_global: true,
454 user_id: String::new(),
455 }
456 }
457
458 pub fn super_admin(_token: SuperAdminToken) -> Self {
466 let nil = "00000000-0000-0000-0000-000000000000".to_string();
467 Self {
468 tenant_id: nil,
469 is_super_admin: true,
470 is_global: false,
471 user_id: String::new(),
472 }
473 }
474
475 pub fn empty() -> Self {
480 Self {
481 tenant_id: String::new(),
482 is_super_admin: false,
483 is_global: false,
484 user_id: String::new(),
485 }
486 }
487
488 pub fn user(user_id: &str) -> Self {
494 Self {
495 tenant_id: String::new(),
496 is_super_admin: false,
497 is_global: false,
498 user_id: user_id.to_string(),
499 }
500 }
501
502 pub fn with_user(mut self, user_id: &str) -> Self {
507 self.user_id = user_id.to_string();
508 self
509 }
510
511 pub fn has_tenant(&self) -> bool {
513 !self.tenant_id.is_empty()
514 }
515
516 pub fn has_user(&self) -> bool {
518 !self.user_id.is_empty()
519 }
520
521 pub fn user_id(&self) -> &str {
523 &self.user_id
524 }
525
526 pub fn bypasses_rls(&self) -> bool {
528 self.is_super_admin
529 }
530
531 pub fn is_global(&self) -> bool {
533 self.is_global
534 }
535}
536
537impl std::fmt::Display for RlsContext {
538 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
539 if self.is_super_admin {
540 write!(f, "RlsContext(super_admin)")
541 } else if self.is_global {
542 write!(f, "RlsContext(global)")
543 } else if !self.tenant_id.is_empty() {
544 write!(f, "RlsContext(tenant={})", self.tenant_id)
545 } else {
546 write!(f, "RlsContext(none)")
547 }
548 }
549}
550
551#[cfg(test)]
552mod tests {
553 use super::*;
554
555 #[test]
556 fn scope_mode_seals_one_way_policy_only_first() {
557 let mode = ScopeModeCoordinator::new();
559 assert_eq!(mode.state(), ScopeRegistryState::Uninitialized);
560 assert_eq!(mode.policy_only_reason(), None);
561
562 mode.declare_policy_only("db policies only").unwrap();
563 assert_eq!(mode.state(), ScopeRegistryState::PolicyOnly);
564 assert_eq!(mode.policy_only_reason(), Some("db policies only"));
565
566 mode.declare_policy_only("second reason").unwrap();
568 assert_eq!(mode.policy_only_reason(), Some("db policies only"));
569
570 let err = mode.declare_initialized().unwrap_err();
572 assert_eq!(err.current, ScopeRegistryState::PolicyOnly);
573 assert_eq!(err.requested, ScopeRegistryState::Initialized);
574 assert_eq!(mode.state(), ScopeRegistryState::PolicyOnly);
575 }
576
577 #[test]
578 fn scope_mode_seals_one_way_initialized_first() {
579 let mode = ScopeModeCoordinator::new();
580 mode.declare_initialized().unwrap();
581 mode.declare_initialized().unwrap();
582 assert_eq!(mode.state(), ScopeRegistryState::Initialized);
583
584 let err = mode.declare_policy_only("too late").unwrap_err();
585 assert_eq!(err.current, ScopeRegistryState::Initialized);
586 assert_eq!(mode.state(), ScopeRegistryState::Initialized);
587 assert_eq!(
588 mode.policy_only_reason(),
589 None,
590 "a refused declaration must not record a reason"
591 );
592 }
593
594 #[test]
595 fn low_level_registration_is_mode_neutral_and_empty_init_refuses_to_seal() {
596 let before = scope_registry_state();
601 tenant::try_register_tenant_tables(&[]).unwrap();
602 owner::try_register_owner_tables(&[]).unwrap();
603 assert_eq!(
604 scope_registry_state(),
605 before,
606 "empty low-level registration must not change the mode"
607 );
608 tenant::try_register_tenant_tables(&[("_mode_neutral_probe", "tenant_id")]).unwrap();
609 assert_eq!(
610 scope_registry_state(),
611 before,
612 "non-empty low-level registration must not change the mode either"
613 );
614 assert!(
615 tenant::try_tenant_table_count().unwrap() > 0,
616 "…but the table IS recorded"
617 );
618 }
619
620 #[test]
621 fn init_from_tables_refuses_empty_then_seals_on_real_registration() {
622 assert_eq!(
626 seal_initialized_if_populated_with(
627 0,
628 ScopeRegistryCounts {
629 tenant: 0,
630 owner: 0
631 }
632 ),
633 Err(ScopeInitError::NoScopedTables)
634 );
635 let counts =
636 init_scope_registries_from_tables(&[("_init_from_tables_t", "tenant_id")], &[])
637 .expect("one real table seals Initialized");
638 assert_eq!(
639 counts,
640 ScopeRegistryCounts {
641 tenant: 1,
642 owner: 0
643 }
644 );
645 assert_eq!(scope_registry_state(), ScopeRegistryState::Initialized);
646 }
647
648 #[test]
649 fn init_from_migrate_schema_reports_registry_failure_instead_of_zero() {
650 let schema = crate::migrate::parse_qail(
651 "table _init_schema_orders {\n id UUID primary_key\n tenant_id UUID\n}\n",
652 )
653 .unwrap();
654 assert_eq!(tenant::register_from_migrate_schema(&schema), Ok(1));
655 assert_eq!(owner::register_from_migrate_schema(&schema), Ok(0));
656 }
657
658 fn poison<T: Send + Sync + 'static>(lock: &'static std::sync::RwLock<T>) {
661 let result = std::thread::spawn(move || {
662 let _guard = lock.write().unwrap();
663 panic!("poison the registry lock");
664 })
665 .join();
666 assert!(result.is_err(), "writer thread must have panicked");
667 assert!(lock.is_poisoned());
668 }
669
670 #[test]
671 fn poisoned_tenant_registry_is_an_error_for_count_and_registration() {
672 let lock: &'static std::sync::RwLock<tenant::TenantRegistry> = Box::leak(Box::new(
675 std::sync::RwLock::new(tenant::TenantRegistry::new()),
676 ));
677 tenant::register_into(lock, &[("_poison_orders", "tenant_id")]).unwrap();
679 poison(lock);
680
681 let count = tenant::count_in(lock).expect_err("poisoned count must not read as 0");
682 assert!(count.contains("poisoned"), "{count}");
683 let reg = tenant::register_into(lock, &[("_poison_more", "tenant_id")])
684 .expect_err("poisoned registration must not be silently discarded");
685 assert!(reg.contains("poisoned"), "{reg}");
686 }
687
688 #[test]
689 fn poisoned_owner_registry_is_an_error_for_count_and_registration() {
690 let lock: &'static std::sync::RwLock<owner::OwnerRegistry> = Box::leak(Box::new(
691 std::sync::RwLock::new(owner::OwnerRegistry::new()),
692 ));
693 owner::register_into(lock, &[("_poison_listings", "seller_id")]).unwrap();
694 poison(lock);
695
696 assert!(
697 owner::count_in(lock)
698 .expect_err("poisoned count must not read as 0")
699 .contains("poisoned")
700 );
701 assert!(
702 owner::register_into(lock, &[("_poison_more", "seller_id")])
703 .expect_err("poisoned registration must not be silently discarded")
704 .contains("poisoned")
705 );
706 }
707
708 #[test]
709 fn poisoned_registry_lookup_is_an_error_not_unregistered() {
710 let tenant_lock: &'static std::sync::RwLock<tenant::TenantRegistry> = Box::leak(Box::new(
714 std::sync::RwLock::new(tenant::TenantRegistry::new()),
715 ));
716 tenant::register_into(tenant_lock, &[("_poison_lookup_orders", "tenant_id")]).unwrap();
717 assert_eq!(
718 tenant::lookup_in(tenant_lock, "_poison_lookup_orders"),
719 Ok(Some("tenant_id".to_string()))
720 );
721 poison(tenant_lock);
722 let err = tenant::lookup_in(tenant_lock, "_poison_lookup_orders")
723 .expect_err("a registered table behind a poisoned lock must NOT read as None");
724 assert!(err.contains("poisoned"), "{err}");
725
726 let owner_lock: &'static std::sync::RwLock<owner::OwnerRegistry> = Box::leak(Box::new(
727 std::sync::RwLock::new(owner::OwnerRegistry::new()),
728 ));
729 owner::register_into(owner_lock, &[("_poison_lookup_listings", "seller_id")]).unwrap();
730 poison(owner_lock);
731 assert!(
732 owner::lookup_in(owner_lock, "_poison_lookup_listings")
733 .expect_err("poisoned owner lookup must error")
734 .contains("poisoned")
735 );
736 }
737
738 #[test]
739 fn compatibility_lookup_collapses_error_but_scoping_does_not_use_it() {
740 let lock: &'static std::sync::RwLock<tenant::TenantRegistry> = Box::leak(Box::new(
743 std::sync::RwLock::new(tenant::TenantRegistry::new()),
744 ));
745 tenant::register_into(lock, &[("_compat_orders", "tenant_id")]).unwrap();
746 poison(lock);
747 assert_eq!(
749 tenant::lookup_in(lock, "_compat_orders").ok().flatten(),
750 None
751 );
752 assert!(tenant::lookup_in(lock, "_compat_orders").is_err());
755 }
756
757 #[test]
758 fn registry_unavailable_never_seals_initialized() {
759 let mode = ScopeModeCoordinator::new();
763 let live: Result<usize, ScopeInitError> = Err(ScopeInitError::RegistryUnavailable(
764 "owner registry lock poisoned".into(),
765 ));
766 let outcome = live.and_then(|n| {
767 if n == 0 {
768 Err(ScopeInitError::NoScopedTables)
769 } else {
770 mode.declare_initialized().map_err(Into::into)
771 }
772 });
773 assert!(matches!(
774 outcome,
775 Err(ScopeInitError::RegistryUnavailable(_))
776 ));
777 assert_eq!(mode.state(), ScopeRegistryState::Uninitialized);
778 }
779
780 #[test]
781 fn test_tenant_context() {
782 let ctx = RlsContext::tenant("t-123");
783 assert_eq!(ctx.tenant_id, "t-123");
784 assert!(!ctx.bypasses_rls());
785 assert!(ctx.has_tenant());
786 }
787
788 #[test]
789 fn test_super_admin_via_named_constructors() {
790 let token = SuperAdminToken::for_system_process("test");
791 let ctx = RlsContext::super_admin(token);
792 assert!(ctx.bypasses_rls());
793
794 let token = SuperAdminToken::for_webhook("test");
795 let ctx = RlsContext::super_admin(token);
796 assert!(ctx.bypasses_rls());
797
798 let token = SuperAdminToken::for_auth("test");
799 let ctx = RlsContext::super_admin(token);
800 assert!(ctx.bypasses_rls());
801 }
802
803 #[test]
804 fn test_display() {
805 let token = SuperAdminToken::for_system_process("test_display");
806 assert_eq!(
807 RlsContext::super_admin(token).to_string(),
808 "RlsContext(super_admin)"
809 );
810 assert_eq!(RlsContext::tenant("x").to_string(), "RlsContext(tenant=x)");
811 }
812
813 #[test]
814 fn test_equality() {
815 let a = RlsContext::tenant("t-1");
816 let b = RlsContext::tenant("t-1");
817 let c = RlsContext::tenant("t-2");
818 assert_eq!(a, b);
819 assert_ne!(a, c);
820 }
821
822 #[test]
823 fn test_empty_context() {
824 let ctx = RlsContext::empty();
825 assert!(!ctx.has_tenant());
826 assert!(!ctx.bypasses_rls());
827 assert!(!ctx.is_global());
828 }
829
830 #[test]
831 fn test_global_context() {
832 let ctx = RlsContext::global();
833 assert!(!ctx.has_tenant());
834 assert!(!ctx.bypasses_rls());
835 assert!(ctx.is_global());
836 assert_eq!(ctx.to_string(), "RlsContext(global)");
837 }
838
839 #[test]
840 fn test_for_system_process() {
841 let token = SuperAdminToken::for_system_process("cron::check_expired_holds");
842 let ctx = RlsContext::super_admin(token);
843 assert!(ctx.bypasses_rls());
844 }
845
846 #[test]
847 fn test_for_webhook() {
848 let token = SuperAdminToken::for_webhook("xendit_callback");
849 let ctx = RlsContext::super_admin(token);
850 assert!(ctx.bypasses_rls());
851 }
852
853 #[test]
854 fn test_for_auth() {
855 let token = SuperAdminToken::for_auth("login");
856 let ctx = RlsContext::super_admin(token);
857 assert!(ctx.bypasses_rls());
858 }
859
860 #[test]
861 fn test_all_constructors_produce_equal_tokens() {
862 let a = SuperAdminToken::for_system_process("a");
863 let b = SuperAdminToken::for_webhook("b");
864 let c = SuperAdminToken::for_auth("c");
865 assert_eq!(a, b);
867 assert_eq!(b, c);
868 }
869
870 #[test]
871 fn test_user_context() {
872 let ctx = RlsContext::user("550e8400-e29b-41d4-a716-446655440000");
873 assert!(!ctx.has_tenant());
874 assert!(!ctx.bypasses_rls());
875 assert!(!ctx.is_global());
876 assert!(ctx.has_user());
877 assert_eq!(ctx.user_id(), "550e8400-e29b-41d4-a716-446655440000");
878 }
879
880 #[test]
881 fn test_with_user_preserves_tenant_scope() {
882 let ctx = RlsContext::tenant("tenant-1").with_user("user-1");
883
884 assert_eq!(ctx.tenant_id, "tenant-1");
885 assert_eq!(ctx.user_id(), "user-1");
886 assert!(ctx.has_tenant());
887 assert!(ctx.has_user());
888 assert!(!ctx.bypasses_rls());
889 }
890
891 #[test]
892 fn test_user_context_display() {
893 let ctx = RlsContext::user("u-123");
894 assert_eq!(ctx.to_string(), "RlsContext(none)");
895 }
898
899 #[test]
900 fn test_other_constructors_have_no_user() {
901 assert!(!RlsContext::tenant("t-1").has_user());
902 assert!(!RlsContext::global().has_user());
903 assert!(!RlsContext::empty().has_user());
904 let token = SuperAdminToken::for_auth("test");
905 assert!(!RlsContext::super_admin(token).has_user());
906 }
907}