systemprompt_sync/jobs/
access_control_sync.rs

1//! Bootstrap job that projects the access-control baseline into the database:
2//! the YAML grants at `access-control/config.yaml` plus the marketplace-access
3//! grants declared in the services config.
4//!
5//! Mirrors [`super::ContentSyncJob`] but with a fixed direction
6//! (config → DB). Disabled by default; operators wire it in via
7//! `scheduler_config.bootstrap_jobs` so it runs once at startup.
8//!
9//! Copyright (c) systemprompt.io — Business Source License 1.1.
10//! See <https://systemprompt.io> for licensing details.
11
12use std::path::PathBuf;
13use std::sync::Arc;
14
15use async_trait::async_trait;
16use systemprompt_database::{Database, DbPool};
17use systemprompt_models::AppPaths;
18use systemprompt_security::authz::{AccessControlIngestionService, IngestOptions};
19use systemprompt_traits::{Job, JobContext, JobResult, ProviderError, ProviderResult};
20
21use crate::local::AccessControlLocalSync;
22
23const DEFAULT_YAML_RELATIVE: &str = "access-control/config.yaml";
24
25#[derive(Debug, Clone, Copy)]
26pub struct AccessControlSyncJob;
27
28#[async_trait]
29impl Job for AccessControlSyncJob {
30    fn name(&self) -> &'static str {
31        "access_control_sync"
32    }
33
34    fn description(&self) -> &'static str {
35        "Project services/access-control YAML into access_control_rules"
36    }
37
38    fn schedule(&self) -> &'static str {
39        ""
40    }
41
42    fn tags(&self) -> Vec<&'static str> {
43        vec!["access-control", "sync", "bootstrap"]
44    }
45
46    fn enabled(&self) -> bool {
47        false
48    }
49
50    async fn execute(&self, ctx: &JobContext) -> ProviderResult<JobResult> {
51        let start = std::time::Instant::now();
52
53        let db_pool: &DbPool = ctx.db_pool::<DbPool>().ok_or_else(|| {
54            ProviderError::Configuration("DbPool not available in job context".into())
55        })?;
56
57        let paths = ctx
58            .app_paths::<Arc<AppPaths>>()
59            .ok_or_else(|| {
60                ProviderError::Configuration("AppPaths not available in job context".into())
61            })?
62            .as_ref();
63
64        let yaml_path = resolve_yaml_path(ctx, paths.system().services());
65        let override_existing = bool_param(ctx, "override_existing", true);
66        let delete_orphans = bool_param(ctx, "delete_orphans", true);
67
68        tracing::info!(
69            yaml_path = %yaml_path.display(),
70            override_existing,
71            delete_orphans,
72            "access_control_sync job started",
73        );
74
75        let sync = AccessControlLocalSync::new(Arc::<Database>::clone(db_pool), yaml_path);
76        let acl = sync
77            .sync_to_db(override_existing, delete_orphans)
78            .await
79            .map_err(|e| ProviderError::RenderFailed(e.to_string()))?;
80
81        let services = systemprompt_loader::ConfigLoader::load().map_err(|e| {
82            ProviderError::Configuration(format!("Failed to load services config: {e}"))
83        })?;
84        let svc = AccessControlIngestionService::new(db_pool)
85            .map_err(|e| ProviderError::Configuration(e.to_string()))?;
86        let options = IngestOptions {
87            override_existing,
88            delete_orphans,
89        };
90        let mkt = svc
91            .ingest_marketplace_access(&services.marketplaces, options)
92            .await
93            .map_err(|e| ProviderError::RenderFailed(e.to_string()))?;
94        let slack = svc
95            .ingest_slack_apps(&services.slack_apps, options)
96            .await
97            .map_err(|e| ProviderError::RenderFailed(e.to_string()))?;
98        let teams = svc
99            .ingest_teams_apps(&services.teams_apps, options)
100            .await
101            .map_err(|e| ProviderError::RenderFailed(e.to_string()))?;
102
103        let items_synced = acl.items_synced
104            + mkt.inserted
105            + mkt.updated
106            + slack.inserted
107            + slack.updated
108            + teams.inserted
109            + teams.updated;
110        let items_skipped = acl.items_skipped + mkt.skipped + slack.skipped + teams.skipped;
111        let items_deleted = acl.items_deleted + mkt.deleted + slack.deleted + teams.deleted;
112
113        let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
114        tracing::info!(
115            items_synced,
116            items_skipped,
117            items_deleted,
118            duration_ms,
119            "access_control_sync job completed",
120        );
121
122        Ok(JobResult::success()
123            .with_stats(items_synced as u64, acl.errors.len() as u64)
124            .with_duration(duration_ms))
125    }
126}
127
128fn resolve_yaml_path(ctx: &JobContext, services_path: &std::path::Path) -> PathBuf {
129    ctx.parameters().get("yaml_path").map_or_else(
130        || services_path.join(DEFAULT_YAML_RELATIVE),
131        |raw| {
132            let p = std::path::Path::new(raw);
133            if p.is_absolute() {
134                p.to_path_buf()
135            } else {
136                services_path.join(p)
137            }
138        },
139    )
140}
141
142fn bool_param(ctx: &JobContext, key: &str, default: bool) -> bool {
143    ctx.parameters().get(key).map_or(default, |v| {
144        matches!(v.as_str(), "true" | "1" | "yes" | "TRUE" | "True")
145    })
146}
147
148systemprompt_provider_contracts::submit_job!(&AccessControlSyncJob);
systemprompt_sync/jobs/access_control_sync.rs

systemprompt_sync/jobs/
access_control_sync.rs
