systemprompt_analytics/repository/
ownership.rs

1//! [`OwnerReassignment`] for the analytics domain: moves every engagement
2//! event from one user to another and rewrites the source id in each
3//! fingerprint's associated-user list, in a single transaction.
4//!
5//! Copyright (c) systemprompt.io — Business Source License 1.1.
6//! See <https://systemprompt.io> for licensing details.
7
8use async_trait::async_trait;
9use sqlx::PgPool;
10use std::sync::Arc;
11use systemprompt_database::DbPool;
12use systemprompt_identifiers::UserId;
13use systemprompt_traits::{OwnerReassignment, ReassignedRows, RepositoryError};
14
15
16#[derive(Debug, Clone)]
17pub struct AnalyticsOwnerReassignment {
18    write_pool: Arc<PgPool>,
19}
20
21impl AnalyticsOwnerReassignment {
22    pub fn new(db: &DbPool) -> Self {
23        let write_pool = db.write_pool();
24        Self { write_pool }
25    }
26}
27
28#[async_trait]
29impl OwnerReassignment for AnalyticsOwnerReassignment {
30    fn domain(&self) -> &'static str {
31        "analytics"
32    }
33
34    async fn reassign_owner(
35        &self,
36        from: &UserId,
37        to: &UserId,
38    ) -> Result<ReassignedRows, RepositoryError> {
39        let mut tx = self
40            .write_pool
41            .begin()
42            .await
43            .map_err(RepositoryError::database)?;
44
45        let engagement = sqlx::query!(
46            "UPDATE engagement_events SET user_id = $2 WHERE user_id = $1",
47            from.as_str(),
48            to.as_str()
49        )
50        .execute(&mut *tx)
51        .await
52        .map_err(RepositoryError::database)?
53        .rows_affected();
54
55        let fingerprints = sqlx::query!(
56            "UPDATE fingerprint_reputation SET associated_user_ids = \
57             array_replace(associated_user_ids, $1, $2) WHERE $1 = ANY(associated_user_ids)",
58            from.as_str(),
59            to.as_str()
60        )
61        .execute(&mut *tx)
62        .await
63        .map_err(RepositoryError::database)?
64        .rows_affected();
65
66        tx.commit().await.map_err(RepositoryError::database)?;
67
68        Ok(ReassignedRows {
69            tables: vec![
70                ("engagement_events", engagement),
71                ("fingerprint_reputation", fingerprints),
72            ],
73        })
74    }
75}
systemprompt_analytics/repository/ownership.rs

systemprompt_analytics/repository/
ownership.rs
