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

systemprompt_api/services/gateway/
policy.rs
