systemprompt_analytics/feedback/
repository.rs

1//! Transactional independent change admission and owner-scoped fact reads.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use super::{ChangeReceipt, StoredFact, validation};
7use crate::Result;
8use sqlx::PgPool;
9use systemprompt_identifiers::UserId;
10use systemprompt_models::feedback::ContentDigest;
11use systemprompt_models::feedback::analytics::{AnalyticsChange, AnalyticsFactKey};
12
13#[derive(Debug, Clone)]
14pub struct FeedbackFactsRepository {
15    pub(super) pool: PgPool,
16}
17
18impl FeedbackFactsRepository {
19    pub const fn new(pool: PgPool) -> Self {
20        Self { pool }
21    }
22
23    pub async fn submit(&self, owner: &UserId, change: &AnalyticsChange) -> Result<ChangeReceipt> {
24        let mut tx = self.pool.begin().await?;
25        let receipt = Self::submit_in(&mut tx, owner, change).await?;
26        tx.commit().await?;
27        Ok(receipt)
28    }
29
30    pub async fn submit_in(
31        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
32        owner: &UserId,
33        change: &AnalyticsChange,
34    ) -> Result<ChangeReceipt> {
35        validation::validate(change)?;
36        sqlx::query!(
37            "INSERT INTO analytics_fact_checkpoints(owner_id) VALUES($1) ON CONFLICT DO NOTHING",
38            owner.as_str()
39        )
40        .execute(&mut **tx)
41        .await?;
42        sqlx::query!(
43            "SELECT generation FROM analytics_fact_checkpoints WHERE owner_id=$1 FOR UPDATE",
44            owner.as_str()
45        )
46        .fetch_one(&mut **tx)
47        .await?;
48        let kind = validation::kind(change.key.kind);
49        let revision = i64::try_from(change.revision).map_err(|_error| validation::invalid())?;
50        let digest = ContentDigest::of(&serde_json::to_vec(&(
51            &change.key,
52            change.revision,
53            change.occurred_at,
54            &change.operation,
55        ))?);
56        let payload = serde_json::to_value(change)?;
57        sqlx::query!("INSERT INTO analytics_fact_changes(owner_id,change_id,fact_kind,source,fact_id,revision,occurred_at,recorded_at,payload,payload_digest) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT DO NOTHING", owner.as_str(), change.change_id.as_str(), kind, &change.key.source, change.key.id.as_str(), revision, change.occurred_at, change.recorded_at, payload, digest.as_str()).execute(&mut **tx).await?;
58        let rows = sqlx::query!("SELECT change_id,state,payload_digest FROM analytics_fact_changes WHERE owner_id=$1 AND (change_id=$2 OR (fact_kind=$3 AND source=$4 AND fact_id=$5 AND revision=$6))", owner.as_str(), change.change_id.as_str(), kind, &change.key.source, change.key.id.as_str(), revision).fetch_all(&mut **tx).await?;
59        if rows.len() != 1 || rows[0].payload_digest != digest.as_str() {
60            return Err(validation::invalid());
61        }
62        Ok(ChangeReceipt {
63            change_id: systemprompt_identifiers::AnalyticsChangeId::new(rows[0].change_id.clone()),
64            state: validation::parse_state(&rows[0].state)?,
65        })
66    }
67
68    pub async fn get_fact(
69        &self,
70        owner: &UserId,
71        key: &AnalyticsFactKey,
72    ) -> Result<Option<StoredFact>> {
73        let kind = validation::kind(key.kind);
74        let row = sqlx::query!("SELECT revision,occurred_at,fact,generation FROM analytics_normalized_facts WHERE owner_id=$1 AND fact_kind=$2 AND source=$3 AND fact_id=$4", owner.as_str(), kind, &key.source, key.id.as_str()).fetch_optional(&self.pool).await?;
75        row.map(|row| {
76            Ok(StoredFact {
77                key: key.clone(),
78                revision: row.revision,
79                occurred_at: row.occurred_at,
80                fact: row.fact.map(serde_json::from_value).transpose()?,
81                generation: row.generation,
82            })
83        })
84        .transpose()
85    }
86    pub async fn list_facts(
87        &self,
88        owner: &UserId,
89        after: Option<&AnalyticsFactKey>,
90        limit: u32,
91    ) -> Result<Vec<StoredFact>> {
92        if !(1..=256).contains(&limit) {
93            return Err(validation::invalid());
94        }
95        let after_kind = after.map(|key| validation::kind(key.kind));
96        let after_source = after.map(|key| key.source.as_str());
97        let after_id = after.map(|key| key.id.as_str());
98        let limit = i64::from(limit);
99        let rows = sqlx::query!("SELECT fact_kind,source,fact_id,revision,occurred_at,fact,generation FROM analytics_normalized_facts WHERE owner_id=$1 AND ($2::text IS NULL OR (fact_kind,source,fact_id)>($2,$3,$4)) ORDER BY fact_kind,source,fact_id LIMIT $5", owner.as_str(), after_kind, after_source, after_id, limit).fetch_all(&self.pool).await?;
100        rows.into_iter()
101            .map(|row| {
102                Ok(StoredFact {
103                    key: AnalyticsFactKey {
104                        kind: validation::parse_kind(&row.fact_kind)?,
105                        source: row.source,
106                        id: systemprompt_identifiers::AnalyticsFactId::new(row.fact_id),
107                    },
108                    revision: row.revision,
109                    occurred_at: row.occurred_at,
110                    fact: row.fact.map(serde_json::from_value).transpose()?,
111                    generation: row.generation,
112                })
113            })
114            .collect()
115    }
116
117    pub async fn change_status(
118        &self,
119        owner: &UserId,
120        id: &systemprompt_identifiers::AnalyticsChangeId,
121    ) -> Result<Option<super::ChangeDiagnostic>> {
122        let row = sqlx::query!("SELECT state,attempts,lease_until,last_error FROM analytics_fact_changes WHERE owner_id=$1 AND change_id=$2", owner.as_str(), id.as_str()).fetch_optional(&self.pool).await?;
123        row.map(|row| {
124            Ok(super::ChangeDiagnostic {
125                change_id: id.clone(),
126                state: validation::parse_state(&row.state)?,
127                attempts: row.attempts,
128                lease_until: row.lease_until,
129                last_error: row.last_error,
130            })
131        })
132        .transpose()
133    }
134}
systemprompt_analytics/feedback/repository.rs

systemprompt_analytics/feedback/
repository.rs
