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 systemprompt_ai::repository::AiGatewayPolicyRepository;
15
16pub use systemprompt_ai::{GatewayPolicySpec, QuotaWindow, SafetyConfig};
17
18const CACHE_TTL: Duration = Duration::from_secs(60);
19
20#[derive(Clone)]
21pub struct PolicyResolver {
22    repo: Arc<AiGatewayPolicyRepository>,
23    cache: Arc<RwLock<Option<CachedEntry>>>,
24}
25
26impl std::fmt::Debug for PolicyResolver {
27    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
28        f.debug_struct("PolicyResolver").finish()
29    }
30}
31
32#[derive(Clone)]
33struct CachedEntry {
34    spec: GatewayPolicySpec,
35    fetched_at: Instant,
36}
37
38impl PolicyResolver {
39    pub fn from_repository(repo: AiGatewayPolicyRepository) -> Self {
40        Self {
41            repo: Arc::new(repo),
42            cache: Arc::new(RwLock::new(None)),
43        }
44    }
45
46    pub async fn resolve(&self) -> GatewayPolicySpec {
47        if let Ok(cache) = self.cache.read()
48            && let Some(entry) = cache.as_ref()
49            && entry.fetched_at.elapsed() < CACHE_TTL
50        {
51            return entry.spec.clone();
52        }
53
54        let rows = match self.repo.list_for_global().await {
55            Ok(r) => r,
56            Err(e) => {
57                tracing::warn!(error = %e, "policy resolve DB error — falling back to permissive");
58                return GatewayPolicySpec::permissive();
59            },
60        };
61
62        let spec = merge(rows);
63        if let Ok(mut cache) = self.cache.write() {
64            *cache = Some(CachedEntry {
65                spec: spec.clone(),
66                fetched_at: Instant::now(),
67            });
68        }
69        spec
70    }
71}
72
73fn merge(rows: Vec<systemprompt_ai::GatewayPolicyRow>) -> GatewayPolicySpec {
74    let mut merged = GatewayPolicySpec::permissive();
75    for row in rows {
76        let Ok(spec) = serde_json::from_value::<GatewayPolicySpec>(row.spec) else {
77            tracing::warn!(policy_id = %row.id, name = %row.name, "policy spec JSON malformed — skipped");
78            continue;
79        };
80        if !spec.quota_windows.is_empty() {
81            merged.quota_windows = spec.quota_windows;
82        }
83        if !spec.safety.scanners.is_empty()
84            || !spec.safety.block_categories.is_empty()
85            || !spec.safety.block_response_categories.is_empty()
86        {
87            merged.safety = spec.safety;
88        }
89    }
90    merged
91}
systemprompt_api/services/gateway/policy.rs

systemprompt_api/services/gateway/
policy.rs
