systemprompt_analytics/snapshots/
retention_owners.rs

1//! Global raw expiry requires atomic compaction of every initialized
2//! organizational scope.
3//!
4//! Copyright (c) systemprompt.io — Business Source License 1.1.
5//! See <https://systemprompt.io> for licensing details.
6
7use super::{FeedbackSnapshotsRepository, RetentionSummary, invalid};
8use chrono::{DateTime, Utc};
9use systemprompt_identifiers::UserId;
10
11impl FeedbackSnapshotsRepository {
12    pub async fn compact_all_in(
13        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
14        now: DateTime<Utc>,
15    ) -> crate::Result<RetentionSummary> {
16        if now > Utc::now() {
17            return Err(invalid("Retention clock cannot be in the future"));
18        }
19        sqlx::query!("SELECT public.prepare_reporting_privacy() AS locked")
20            .fetch_one(&mut **tx)
21            .await?;
22        sqlx::query!("LOCK TABLE analytics_ingestion_producers,analytics_fact_checkpoints,analytics_fact_backfills,analytics_fact_changes,analytics_fact_consumers,analytics_fact_deltas IN EXCLUSIVE MODE").execute(&mut **tx).await?;
23        let owners=sqlx::query_scalar!("SELECT owner_id FROM analytics_fact_checkpoints ORDER BY owner_id LIMIT 10001 FOR UPDATE").fetch_all(&mut **tx).await?;
24        if owners.len() > 10000 {
25            return Err(invalid("Retention organization bound exceeded"));
26        }
27        let mut summary = RetentionSummary {
28            organizations: 0,
29            removed_facts: 0,
30            removed_daily: 0,
31        };
32        for owner in owners {
33            let result = Self::compact_in(tx, &UserId::new(owner), now).await?;
34            summary.organizations += 1;
35            summary.removed_facts = summary
36                .removed_facts
37                .checked_add(result.removed_facts)
38                .ok_or_else(|| invalid("Retention count overflow"))?;
39            summary.removed_daily = summary
40                .removed_daily
41                .checked_add(result.removed_daily)
42                .ok_or_else(|| invalid("Retention count overflow"))?;
43        }
44        sqlx::query!(
45            "SELECT public.finish_reporting_privacy($1) AS processed",
46            now - chrono::Duration::days(90)
47        )
48        .fetch_one(&mut **tx)
49        .await?;
50        Ok(summary)
51    }
52}
systemprompt_analytics/snapshots/retention_owners.rs

systemprompt_analytics/snapshots/
retention_owners.rs
