allowthem_saas/
router.rs

1use std::path::PathBuf;
2use std::sync::Arc;
3use std::time::{Duration, Instant};
4
5use axum::body::Body;
6use axum::extract::State;
7use axum::http::{Request, StatusCode, header, request::Parts};
8use axum::middleware::Next;
9use axum::response::{IntoResponse, Response};
10use dashmap::DashMap;
11use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePool};
12
13use allowthem_core::AllowThem;
14
15use crate::cache::{HandleCache, SlugCache, TenantMeta};
16use crate::control_db::ControlDb;
17use crate::error::SaasError;
18use crate::tenants::{TenantBuilderConfig, TenantId, TenantStatus};
19
20pub(crate) const RESERVED_SLUGS: &[&str] = &[
21    "www",
22    "api",
23    "admin",
24    "auth",
25    "manage",
26    "oauth",
27    "app",
28    "dashboard",
29    "status",
30    "mail",
31    "docs",
32    "help",
33    "support",
34    "static",
35    "cdn",
36    "assets",
37    "id",
38    "sso",
39    "allowthem",
40];
41
42pub(crate) fn is_reserved_slug(slug: &str) -> bool {
43    RESERVED_SLUGS.contains(&slug)
44}
45
46#[derive(Debug, PartialEq)]
47pub enum SlugOrRoot {
48    Slug(String),
49    Root,
50}
51
52/// Extract the tenant slug from a Host header value.
53///
54/// `host_raw` may contain a port (e.g. `"acme.example.com:8080"`). Returns:
55/// - `Some(Root)` if host (without port) equals `base_domain` exactly
56/// - `Some(Slug(s))` if host is `<s>.<base_domain>` and `s` has no nested dots
57/// - `None` for any other host (unknown domain, multi-level subdomain, etc.)
58pub fn parse_slug(host_raw: &str, base_domain: &str) -> Option<SlugOrRoot> {
59    let host = host_raw
60        .split(':')
61        .next()
62        .unwrap_or(host_raw)
63        .to_lowercase();
64
65    if host == base_domain {
66        return Some(SlugOrRoot::Root);
67    }
68
69    let suffix = format!(".{base_domain}");
70    let label = host.strip_suffix(suffix.as_str())?;
71
72    if label.is_empty() || label.contains('.') {
73        return None;
74    }
75
76    Some(SlugOrRoot::Slug(label.to_owned()))
77}
78
79#[derive(Clone)]
80pub struct TenantRouterState {
81    pub control_db: Arc<ControlDb>,
82    pub slug_cache: SlugCache,
83    pub handle_cache: HandleCache,
84    pub tenant_data_dir: PathBuf,
85    pub config: Arc<TenantBuilderConfig>,
86    pub seen_times: Arc<DashMap<TenantId, Instant>>,
87    /// When `Some`, requests for the root domain (`<base_domain>` exactly)
88    /// get this handle inserted into request extensions before falling
89    /// through to the next layer. Used so the shared auth routes can serve
90    /// the dashboard from `dashboard.db` without per-handler branching.
91    /// `None` is the default for tests and any non-SaaS embedding.
92    pub dashboard_handle: Option<AllowThem>,
93}
94
95pub async fn tenant_router_middleware(
96    State(state): State<TenantRouterState>,
97    mut request: Request<Body>,
98    next: Next,
99) -> Response {
100    let host = match request
101        .headers()
102        .get(header::HOST)
103        .and_then(|v| v.to_str().ok())
104    {
105        Some(h) => h.to_owned(),
106        None => return StatusCode::BAD_REQUEST.into_response(),
107    };
108
109    let base_domain = &state.config.base_domain;
110    let slug = match parse_slug(&host, base_domain) {
111        Some(SlugOrRoot::Slug(s)) => s,
112        Some(SlugOrRoot::Root) => {
113            if let Some(ref dh) = state.dashboard_handle {
114                request.extensions_mut().insert(dh.clone());
115            }
116            return next.run(request).await;
117        }
118        None => return StatusCode::NOT_FOUND.into_response(),
119    };
120
121    let meta = match state
122        .slug_cache
123        .get_or_fetch(&slug, || {
124            let ctrl = state.control_db.clone();
125            let s = slug.clone();
126            async move { ctrl.tenant_meta_by_slug(&s).await }
127        })
128        .await
129    {
130        Ok(Some(m)) => m,
131        Ok(None) => return StatusCode::NOT_FOUND.into_response(),
132        Err(e) => {
133            tracing::error!(error = %e, slug, "slug cache fetch failed");
134            return StatusCode::INTERNAL_SERVER_ERROR.into_response();
135        }
136    };
137
138    if matches!(meta.status, TenantStatus::Deleted) {
139        return StatusCode::NOT_FOUND.into_response();
140    }
141
142    if matches!(meta.status, TenantStatus::Suspended) {
143        request.extensions_mut().insert(meta);
144        return next.run(request).await;
145    }
146
147    let tenant_id = meta.id;
148    let handle = {
149        let ctrl = state.control_db.clone();
150        let dir = state.tenant_data_dir.clone();
151        let cfg = state.config.clone();
152        let slug_owned = slug.clone();
153        match state
154            .handle_cache
155            .get_or_init(tenant_id, async move {
156                build_handle(ctrl, dir, cfg, tenant_id, &slug_owned).await
157            })
158            .await
159        {
160            Ok(h) => h,
161            Err(e) => {
162                tracing::error!(error = %e, slug = slug.as_str(), "handle cache init failed");
163                return StatusCode::INTERNAL_SERVER_ERROR.into_response();
164            }
165        }
166    };
167
168    tracing::debug!(tenant_id = %tenant_id.as_uuid(), slug, "tenant request");
169
170    request.extensions_mut().insert(meta);
171    request.extensions_mut().insert(handle);
172
173    debounce_last_seen(&state, tenant_id);
174
175    next.run(request).await
176}
177
178pub(crate) async fn build_handle(
179    control_db: Arc<ControlDb>,
180    tenant_data_dir: PathBuf,
181    config: Arc<TenantBuilderConfig>,
182    tenant_id: TenantId,
183    slug: &str,
184) -> Result<AllowThem, SaasError> {
185    let tenant = control_db
186        .tenant_by_id_raw(tenant_id.as_bytes())
187        .await?
188        .ok_or(SaasError::TenantNotFound)?;
189    build_handle_with_path(&tenant.db_path, tenant_id, &tenant_data_dir, &config, slug).await
190}
191
192pub async fn build_handle_with_path(
193    db_file: &str,
194    tenant_id: TenantId,
195    tenant_data_dir: &std::path::Path,
196    config: &TenantBuilderConfig,
197    slug: &str,
198) -> Result<AllowThem, SaasError> {
199    let full_path = tenant_data_dir.join(db_file);
200
201    let opts = SqliteConnectOptions::new()
202        .filename(&full_path)
203        .create_if_missing(false)
204        .pragma("foreign_keys", "ON")
205        .journal_mode(SqliteJournalMode::Wal)
206        .busy_timeout(Duration::from_millis(5000));
207    let pool = SqlitePool::connect_with(opts)
208        .await
209        .map_err(|e| SaasError::ProvisionFailed(e.to_string()))?;
210
211    // No `cookie_domain` — host-only cookies satisfy the `__Host-` prefix
212    // (see `crates/core/sessions.rs:313`). Cookie name + Secure attribute
213    // are picked per environment via `tenant_cookie_name`.
214    let mut builder = allowthem_core::AllowThemBuilder::with_pool(pool.clone())
215        .mfa_key(config.mfa_key)
216        .signing_key(config.signing_key)
217        .csrf_key(config.csrf_key)
218        .base_url(format!("https://{slug}.{}", config.base_domain))
219        .cookie_name(crate::tenants::tenant_cookie_name(config.is_production))
220        .cookie_secure(config.is_production);
221
222    // Prefer the per-tenant email factory when present; fall back to the
223    // shared sender for embedded/test paths that haven't migrated. The
224    // factory needs the tenant pool because it reads `allowthem_email_config`
225    // and decrypts secrets with the deployment's mfa_key.
226    if let Some(factory) = &config.email_sender_factory {
227        let sender = factory
228            .for_tenant(tenant_id, &pool)
229            .await
230            .map_err(|e| SaasError::ProvisionFailed(e.to_string()))?;
231        builder = builder.email_sender(Box::new(sender));
232    } else if let Some(sender) = &config.email_sender {
233        builder = builder.email_sender(Box::new(sender.clone()));
234    }
235
236    // Prefer the per-tenant factory when present; fall back to the shared
237    // sink for embedded/test paths that haven't migrated to the factory.
238    if let Some(factory) = &config.event_sink_factory {
239        let sink = factory.for_tenant(tenant_id);
240        builder = builder.event_sink(Box::new(sink));
241    } else if let Some(sink) = &config.event_sink {
242        builder = builder.event_sink(Box::new(sink.clone()));
243    }
244
245    if let Some(mau_sink) = &config.mau_sink {
246        let mau_sink = mau_sink.clone();
247        let cb: allowthem_core::OnUserActive = std::sync::Arc::new(move |user_id, ts| {
248            let mau_sink = mau_sink.clone();
249            tokio::spawn(async move {
250                if let Err(e) = mau_sink.record_active(tenant_id, user_id, ts).await {
251                    tracing::warn!(
252                        tenant_id = %tenant_id.as_uuid(),
253                        user_id = %user_id,
254                        error = %e,
255                        "MauSink::record_active failed"
256                    );
257                }
258            });
259        });
260        builder = builder.on_user_active(cb);
261    }
262
263    builder
264        .build()
265        .await
266        .map_err(|e| SaasError::ProvisionFailed(e.to_string()))
267}
268
269/// Pre-warm the handle cache with the most recently active tenants.
270///
271/// Called at server startup to avoid cold-cache latency spikes for the first
272/// requests to each tenant. Errors per tenant are logged and swallowed so one
273/// bad tenant can't block the rest.
274pub async fn pre_warm(
275    control_db: Arc<ControlDb>,
276    cache: &HandleCache,
277    tenant_data_dir: PathBuf,
278    config: Arc<TenantBuilderConfig>,
279    count: i64,
280) {
281    let ids = match control_db.most_recently_seen_tenants(count).await {
282        Ok(v) => v,
283        Err(e) => {
284            tracing::warn!(error = %e, "pre_warm: failed to fetch tenant list");
285            return;
286        }
287    };
288
289    let futs: Vec<_> = ids
290        .into_iter()
291        .map(|tenant_id| {
292            let cache = cache.clone();
293            let ctrl = control_db.clone();
294            let dir = tenant_data_dir.clone();
295            let cfg = config.clone();
296            async move {
297                let tenant = match ctrl.tenant_by_id_raw(tenant_id.as_bytes()).await {
298                    Ok(Some(t)) => t,
299                    Ok(None) => {
300                        tracing::warn!(
301                            tenant_id = %tenant_id.as_uuid(),
302                            "pre_warm: tenant not found"
303                        );
304                        return;
305                    }
306                    Err(e) => {
307                        tracing::warn!(
308                            tenant_id = %tenant_id.as_uuid(),
309                            error = %e,
310                            "pre_warm: fetch failed"
311                        );
312                        return;
313                    }
314                };
315                let db_file = tenant.db_path.clone();
316                let slug = tenant.slug.clone();
317                let result = cache
318                    .get_or_init(tenant_id, async move {
319                        build_handle_with_path(&db_file, tenant_id, &dir, &cfg, &slug).await
320                    })
321                    .await;
322                if let Err(e) = result {
323                    tracing::warn!(
324                        tenant_id = %tenant_id.as_uuid(),
325                        error = %e,
326                        "pre_warm: handle init failed"
327                    );
328                }
329            }
330        })
331        .collect();
332
333    futures::future::join_all(futs).await;
334}
335
336pub(crate) fn debounce_last_seen(state: &TenantRouterState, tenant_id: TenantId) {
337    const DEBOUNCE: Duration = Duration::from_secs(60);
338
339    let now = Instant::now();
340    let last = state.seen_times.get(&tenant_id).map(|v| *v);
341
342    if last.is_some_and(|t| now.duration_since(t) < DEBOUNCE) {
343        return;
344    }
345
346    state.seen_times.insert(tenant_id, now);
347
348    let ctrl = state.control_db.clone();
349    tokio::spawn(async move {
350        if let Err(e) = ctrl.touch_last_seen(&tenant_id).await {
351            tracing::warn!(error = %e, tenant_id = %tenant_id.as_uuid(), "touch_last_seen failed");
352        }
353    });
354}
355
356/// Extractor that rejects requests for suspended or deleted tenants.
357///
358/// Returns 503 for suspended tenants, 404 for deleted, and Ok for active tenants
359/// or routes without a tenant context (root domain).
360#[derive(Debug, Clone)]
361pub struct RequireActiveTenant;
362
363impl<S: Send + Sync> axum::extract::FromRequestParts<S> for RequireActiveTenant {
364    type Rejection = Response;
365
366    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
367        let Some(meta) = parts.extensions.get::<TenantMeta>() else {
368            return Ok(RequireActiveTenant);
369        };
370        match meta.status {
371            TenantStatus::Active => Ok(RequireActiveTenant),
372            TenantStatus::Suspended => Err(StatusCode::SERVICE_UNAVAILABLE.into_response()),
373            TenantStatus::Deleted => Err(StatusCode::NOT_FOUND.into_response()),
374        }
375    }
376}
377
378#[cfg(test)]
379mod tests {
380    use super::*;
381    use axum::extract::FromRequestParts;
382    use axum::http::Request;
383
384    fn base() -> &'static str {
385        "example.com"
386    }
387
388    #[test]
389    fn parse_slug_root_domain_returns_root() {
390        assert_eq!(parse_slug("example.com", base()), Some(SlugOrRoot::Root));
391    }
392
393    #[test]
394    fn parse_slug_subdomain_returns_slug() {
395        assert_eq!(
396            parse_slug("acme.example.com", base()),
397            Some(SlugOrRoot::Slug("acme".into()))
398        );
399    }
400
401    #[test]
402    fn parse_slug_strips_port() {
403        assert_eq!(
404            parse_slug("acme.example.com:8080", base()),
405            Some(SlugOrRoot::Slug("acme".into()))
406        );
407    }
408
409    #[test]
410    fn parse_slug_root_with_port() {
411        assert_eq!(
412            parse_slug("example.com:443", base()),
413            Some(SlugOrRoot::Root)
414        );
415    }
416
417    #[test]
418    fn parse_slug_multi_level_returns_none() {
419        assert_eq!(parse_slug("a.b.example.com", base()), None);
420    }
421
422    #[test]
423    fn parse_slug_unknown_domain_returns_none() {
424        assert_eq!(parse_slug("other.io", base()), None);
425    }
426
427    #[test]
428    fn parse_slug_lowercases_slug() {
429        assert_eq!(
430            parse_slug("ACME.example.com", base()),
431            Some(SlugOrRoot::Slug("acme".into()))
432        );
433    }
434
435    #[test]
436    fn parse_slug_empty_host_returns_none() {
437        assert_eq!(parse_slug("", base()), None);
438    }
439
440    #[test]
441    fn parse_slug_unrelated_subdomain_returns_none() {
442        assert_eq!(parse_slug("acme.other.com", base()), None);
443    }
444
445    #[test]
446    fn parse_slug_does_not_reject_reserved() {
447        // Reserved slugs are blocked at provisioning, not at routing time.
448        assert_eq!(
449            parse_slug("admin.example.com", base()),
450            Some(SlugOrRoot::Slug("admin".into()))
451        );
452    }
453
454    fn make_meta(status: TenantStatus) -> TenantMeta {
455        use uuid::Uuid;
456        TenantMeta {
457            id: TenantId::from(Uuid::nil()),
458            status,
459            plan_id: vec![],
460        }
461    }
462
463    async fn make_state() -> TenantRouterState {
464        use std::str::FromStr;
465        let opts = sqlx::sqlite::SqliteConnectOptions::from_str("sqlite::memory:")
466            .unwrap()
467            .pragma("foreign_keys", "ON");
468        let pool = sqlx::SqlitePool::connect_with(opts).await.unwrap();
469        let ctrl = Arc::new(crate::control_db::ControlDb::new(pool).await.unwrap());
470        TenantRouterState {
471            control_db: ctrl,
472            slug_cache: SlugCache::new(10, 60),
473            handle_cache: HandleCache::new(10),
474            tenant_data_dir: PathBuf::from("/tmp"),
475            config: Arc::new(TenantBuilderConfig {
476                mfa_key: [0u8; 32],
477                signing_key: [0u8; 32],
478                csrf_key: [0u8; 32],
479                base_domain: "example.com".into(),
480                is_production: false,
481                email_sender: None,
482                event_sink: None,
483                event_sink_factory: None,
484                mau_sink: None,
485                email_sender_factory: None,
486            }),
487            seen_times: Arc::new(DashMap::new()),
488            dashboard_handle: None,
489        }
490    }
491
492    #[tokio::test]
493    async fn extractor_passes_when_no_meta() {
494        let (mut parts, _) = Request::new(()).into_parts();
495        let result: Result<RequireActiveTenant, _> =
496            RequireActiveTenant::from_request_parts(&mut parts, &()).await;
497        assert!(result.is_ok());
498    }
499
500    #[tokio::test]
501    async fn extractor_passes_when_active() {
502        let (mut parts, _) = Request::new(()).into_parts();
503        parts.extensions.insert(make_meta(TenantStatus::Active));
504        let result: Result<RequireActiveTenant, _> =
505            RequireActiveTenant::from_request_parts(&mut parts, &()).await;
506        assert!(result.is_ok());
507    }
508
509    #[tokio::test]
510    async fn extractor_returns_503_when_suspended() {
511        let (mut parts, _) = Request::new(()).into_parts();
512        parts.extensions.insert(make_meta(TenantStatus::Suspended));
513        let result: Result<RequireActiveTenant, _> =
514            RequireActiveTenant::from_request_parts(&mut parts, &()).await;
515        let err = result.unwrap_err();
516        assert_eq!(err.status(), StatusCode::SERVICE_UNAVAILABLE);
517    }
518
519    #[tokio::test]
520    async fn debounce_inserts_into_seen_times() {
521        let state = make_state().await;
522        let id = TenantId::from(uuid::Uuid::nil());
523        debounce_last_seen(&state, id);
524        assert!(state.seen_times.contains_key(&id));
525    }
526
527    #[tokio::test]
528    async fn debounce_suppresses_rapid_repeat() {
529        let state = make_state().await;
530        let id = TenantId::from(uuid::Uuid::from_bytes([0x11; 16]));
531        debounce_last_seen(&state, id);
532        let first_time = *state.seen_times.get(&id).unwrap();
533        // Second call within DEBOUNCE window should not update the timestamp.
534        debounce_last_seen(&state, id);
535        let second_time = *state.seen_times.get(&id).unwrap();
536        assert_eq!(first_time, second_time);
537    }
538
539    #[tokio::test]
540    async fn pre_warm_noop_when_no_active_tenants() {
541        let state = make_state().await;
542        // Empty DB — most_recently_seen_tenants returns an empty list.
543        pre_warm(
544            state.control_db.clone(),
545            &state.handle_cache,
546            state.tenant_data_dir.clone(),
547            state.config.clone(),
548            10,
549        )
550        .await;
551        // The test passing (no panic, entry_count stays 0) is the assertion.
552        assert_eq!(state.handle_cache.entry_count(), 0);
553    }
554
555    #[tokio::test]
556    async fn pre_warm_swallows_error_for_missing_tenant_db() {
557        use uuid::Uuid;
558        let state = make_state().await;
559
560        // Insert a tenant row with last_seen_at set so it shows up in pre_warm.
561        let plan_row = sqlx::query("SELECT id FROM tenant_plans LIMIT 1")
562            .fetch_one(state.control_db.pool())
563            .await
564            .unwrap();
565        let plan_id: Vec<u8> = sqlx::Row::try_get(&plan_row, "id").unwrap();
566        let tid = Uuid::now_v7();
567        sqlx::query(
568            "INSERT INTO tenants (id, name, slug, owner_email, plan_id, status, db_path, last_seen_at) \
569             VALUES (?1, 'Test', 'pretest', 't@t.com', ?2, 'active', 'pretest.db', datetime('now'))",
570        )
571        .bind(tid.as_bytes().as_ref())
572        .bind(&plan_id)
573        .execute(state.control_db.pool())
574        .await
575        .unwrap();
576
577        // pre_warm should attempt to open pretest.db (which doesn't exist), log the error,
578        // and return without panicking.
579        pre_warm(
580            state.control_db.clone(),
581            &state.handle_cache,
582            state.tenant_data_dir.clone(),
583            state.config.clone(),
584            10,
585        )
586        .await;
587    }
588
589    // -- Root-branch dashboard handle injection (99c.1 Task 4) -----------------
590
591    /// Build the middleware as a tower-style closure and run a single request
592    /// through it, returning whether `Extension<AllowThem>` was visible to the
593    /// inner handler.
594    async fn root_request_sees_extension(state: TenantRouterState, host: &str) -> bool {
595        use axum::Router;
596        use axum::body::Body;
597        use axum::http::header;
598        use axum::middleware::from_fn_with_state;
599        use axum::routing::get;
600        use std::sync::Arc;
601        use std::sync::atomic::{AtomicBool, Ordering};
602        use tower::ServiceExt;
603
604        let observed = Arc::new(AtomicBool::new(false));
605        let observed_for_handler = observed.clone();
606
607        let app = Router::new()
608            .route(
609                "/",
610                get(move |req: Request<Body>| {
611                    let observed = observed_for_handler.clone();
612                    async move {
613                        if req.extensions().get::<AllowThem>().is_some() {
614                            observed.store(true, Ordering::SeqCst);
615                        }
616                        StatusCode::OK
617                    }
618                }),
619            )
620            .layer(from_fn_with_state(state, tenant_router_middleware));
621
622        let req = Request::builder()
623            .uri("/")
624            .header(header::HOST, host)
625            .body(Body::empty())
626            .unwrap();
627        let _resp = app.oneshot(req).await.unwrap();
628        observed.load(Ordering::SeqCst)
629    }
630
631    async fn make_dashboard_handle() -> AllowThem {
632        allowthem_core::AllowThemBuilder::new("sqlite::memory:")
633            .mfa_key([0u8; 32])
634            .signing_key([0u8; 32])
635            .csrf_key([0u8; 32])
636            .base_url("https://example.com")
637            .cookie_secure(false)
638            .build()
639            .await
640            .unwrap()
641    }
642
643    #[tokio::test]
644    async fn root_branch_injects_dashboard_handle_when_set() {
645        let dh = make_dashboard_handle().await;
646        let mut state = make_state().await;
647        state.dashboard_handle = Some(dh);
648        assert!(root_request_sees_extension(state, "example.com").await);
649    }
650
651    #[tokio::test]
652    async fn root_branch_no_injection_when_none() {
653        let state = make_state().await;
654        assert!(!root_request_sees_extension(state, "example.com").await);
655    }
656
657    // -- on_user_active wiring for MAU counting (eua.2) ------------------------
658
659    fn mau_test_builder_config(mau_sink: Option<Arc<crate::mau::MauSink>>) -> TenantBuilderConfig {
660        TenantBuilderConfig {
661            mfa_key: [1u8; 32],
662            signing_key: [2u8; 32],
663            csrf_key: [3u8; 32],
664            base_domain: "test.local".into(),
665            is_production: false,
666            email_sender: None,
667            event_sink: None,
668            event_sink_factory: None,
669            mau_sink,
670            email_sender_factory: None,
671        }
672    }
673
674    #[tokio::test]
675    async fn build_handle_with_path_registers_mau_callback_when_sink_present() {
676        use allowthem_core::types::UserId;
677
678        let dir = tempfile::tempdir().expect("tempdir");
679        let state = make_state().await;
680        let mau_sink = Arc::new(crate::mau::MauSink::new(state.control_db.clone()));
681
682        // Provision a tenant — creates the SQLite file `build_handle_with_path`
683        // will reopen below. The `mau_sink` on this provisioning config is
684        // ignored (provision_tenant builds its own handle inline); the sink
685        // wiring under test is the build_handle_with_path call afterwards.
686        let provision_config = mau_test_builder_config(None);
687        let provisioned = state
688            .control_db
689            .provision_tenant(
690                "Mau Co".into(),
691                "mauco".into(),
692                "owner@mauco.test".into(),
693                dir.path(),
694                &provision_config,
695            )
696            .await
697            .expect("provision_tenant");
698        let tenant_id = TenantId::from(provisioned.tenant.id_as_uuid().unwrap());
699
700        // Build a fresh handle wired with the MAU sink.
701        let wired_config = mau_test_builder_config(Some(mau_sink.clone()));
702        let ath = build_handle_with_path(
703            &provisioned.tenant.db_path,
704            tenant_id,
705            dir.path(),
706            &wired_config,
707            "mauco",
708        )
709        .await
710        .expect("build_handle_with_path");
711
712        // Fire the callback synthetically. The closure spawns a task that
713        // writes to the control DB; poll until the row appears.
714        let cb = ath
715            .on_user_active()
716            .expect("on_user_active must be registered when mau_sink is set");
717        let user_id = UserId::new();
718        cb(user_id, chrono::Utc::now());
719
720        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
721        loop {
722            let count: i64 =
723                sqlx::query_scalar("SELECT COUNT(*) FROM tenant_active_users WHERE tenant_id = ?1")
724                    .bind(tenant_id.as_bytes())
725                    .fetch_one(state.control_db.pool())
726                    .await
727                    .unwrap();
728            if count == 1 {
729                break;
730            }
731            if std::time::Instant::now() > deadline {
732                panic!("MAU row never appeared in tenant_active_users");
733            }
734            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
735        }
736    }
737
738    #[tokio::test]
739    async fn build_handle_with_path_no_callback_when_sink_absent() {
740        let dir = tempfile::tempdir().expect("tempdir");
741        let state = make_state().await;
742
743        let config = mau_test_builder_config(None);
744        let provisioned = state
745            .control_db
746            .provision_tenant(
747                "Plain Co".into(),
748                "plainco".into(),
749                "owner@plainco.test".into(),
750                dir.path(),
751                &config,
752            )
753            .await
754            .expect("provision_tenant");
755        let tenant_id = TenantId::from(provisioned.tenant.id_as_uuid().unwrap());
756
757        let ath = build_handle_with_path(
758            &provisioned.tenant.db_path,
759            tenant_id,
760            dir.path(),
761            &config,
762            "plainco",
763        )
764        .await
765        .expect("build_handle_with_path");
766
767        assert!(
768            ath.on_user_active().is_none(),
769            "no callback should be registered when mau_sink is None"
770        );
771    }
772
773    // -- email_sender_factory wiring (c8m.3 Task 5) ----------------------------
774
775    /// Stub `EmailSenderFactory` that captures the `tenant_id` it was called
776    /// with. Used to verify `build_handle_with_path` routes through the
777    /// factory (and not the shared `email_sender` fallback) when both are
778    /// present.
779    struct StubEmailFactory {
780        captured: Arc<std::sync::Mutex<Option<TenantId>>>,
781    }
782
783    /// Sentinel sender — never sends; presence of this type is the
784    /// "factory path was chosen" signal in the wiring test.
785    struct SentinelEmailSender;
786
787    impl allowthem_core::EmailSender for SentinelEmailSender {
788        fn send<'a>(
789            &'a self,
790            _message: &'a allowthem_core::email::EmailMessage,
791        ) -> std::pin::Pin<
792            Box<
793                dyn std::future::Future<Output = Result<(), allowthem_core::error::AuthError>>
794                    + Send
795                    + 'a,
796            >,
797        > {
798            Box::pin(std::future::ready(Ok(())))
799        }
800    }
801
802    impl crate::managed_email::EmailSenderFactory for StubEmailFactory {
803        fn for_tenant<'a>(
804            &'a self,
805            tenant_id: TenantId,
806            _tenant_pool: &'a sqlx::SqlitePool,
807        ) -> allowthem_core::auth_client::AuthFuture<'a, Arc<dyn allowthem_core::EmailSender>>
808        {
809            let captured = self.captured.clone();
810            Box::pin(async move {
811                *captured.lock().unwrap() = Some(tenant_id);
812                Ok(Arc::new(SentinelEmailSender) as Arc<dyn allowthem_core::EmailSender>)
813            })
814        }
815    }
816
817    #[tokio::test]
818    async fn build_handle_with_path_uses_email_sender_factory_when_set() {
819        // Plan c8m.3 §6 explicitly lists this as the integration test for
820        // Task 5. Asserts that when both `email_sender` and
821        // `email_sender_factory` are set, the factory wins — and that it
822        // is invoked with the supplied tenant_id (so per-tenant resolution
823        // actually receives the correct id).
824        use allowthem_core::email::{EmailMessage, EmailTemplate};
825
826        let dir = tempfile::tempdir().expect("tempdir");
827        let state = make_state().await;
828        let captured: Arc<std::sync::Mutex<Option<TenantId>>> =
829            Arc::new(std::sync::Mutex::new(None));
830        let factory = Arc::new(StubEmailFactory {
831            captured: captured.clone(),
832        });
833
834        // Set both fields so the precedence assertion is meaningful: the
835        // shared sender is a real LogEmailSender; the factory yields a
836        // SentinelEmailSender. Either could win in principle; `Some(factory)
837        // wins` is the contract under test.
838        let provision_config = TenantBuilderConfig {
839            mfa_key: [1u8; 32],
840            signing_key: [2u8; 32],
841            csrf_key: [3u8; 32],
842            base_domain: "test.local".into(),
843            is_production: false,
844            email_sender: Some(Arc::new(allowthem_core::LogEmailSender)),
845            event_sink: None,
846            event_sink_factory: None,
847            mau_sink: None,
848            email_sender_factory: None,
849        };
850
851        let provisioned = state
852            .control_db
853            .provision_tenant(
854                "Email Co".into(),
855                "emailco".into(),
856                "owner@emailco.test".into(),
857                dir.path(),
858                &provision_config,
859            )
860            .await
861            .expect("provision_tenant");
862        let tenant_id = TenantId::from(provisioned.tenant.id_as_uuid().unwrap());
863
864        let wired_config = TenantBuilderConfig {
865            email_sender_factory: Some(factory.clone()),
866            ..provision_config
867        };
868        let ath = build_handle_with_path(
869            &provisioned.tenant.db_path,
870            tenant_id,
871            dir.path(),
872            &wired_config,
873            "emailco",
874        )
875        .await
876        .expect("build_handle_with_path");
877
878        // The factory should have been invoked with our tenant_id at handle
879        // build time — not deferred to first send.
880        let observed = captured.lock().unwrap().clone();
881        assert_eq!(
882            observed,
883            Some(tenant_id),
884            "EmailSenderFactory::for_tenant must run with the supplied tenant_id"
885        );
886
887        // Sanity: send an email through the handle. Our SentinelEmailSender
888        // returns Ok immediately; LogEmailSender would also return Ok, so
889        // this isn't the discriminating signal — but it ensures the
890        // returned sender is actually wired through (not dropped silently).
891        ath.email_sender()
892            .send(&EmailMessage {
893                to: "user@example.com".into(),
894                subject: "x".into(),
895                template: EmailTemplate::PasswordReset {
896                    url: "https://x".into(),
897                    username: "user".into(),
898                },
899            })
900            .await
901            .expect("send via factory-resolved sender");
902    }
903}
allowthem_saas/router.rs

allowthem_saas/
router.rs
