systemprompt_api/services/gateway/
policy.rs

1//! Resolution and caching of the effective gateway policy.
2//!
3//! [`PolicyResolver`] loads the global policy rows, merges them into a single
4//! [`GatewayPolicySpec`], and caches the result for a short TTL; a DB error or
5//! a malformed spec degrades to a permissive policy rather than failing the
6//! request.
7
8use std::sync::{Arc, RwLock};
9use std::time::{Duration, Instant};
10
11use anyhow::Result;
12use systemprompt_ai::repository::AiGatewayPolicyRepository;
13use systemprompt_database::DbPool;
14
15// The gateway-policy spec types are owned by `systemprompt-ai` so the
16// version-controlled `services/gateway/policies.yaml` and the persisted
17// `ai_gateway_policies.spec` column share one schema. Re-exported here so
18// existing `super::policy::{...}` call sites are unaffected.
19pub use systemprompt_ai::{GatewayPolicySpec, QuotaWindow, SafetyConfig};
20
21const CACHE_TTL: Duration = Duration::from_secs(60);
22
23#[derive(Clone)]
24pub struct PolicyResolver {
25    repo: Arc<AiGatewayPolicyRepository>,
26    cache: Arc<RwLock<Option<CachedEntry>>>,
27}
28
29impl std::fmt::Debug for PolicyResolver {
30    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
31        f.debug_struct("PolicyResolver").finish()
32    }
33}
34
35#[derive(Clone)]
36struct CachedEntry {
37    spec: GatewayPolicySpec,
38    fetched_at: Instant,
39}
40
41impl PolicyResolver {
42    pub fn new(db: &DbPool) -> Result<Self> {
43        Ok(Self {
44            repo: Arc::new(
45                AiGatewayPolicyRepository::new(db)
46                    .map_err(|e| anyhow::anyhow!("policy repo init: {e}"))?,
47            ),
48            cache: Arc::new(RwLock::new(None)),
49        })
50    }
51
52    pub async fn resolve(&self) -> GatewayPolicySpec {
53        if let Ok(cache) = self.cache.read()
54            && let Some(entry) = cache.as_ref()
55            && entry.fetched_at.elapsed() < CACHE_TTL
56        {
57            return entry.spec.clone();
58        }
59
60        let rows = match self.repo.list_for_global().await {
61            Ok(r) => r,
62            Err(e) => {
63                tracing::warn!(error = %e, "policy resolve DB error — falling back to permissive");
64                return GatewayPolicySpec::permissive();
65            },
66        };
67
68        let spec = merge(rows);
69        if let Ok(mut cache) = self.cache.write() {
70            *cache = Some(CachedEntry {
71                spec: spec.clone(),
72                fetched_at: Instant::now(),
73            });
74        }
75        spec
76    }
77}
78
79fn merge(rows: Vec<systemprompt_ai::GatewayPolicyRow>) -> GatewayPolicySpec {
80    let mut merged = GatewayPolicySpec::permissive();
81    for row in rows {
82        let Ok(spec) = serde_json::from_value::<GatewayPolicySpec>(row.spec) else {
83            tracing::warn!(policy_id = %row.id, name = %row.name, "policy spec JSON malformed — skipped");
84            continue;
85        };
86        if !spec.quota_windows.is_empty() {
87            merged.quota_windows = spec.quota_windows;
88        }
89        if !spec.safety.scanners.is_empty() || !spec.safety.block_categories.is_empty() {
90            merged.safety = spec.safety;
91        }
92    }
93    merged
94}
systemprompt_api/services/gateway/policy.rs

systemprompt_api/services/gateway/
policy.rs
