systemprompt_api/services/gateway/
policy.rs

1//! Resolution and caching of the effective gateway policy.
2//!
3//! [`PolicyResolver`] loads the global policy rows in ascending
4//! `(priority, name)` order and merges them into a single
5//! [`GatewayPolicySpec`] — each non-empty section overrides the previous, so
6//! the highest-priority row wins. The result is cached for a short TTL; a DB
7//! error or
8//! a malformed spec degrades to a permissive policy rather than failing the
9//! request.
10//!
11//! Copyright (c) systemprompt.io — Business Source License 1.1.
12//! See <https://systemprompt.io> for licensing details.
13
14use std::sync::{Arc, RwLock};
15use std::time::{Duration, Instant};
16
17use systemprompt_ai::repository::AiGatewayPolicyRepository;
18
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 from_repository(repo: AiGatewayPolicyRepository) -> Self {
43        Self {
44            repo: Arc::new(repo),
45            cache: Arc::new(RwLock::new(None)),
46        }
47    }
48
49    pub async fn resolve(&self) -> GatewayPolicySpec {
50        if let Ok(cache) = self.cache.read()
51            && let Some(entry) = cache.as_ref()
52            && entry.fetched_at.elapsed() < CACHE_TTL
53        {
54            return entry.spec.clone();
55        }
56
57        let rows = match self.repo.list_for_global().await {
58            Ok(r) => r,
59            Err(e) => {
60                tracing::warn!(error = %e, "policy resolve DB error — falling back to permissive");
61                return GatewayPolicySpec::permissive();
62            },
63        };
64
65        let spec = merge(rows);
66        if let Ok(mut cache) = self.cache.write() {
67            *cache = Some(CachedEntry {
68                spec: spec.clone(),
69                fetched_at: Instant::now(),
70            });
71        }
72        spec
73    }
74}
75
76fn merge(rows: Vec<systemprompt_ai::GatewayPolicyRow>) -> GatewayPolicySpec {
77    let mut merged = GatewayPolicySpec::permissive();
78    for row in rows {
79        let Ok(spec) = serde_json::from_value::<GatewayPolicySpec>(row.spec) else {
80            tracing::warn!(policy_id = %row.id, name = %row.name, "policy spec JSON malformed — skipped");
81            continue;
82        };
83        if !spec.quota_windows.is_empty() {
84            merged.quota_windows = spec.quota_windows;
85        }
86        if !spec.safety.scanners.is_empty()
87            || !spec.safety.block_categories.is_empty()
88            || !spec.safety.block_response_categories.is_empty()
89        {
90            merged.safety = spec.safety;
91        }
92    }
93    merged
94}
systemprompt_api/services/gateway/policy.rs

systemprompt_api/services/gateway/
policy.rs
