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

systemprompt_api/services/gateway/
policy.rs
