//! Library half of the picloud all-in-one. `main.rs` is a thin wrapper //! that opens the pool, runs migrations, calls `build_app`, and binds //! the listener. Tests use the same `build_app` against an //! ephemeral test database. use std::sync::Arc; use std::time::Duration; use axum::middleware::from_fn_with_state; use axum::{routing::get, Json, Router}; use picloud_executor_core::{Engine, Limits}; use picloud_manager_core::{ admin_router, admins_router, api_keys_router, app_members_router, apply_router, apps_api, apps_router, attach_principal_if_present, auth_router, compile_routes, dead_letters_router, dev_emails_router, email_inbound_router, files_admin_router, groups_router, kv_admin_router, migrations, require_authenticated, route_admin_router, secrets_router, topics_router, triggers_router, AbandonedRepo, AdminPrincipalResolver, AdminSessionRepository, AdminState, AdminUserRepository, AdminsState, ApiKeyRepository, ApiKeysState, AppDomainRepository, AppMembersRepository, AppMembersState, AppRepository, ApplyService, AppsState, AuthState, AuthzRepo, DeadLetterRepo, DeadLettersState, DevEmailState, Dispatcher, DocsServiceImpl, EmailInboundState, EmailServiceImpl, FilesAdminState, FilesConfig, FilesServiceImpl, FsFilesRepo, GroupMembersRepository, GroupRepository, GroupsState, HttpConfig, HttpServiceImpl, InboundNonceDedup, KvAdminState, KvServiceImpl, OutboxEventEmitter, OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository, PostgresAdminUserRepository, PostgresApiKeyRepository, PostgresAppDomainRepository, PostgresAppMembersRepository, PostgresAppRepository, PostgresAppSecretsRepo, PostgresAppUserInvitationRepo, PostgresAppUserPasswordResetRepo, PostgresAppUserRepository, PostgresAppUserRoleRepo, PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo, PostgresDeadLetterRepo, PostgresDeadLetterService, PostgresDocsRepo, PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresGroupMembersRepository, PostgresGroupRepository, PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo, PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo, PostgresTriggerRepo, PrincipalResolver, PubsubServiceImpl, RealtimeAuthorityImpl, RepoResolver, RouteAdminState, RouteRepository, SandboxCeiling, ScriptRepository, SecretsConfig, SecretsServiceImpl, SecretsState, SubscriberTokenConfig, TopicRepo, TopicsState, TriggerConfig, TriggerRepo, TriggersState, UsersServiceConfig, UsersServiceImpl, }; use picloud_orchestrator_core::realtime::DEFAULT_GC_INTERVAL_SECS; use picloud_orchestrator_core::routing::{AppDomainTable, RouteTable}; use picloud_orchestrator_core::{ data_plane_router, realtime_router, spawn_realtime_gc, user_routes_router, DataPlaneState, ExecutionGate, InProcessBroadcaster, InboxRegistry, LocalExecutorClient, RealtimeState, }; use picloud_shared::{ DeadLetterService, DocsService, EmailService, ExecutionLogSink, FilesService, HttpService, InboxResolver, KvService, MasterKey, OutboxWriter, PubsubService, RealtimeAuthority, RealtimeBroadcaster, SecretsService, ServiceEventEmitter, Services, UsersService, API_VERSION, PRODUCT_VERSION, SDK_VERSION, WIRE_VERSION, }; use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; use tower_http::trace::TraceLayer; /// Default session TTL when `PICLOUD_SESSION_TTL_HOURS` isn't set. const DEFAULT_SESSION_TTL_HOURS: u64 = 24; /// Bundles the auth-related dependencies that both `build_app` and the /// startup bootstrap need. Built once in `main.rs` from the shared pool. pub struct AuthDeps { pub users: Arc, pub sessions: Arc, pub keys: Arc, pub ttl: Duration, } impl AuthDeps { /// Construct from a pool with the binary's standard defaults. #[must_use] pub fn from_pool(pool: PgPool) -> Self { Self { users: Arc::new(PostgresAdminUserRepository::new(pool.clone())), sessions: Arc::new(PostgresAdminSessionRepository::new(pool.clone())), keys: Arc::new(PostgresApiKeyRepository::new(pool)), ttl: read_session_ttl(), } } } fn read_session_ttl() -> Duration { let hours = std::env::var("PICLOUD_SESSION_TTL_HOURS") .ok() .and_then(|s| s.parse::().ok()) .filter(|h| *h > 0) .unwrap_or(DEFAULT_SESSION_TTL_HOURS); Duration::from_secs(hours * 3600) } /// Compose the manager + orchestrator routes on top of a shared /// Postgres pool, returning an Axum router ready to be served. /// /// All API routes live under `/api/v{API_VERSION}/...`. The dashboard /// is mounted by Caddy at `/admin/*` (its base path). Anything else /// falls through to the user-route table — user scripts can bind to /// arbitrary paths (subject to the reserved-prefix list). /// /// `auth` carries the admin user/session repositories and the /// configured session TTL. The manager-side admin endpoints /// (`/api/v1/admin/scripts/*`, `/api/v1/admin/routes/*`, /// `/api/v1/admin/admins/*`, `/api/v1/admin/auth/me`) are guarded by /// the `require_admin` middleware. The data plane /// (`/api/v1/execute/{id}`, the user-route fallthrough, `/healthz`, /// `/version`) stays open — it's the public ingress for user scripts. #[allow(clippy::too_many_lines)] pub async fn build_app( pool: PgPool, auth: AuthDeps, master_key: MasterKey, ) -> anyhow::Result { let script_repo = Arc::new(PostgresScriptRepository::new(pool.clone())); let log_repo = Arc::new(PostgresExecutionLogRepository::new(pool.clone())); let log_sink: Arc = Arc::new(PostgresExecutionLogSink::new(pool.clone())); let route_repo = Arc::new(PostgresRouteRepository::new(pool.clone())); let apps_repo: Arc = Arc::new(PostgresAppRepository::new(pool.clone())); let groups_repo: Arc = Arc::new(PostgresGroupRepository::new(pool.clone())); let group_members_repo: Arc = Arc::new(PostgresGroupMembersRepository::new(pool.clone())); let domains_repo: Arc = Arc::new(PostgresAppDomainRepository::new(pool.clone())); // The Postgres app_members repo implements both `AppMembersRepository` // (CRUD over the table) and `AuthzRepo` (single-row membership lookup // for capability checks). Construct it once and clone the Arc into // both trait views — same allocation, two vtables. let members_concrete = Arc::new(PostgresAppMembersRepository::new(pool.clone())); let members: Arc = members_concrete.clone(); let authz: Arc = members_concrete; // Triggers framework storage. The outbox event emitter routes // KV mutations into the outbox; the dispatcher fans them out. let trigger_repo: Arc = Arc::new(PostgresTriggerRepo::new(pool.clone())); // PostgresOutboxRepo implements both `OutboxRepo` (the dispatcher // surface) and `OutboxWriter` (the orchestrator surface). Construct // the concrete Arc once, clone it into each trait view — same // allocation, two vtables (mirrors how `members_concrete` above is // used as both `AppMembersRepository` and `AuthzRepo`). let outbox_concrete = Arc::new(PostgresOutboxRepo::new(pool.clone())); let outbox_repo: Arc = outbox_concrete.clone(); let outbox_writer: Arc = outbox_concrete; let dl_repo: Arc = Arc::new(PostgresDeadLetterRepo::new(pool.clone())); let abandoned_repo: Arc = Arc::new(PostgresAbandonedRepo::new(pool.clone())); let trigger_config = TriggerConfig::from_env(); // SDK services bundle. v1.1.1 added KV + dead-letter; v1.1.2 added // the docs store; v1.1.3 adds the module source backing the Rhai // resolver. All bound services share the outbox-backed event // emitter so KV and docs mutations both fan out through the same // dispatcher. let kv_repo = Arc::new(PostgresKvRepo::new(pool.clone())); let docs_repo = Arc::new(PostgresDocsRepo::new(pool.clone())); let events: Arc = Arc::new(OutboxEventEmitter::new( trigger_repo.clone(), outbox_repo.clone(), )); let kv: Arc = Arc::new(KvServiceImpl::with_max_value_bytes( kv_repo.clone(), authz.clone(), events.clone(), picloud_manager_core::kv_service::kv_max_value_bytes_from_env(), )); let docs: Arc = Arc::new(DocsServiceImpl::with_max_value_bytes( docs_repo, authz.clone(), events.clone(), picloud_manager_core::docs_service::docs_max_value_bytes_from_env(), )); let dl_service: Arc = Arc::new(PostgresDeadLetterService::new( dl_repo.clone(), outbox_repo.clone(), authz.clone(), )); let modules: Arc = Arc::new( picloud_manager_core::PostgresModuleSource::new(pool.clone()), ); // v1.1.4 outbound HTTP. The reqwest client is built once here with // the SSRF deny-list resolver. `PICLOUD_HTTP_ALLOW_PRIVATE=true` // disables the deny-list entirely — dev/test only, so warn loudly. let http_config = HttpConfig::from_env(); if http_config.allow_private { tracing::warn!( "PICLOUD_HTTP_ALLOW_PRIVATE is set — the outbound-HTTP SSRF deny-list is DISABLED. \ Scripts can reach loopback/private/link-local addresses. Do NOT use in production." ); } let http: Arc = Arc::new(HttpServiceImpl::new(http_config, authz.clone())); // v1.1.5 filesystem-backed blob storage. Metadata lives in Postgres; // the bytes live on disk under `PICLOUD_FILES_ROOT` (default ./data). let files_config = FilesConfig::from_env(); let files_max_size = files_config.max_file_size_bytes; // Kept for the v1.1.6 orphan sweeper (cleans stale `*.tmp.*` files). let files_root = files_config.root.clone(); let files_repo = Arc::new(FsFilesRepo::new(pool.clone(), files_config)); let files: Arc = Arc::new(FilesServiceImpl::new( files_repo.clone(), authz.clone(), events.clone(), files_max_size, )); // v1.1.6 realtime: the in-process broadcaster is shared between the // publish path (PubsubServiceImpl fans out to SSE subscribers after // the durable outbox fan-out) and the SSE endpoint (subscribe side). // The topic registry + app-secrets repo back the subscriber-token // mint + SSE subscribe-authorization. let broadcaster_concrete = Arc::new(InProcessBroadcaster::from_env()); let broadcaster: Arc = broadcaster_concrete.clone(); let topic_repo: Arc = Arc::new(PostgresTopicRepo::new(pool.clone())); let app_secrets_repo = Arc::new(PostgresAppSecretsRepo::new( pool.clone(), master_key.clone(), )); // v1.1.7's migrate_plaintext_keys startup sweep was deleted in // v1.1.8 F1 alongside migration 0032, which drops the plaintext // column. Operators upgrading from v1.1.6 or earlier MUST apply // v1.1.7 first; the 0032 migration's guard refuses to run // otherwise. // v1.1.5 durable pub/sub, extended in v1.1.6 with the realtime // broadcast + subscriber-token mint. Publishes fan out to matching // pubsub triggers at publish time (one outbox row each, delivered by // the same dispatcher as every other async trigger) AND, best-effort, // to in-process SSE subscribers. let pubsub_repo = Arc::new(PostgresPubsubRepo::new(pool.clone())); let pubsub: Arc = Arc::new( PubsubServiceImpl::new(pubsub_repo, authz.clone()) .with_max_message_bytes( picloud_manager_core::pubsub_service::pubsub_max_message_bytes_from_env(), ) .with_realtime( broadcaster.clone(), topic_repo.clone(), app_secrets_repo.clone(), SubscriberTokenConfig::from_env(), ), ); // v1.1.7 encrypted per-app secrets. Values are AES-256-GCM-sealed // with the process master key before they touch Postgres; the repo // only ever sees ciphertext + nonce. The admin surface reuses the // same repo + master key (see `secrets_state` below). let secrets_config = SecretsConfig::from_env(); let secrets_max_value_bytes = secrets_config.max_value_bytes; let secrets_repo: Arc = Arc::new(PostgresSecretsRepo::new(pool.clone())); let secrets: Arc = Arc::new(SecretsServiceImpl::new( secrets_repo.clone(), authz.clone(), master_key.clone(), secrets_config, )); // v1.1.7 outbound email. Builds a lettre SMTP transport from // PICLOUD_SMTP_* env (disabled mode + warning if unconfigured). G5: in // dev mode with no relay, captures mail in memory instead of erroring; // `dev_email_sink` is `Some` then, and we mount the dev inspection // endpoint below. let (email_impl, dev_email_sink) = EmailServiceImpl::from_env_with_dev_capture(authz.clone()); let email: Arc = Arc::new(email_impl); // v1.1.8 data-plane user management. Wires Argon2id-hashed user // rows + SHA-256-hashed sliding-window sessions to the Rhai // `users::*` namespace and the admin /apps/{id}/users HTTP surface. let app_users_repo = Arc::new(PostgresAppUserRepository::new(pool.clone())); let app_user_sessions_repo = Arc::new(PostgresAppUserSessionRepository::new(pool.clone())); let app_user_verifications_repo = Arc::new(PostgresAppUserVerificationRepo::new(pool.clone())); let app_user_password_resets_repo = Arc::new(PostgresAppUserPasswordResetRepo::new(pool.clone())); let app_user_invitations_repo = Arc::new(PostgresAppUserInvitationRepo::new(pool.clone())); let app_user_roles_repo = Arc::new(PostgresAppUserRoleRepo::new(pool.clone())); let users: Arc = Arc::new(UsersServiceImpl::new( app_users_repo.clone(), app_user_sessions_repo.clone(), app_user_verifications_repo.clone(), app_user_password_resets_repo.clone(), app_user_invitations_repo.clone(), app_user_roles_repo.clone(), email.clone(), authz.clone(), events.clone(), UsersServiceConfig::from_env(), )); // v1.1.8 F3: build the realtime authority now that UsersService // is constructed; the Session arm of authorize_subscribe calls // users.verify_session_for_realtime. let realtime_authority: Arc = Arc::new(RealtimeAuthorityImpl::new( topic_repo.clone(), app_secrets_repo.clone(), users.clone(), )); // v1.1.9 durable per-app queues. Producers write to queue_messages // via QueueService; the dispatcher's queue arm (commit 6) consumes // claimed messages and fires the registered queue:receive trigger. let queue_repo: Arc = Arc::new( picloud_manager_core::queue_repo::PostgresQueueRepo::new(pool.clone()), ); let queue: Arc = Arc::new( picloud_manager_core::queue_service::QueueServiceImpl::with_max_payload_bytes( queue_repo.clone(), authz.clone(), picloud_manager_core::queue_service::queue_max_payload_bytes_from_env(), ), ); // Route table created early (before Services) so InvokeServiceImpl // can use it for path resolution. It's populated below from the // route_repo, then re-populated whenever the admin layer writes // routes. let route_table = Arc::new(RouteTable::new()); let initial = route_repo.list_all().await?; // Lenient: a single un-compilable stored route (e.g. one whose path // became reserved under stricter validation) is skipped-with-warning // inside compile_routes, never aborting boot. (H1) let compiled = compile_routes(&initial); route_table.replace_all(compiled); // v1.1.9 function composition. InvokeServiceImpl resolves targets // by Id, Name (via get_by_name), or Path (via the orchestrator's // RouteTable). invoke_async writes an OutboxSourceKind::Invoke row // that the dispatcher fires through the standard executor path. let invoke: Arc = Arc::new( picloud_manager_core::invoke_service::InvokeServiceImpl::new( script_repo.clone(), route_table.clone(), outbox_repo.clone(), ) // F-S-012: gate authenticated invoke() callers on AppInvoke so // an editor-role principal can't trigger admin-only worker // scripts in the same app. Anonymous callers skip the check // under script-as-gate semantics. .with_authz(authz.clone()), ); let services = Services::new( kv, docs, dl_service.clone(), events, modules, http, files, pubsub, secrets, email, users.clone(), queue, invoke, ); // v1.1.9: keep the invoke depth bound aligned with the dispatcher's // trigger-depth bound (same counter under the hood). let engine_limits = Limits { trigger_depth_max: trigger_config.max_trigger_depth, ..Limits::default() }; let engine = Arc::new(Engine::new(engine_limits, services)); // v1.1.9: install the back-reference the `invoke` SDK bridge needs // for synchronous re-entry. Weak so the Arc-cycle stays loose; // OnceLock-backed so it's idempotent. engine.set_self_weak(Arc::downgrade(&engine)); // Same shape for app domains (Host → app_id cache). let app_domain_table = Arc::new(AppDomainTable::new()); let initial_domains = domains_repo.list_all().await?; let compiled_domains: Vec<_> = initial_domains .iter() .filter_map(|d| { picloud_orchestrator_core::routing::parse_app_domain(&d.pattern) .ok() .map(|p| picloud_orchestrator_core::routing::CompiledAppDomain { app_id: d.app_id, pattern: p.pattern, shape_key: p.shape_key, }) }) .collect(); app_domain_table.replace(compiled_domains); let resolver = Arc::new(RepoResolver::new(script_repo.clone())); // Single global gate — overflow is rejected with 503 + Retry-After. // See `ExecutionGate` docs and `PICLOUD_MAX_CONCURRENT_EXECUTIONS`. let gate = Arc::new(ExecutionGate::from_env()); let executor = Arc::new(LocalExecutorClient::new(engine.clone(), gate.clone())); // Dispatcher — single tokio task that polls the outbox and routes // due rows to the executor. Shares the `ExecutionGate` with sync // HTTP per design notes §2 (one cap for everything). let dispatcher_script_repo: Arc = script_repo.clone(); let principals: Arc = Arc::new(AdminPrincipalResolver::new(auth.users.clone())); // The InboxRegistry is constructed once and shared between the // orchestrator (registers receivers, awaits) and the dispatcher // (delivers results). Two Arc views on the same allocation. let inbox_registry = Arc::new(InboxRegistry::new()); let inbox_resolver: Arc = inbox_registry.clone(); Dispatcher { outbox: outbox_repo.clone(), triggers: trigger_repo.clone(), scripts: dispatcher_script_repo, dead_letters: dl_repo.clone(), abandoned: abandoned_repo.clone(), principals, executor: executor.clone(), gate, log_sink: log_sink.clone(), inbox: inbox_resolver, queue: queue_repo.clone(), config: trigger_config, instance_id: format!("picloud-{}", std::process::id()), } .spawn(); let admin = AdminState { repo: script_repo.clone(), logs: log_repo, apps: apps_repo.clone(), authz: authz.clone(), validator: engine.clone(), sandbox_ceiling: SandboxCeiling::from_env(), }; let route_admin = RouteAdminState { routes: route_repo.clone(), scripts: script_repo.clone(), domains: domains_repo.clone(), table: route_table.clone(), authz: authz.clone(), }; let data_plane = DataPlaneState { executor, resolver, log_sink, app_domains: app_domain_table.clone(), routes: route_table.clone(), inbox: inbox_registry, outbox: outbox_writer, }; // Weekly retention sweepers for dead_letters + abandoned_executions. // Defaults: 30 days / 7 days (design notes §3 #9 + §4 retention). picloud_manager_core::spawn_dead_letter_gc( dl_repo.clone(), trigger_config.dead_letter_retention_days, ); picloud_manager_core::spawn_abandoned_gc( abandoned_repo.clone(), trigger_config.abandoned_retention_days, ); // v1.1.8 weekly sweep for app-user sessions + tokens. No // configurable retention — rows are pruned as soon as they're // either expired (by TTL) or consumed/revoked. picloud_manager_core::spawn_app_user_token_gc( app_user_sessions_repo.clone(), app_user_verifications_repo.clone(), app_user_password_resets_repo.clone(), app_user_invitations_repo.clone(), ); // v1.1.4: cron scheduler. Polls cron_trigger_details on a tick and // enqueues due triggers into the outbox; the dispatcher above // delivers them like any other async trigger. picloud_manager_core::spawn_cron_scheduler(pool.clone(), trigger_config.cron_tick_interval_ms); // v1.1.6: GC empty realtime broadcast channels (one-shot subscribers) // and sweep orphaned `*.tmp.*` blobs left by crashed file writes. spawn_realtime_gc(broadcaster_concrete, DEFAULT_GC_INTERVAL_SECS); picloud_manager_core::spawn_files_orphan_sweep(files_root); let triggers_state = TriggersState { triggers: trigger_repo.clone(), apps: apps_repo.clone(), authz: authz.clone(), scripts: script_repo.clone(), config: trigger_config, master_key: master_key.clone(), }; // Declarative reconcile engine (pic plan / apply). Trait-object repos // for the read/diff path; shares the same handles as the CRUD routers. let apply_service = ApplyService { pool: pool.clone(), scripts: script_repo.clone(), routes: route_repo.clone(), triggers: trigger_repo.clone(), secrets: secrets_repo.clone(), apps: apps_repo.clone(), domains: domains_repo.clone(), authz: authz.clone(), validator: engine.clone(), sandbox_ceiling: SandboxCeiling::from_env(), trigger_config, route_table: route_table.clone(), master_key: master_key.clone(), }; // v1.1.9: keep a clone for the queues-api state (built later). let trigger_repo_for_queues = trigger_repo.clone(); // v1.1.7 public inbound-email receiver. Outside the admin auth layer // (the URL + per-trigger HMAC secret are the security boundary). let email_inbound_state = EmailInboundState { triggers: trigger_repo, outbox: outbox_repo.clone(), master_key: master_key.clone(), nonce_dedup: Arc::new(InboundNonceDedup::new()), bad_sig_limiter: Arc::new( picloud_manager_core::email_inbound_api::BadSignatureLimiter::new(), ), }; let dead_letters_state = DeadLettersState { repo: dl_repo, service: dl_service, apps: apps_repo.clone(), authz: authz.clone(), }; let files_admin_state = FilesAdminState { files: files_repo, apps: apps_repo.clone(), authz: authz.clone(), }; let topics_state = TopicsState { topics: topic_repo, apps: apps_repo.clone(), authz: authz.clone(), broadcaster: broadcaster.clone(), }; let secrets_state = SecretsState { repo: secrets_repo, apps: apps_repo.clone(), authz: authz.clone(), master_key, max_value_bytes: secrets_max_value_bytes, }; let apps_state = AppsState { apps: apps_repo.clone(), domains: domains_repo, routes: route_repo, domain_table: app_domain_table.clone(), authz: authz.clone(), groups: groups_repo.clone(), }; // Audit 2026-06-11 H-B1 — login DoS defenses. PICLOUD_LOGIN_ARGON2_PARALLELISM // overrides the per-process Argon2-during-login concurrency cap. let argon2_login_parallelism = std::env::var("PICLOUD_LOGIN_ARGON2_PARALLELISM") .ok() .and_then(|s| s.parse::().ok()) .filter(|n| *n > 0) .unwrap_or(2); // Audit 2026-06-11 (PrincipalCache revocation-lag) — one shared // cache instance threaded into every state that performs auth or // revocation, so an evict on the revocation side is visible to the // middleware's resolve side. A per-state cache would let a revoked // principal keep authenticating from the middleware's own copy. let principal_cache = Arc::new(picloud_manager_core::auth_middleware::PrincipalCache::new()); let auth_state = AuthState { users: auth.users.clone(), sessions: auth.sessions.clone(), keys: auth.keys.clone(), ttl: auth.ttl, principal_cache: principal_cache.clone(), login_rate_limiter: Arc::new( picloud_manager_core::login_rate_limit::LoginRateLimiter::new(), ), argon2_login_semaphore: Arc::new(tokio::sync::Semaphore::new(argon2_login_parallelism)), }; let admins_state = AdminsState { users: auth.users.clone(), sessions: auth.sessions, keys: auth.keys.clone(), authz: authz.clone(), principal_cache: principal_cache.clone(), }; let app_members_state = AppMembersState { apps: apps_state.apps.clone(), users: auth.users.clone(), members, authz: authz.clone(), }; let groups_state = GroupsState { groups: groups_repo.clone(), group_members: group_members_repo.clone(), apps: apps_state.apps.clone(), users: auth.users.clone(), authz: authz.clone(), }; let app_users_admin_state = picloud_manager_core::AppUsersState { apps: apps_state.apps.clone(), authz: authz.clone(), users: users.clone(), }; let api_keys_state = ApiKeysState { keys: auth.keys, principal_cache: principal_cache.clone(), }; // /admin/auth/login + /logout are unguarded by design (login is how // you get in). /admin/auth/me applies the middleware internally so // the same Router::with_state machinery composes cleanly. Everything // else under /admin gets the require_authenticated layer; capability // checks live in each handler (after the resource is loaded so the // capability binds to the resource's actual app_id). let mut guarded_admin = Router::new() .merge(admin_router(admin)) .merge(route_admin_router(route_admin)) .merge(admins_router(admins_state)) .merge(apps_router(apps_state)) .merge(app_members_router(app_members_state)) .merge(groups_router(groups_state)) .merge(picloud_manager_core::app_users_router( app_users_admin_state, )) .merge(api_keys_router(api_keys_state)) .merge(triggers_router(triggers_state)) .merge(apply_router(apply_service)) .merge(picloud_manager_core::queues_api::queues_router( picloud_manager_core::queues_api::QueuesState { queues: queue_repo.clone(), triggers: trigger_repo_for_queues.clone(), apps: apps_repo.clone(), authz: authz.clone(), scripts: script_repo.clone(), }, )) .merge(files_admin_router(files_admin_state)) .merge(kv_admin_router(KvAdminState { kv: kv_repo.clone(), apps: apps_repo.clone(), authz: authz.clone(), })) .merge(topics_router(topics_state)) .merge(secrets_router(secrets_state)) .merge(dead_letters_router(dead_letters_state)); // G5: dev-only mail inspection — mounted exactly when the email // service is capturing in memory (dev mode + no relay). Same // `require_authenticated` layer as the rest of /admin applies below. if let Some(sink) = dev_email_sink { guarded_admin = guarded_admin.merge(dev_emails_router(DevEmailState { sink })); } let guarded_admin = guarded_admin.layer(from_fn_with_state( auth_state.clone(), require_authenticated, )); // Silence "unused import" lint on `apps_api` — we re-export via the // facade above; the bare module path is retained so it's discoverable. let _ = apps_api::AppsState::clone; // Opportunistic principal extraction on every data-plane request. // Always inserts `Extension>`: Some for authed // ingress (bearer / cookie), None otherwise. Handlers depend on // this layer being applied — scoped to the data-plane routers so // the admin path (which uses `require_authenticated`) doesn't // double-resolve the same token. let data_plane_routed = data_plane_router(data_plane.clone()).layer(from_fn_with_state( auth_state.clone(), attach_principal_if_present, )); let user_routes = user_routes_router(data_plane).layer(from_fn_with_state( auth_state.clone(), attach_principal_if_present, )); let api_v1 = Router::new() .nest("/admin", auth_router(auth_state)) .nest("/admin", guarded_admin) .merge(email_inbound_router(email_inbound_state)) .merge(data_plane_routed); // v1.1.6 SSE realtime surface, merged at the root (deliberately NOT // under /api/ — realtime is its own versioning surface). Public auth // is per-topic; no principal middleware (token verification is the // gate, handled inside the authority). let realtime = realtime_router(RealtimeState::new( app_domain_table, broadcaster, realtime_authority, )); Ok(Router::new() .route("/healthz", get(healthz)) .route("/version", get(version)) .nest(&format!("/api/v{API_VERSION}"), api_v1) .merge(realtime) .merge(user_routes) .layer(TraceLayer::new_for_http())) } /// Default Postgres pool size. Scaled to match the execution gate /// default (32 concurrent script executions) so a hot pool doesn't /// starve the data plane. Override with `PICLOUD_DB_MAX_CONNECTIONS`. /// Sizing relationship: each script run does multiple sequential DB /// calls (script resolve, SDK calls, log sink, outbox emit) plus the /// dispatcher tick / cron tick / GC sweeps / auth middleware all draw /// from the same pool, so this is a floor not a ceiling. pub const DEFAULT_DB_MAX_CONNECTIONS: u32 = 32; /// Open a Postgres pool with the binary's standard timeout settings. /// Exposed so tests reach for the same configuration when needed. pub async fn init_db(url: &str) -> anyhow::Result { let max_connections = std::env::var("PICLOUD_DB_MAX_CONNECTIONS") .ok() .and_then(|v| v.trim().parse::().ok()) .filter(|n| *n > 0) .unwrap_or(DEFAULT_DB_MAX_CONNECTIONS); let pool = PgPoolOptions::new() .max_connections(max_connections) .acquire_timeout(Duration::from_secs(5)) .connect(url) .await?; Ok(pool) } async fn healthz() -> &'static str { "ok" } /// Snapshot of every compatibility-surface version this process speaks /// plus the operator-configured public base URL (so the dashboard can /// render full URLs for user routes). /// /// Source of truth: `shared::version`, the embedded migrations, and /// the `PICLOUD_PUBLIC_BASE_URL` env var (default /// `http://localhost:8000`). async fn version() -> Json { let public_base_url = std::env::var("PICLOUD_PUBLIC_BASE_URL") .unwrap_or_else(|_| "http://localhost:8000".to_string()); Json(serde_json::json!({ "product": PRODUCT_VERSION, "sdk": SDK_VERSION, "api": API_VERSION, "schema": migrations::latest_version(), "wire": WIRE_VERSION, "public_base_url": public_base_url, })) } // F-Q-011: the PostgresScriptRepoHandle newtype lived here so the // resolver could take an owned `impl ScriptRepository` from the shared // Arc. The blanket // `impl ScriptRepository for Arc` in // manager-core::repo lets us pass `script_repo.clone()` directly, // eliminating 70 lines of hand-delegated boilerplate that had to be // rewritten every time the trait gained a method.