systemprompt_users/repository/user/
merge.rs

1//! Anonymous-to-identified user merge operations.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use sqlx::{Acquire, Postgres, Transaction};
7use systemprompt_identifiers::{ContextId, SessionId, UserId};
8
9use crate::error::Result;
10use crate::repository::UserRepository;
11
12const MERGE_TOOL_NAME: &str = "users.merge";
13const MERGE_POLICY: &str = "account_merge";
14
15#[derive(Debug, Clone, Copy)]
16pub struct MergeResult {
17    pub sessions: u64,
18    pub tasks: u64,
19    pub total_rows: u64,
20}
21
22pub const MERGE_EXCLUDED_SECURITY_TABLES: &[&str] = &[
23    "oauth_auth_codes",
24    "oauth_refresh_tokens",
25    "oauth_clients",
26    "webauthn_credentials",
27    "webauthn_challenges",
28    "webauthn_setup_tokens",
29    "user_api_keys",
30    "user_device_certs",
31    "bridge_sessions",
32    "bridge_exchange_codes",
33    "federated_identities",
34];
35
36impl UserRepository {
37    pub async fn merge_users(&self, source_id: &UserId, target_id: &UserId) -> Result<MergeResult> {
38        let mut conn = self.write_pool.acquire().await?;
39        let mut tx = conn.begin().await?;
40        let source = source_id.as_str();
41        let target = target_id.as_str();
42
43        let sessions = transfer_sessions(&mut tx, source, target).await?;
44        let tasks = transfer_tasks(&mut tx, source, target).await?;
45        let mut total_rows = sessions + tasks;
46        total_rows += transfer_audit_rows(&mut tx, source, target).await?;
47        total_rows += transfer_content_rows(&mut tx, source, target).await?;
48        record_merge_attribution(&mut tx, source, target).await?;
49
50        sqlx::query!(
51            "UPDATE fingerprint_reputation SET associated_user_ids = \
52             array_replace(associated_user_ids, $2, $1) WHERE $2 = ANY(associated_user_ids)",
53            target,
54            source
55        )
56        .execute(&mut *tx)
57        .await?;
58
59        sqlx::query!(
60            "DELETE FROM ai_quota_buckets WHERE subject_kind = 'user' AND subject_id = $1",
61            source
62        )
63        .execute(&mut *tx)
64        .await?;
65
66        sqlx::query!("DELETE FROM users WHERE id = $1", source)
67            .execute(&mut *tx)
68            .await?;
69
70        tx.commit().await?;
71        Ok(MergeResult {
72            sessions,
73            tasks,
74            total_rows,
75        })
76    }
77}
78
79async fn transfer_sessions(
80    tx: &mut Transaction<'_, Postgres>,
81    source: &str,
82    target: &str,
83) -> Result<u64> {
84    let result = sqlx::query!(
85        "UPDATE user_sessions SET user_id = $1 WHERE user_id = $2",
86        target,
87        source
88    )
89    .execute(&mut **tx)
90    .await?;
91    Ok(result.rows_affected())
92}
93
94async fn transfer_tasks(
95    tx: &mut Transaction<'_, Postgres>,
96    source: &str,
97    target: &str,
98) -> Result<u64> {
99    let result = sqlx::query!(
100        "UPDATE agent_tasks SET user_id = $1 WHERE user_id = $2",
101        target,
102        source
103    )
104    .execute(&mut **tx)
105    .await?;
106    Ok(result.rows_affected())
107}
108
109// Why: governance_decisions is append-only — a decision is evidence of what was
110// authorised for whom at the time, so the merge is recorded as a new decision
111// rather than by re-attributing the source user's history to the target. A
112// reader following the target's trail finds this row and the source id in it.
113async fn record_merge_attribution(
114    tx: &mut Transaction<'_, Postgres>,
115    source: &str,
116    target: &str,
117) -> Result<()> {
118    let id = uuid::Uuid::new_v4().to_string();
119    let context_id = ContextId::derived_from_session(&SessionId::new(id.clone()));
120    sqlx::query!(
121        "INSERT INTO governance_decisions (id, user_id, session_id, tool_name, decision, policy, \
122         reason, actor_kind, actor_id, context_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, \
123         $10)",
124        id,
125        target,
126        id,
127        MERGE_TOOL_NAME,
128        "allow",
129        MERGE_POLICY,
130        format!("account merge: {source} merged into {target}"),
131        "system",
132        target,
133        context_id.as_str(),
134    )
135    .execute(&mut **tx)
136    .await?;
137    Ok(())
138}
139
140async fn transfer_audit_rows(
141    tx: &mut Transaction<'_, Postgres>,
142    source: &str,
143    target: &str,
144) -> Result<u64> {
145    let mut moved = 0;
146    moved += sqlx::query!(
147        "UPDATE task_messages SET user_id = $1 WHERE user_id = $2",
148        target,
149        source
150    )
151    .execute(&mut **tx)
152    .await?
153    .rows_affected();
154    moved += sqlx::query!(
155        "UPDATE user_contexts SET user_id = $1 WHERE user_id = $2",
156        target,
157        source
158    )
159    .execute(&mut **tx)
160    .await?
161    .rows_affected();
162    moved += sqlx::query!(
163        "UPDATE mcp_tool_executions SET user_id = $1 WHERE user_id = $2",
164        target,
165        source
166    )
167    .execute(&mut **tx)
168    .await?
169    .rows_affected();
170    moved += sqlx::query!(
171        "UPDATE mcp_artifacts SET user_id = $1 WHERE user_id = $2",
172        target,
173        source
174    )
175    .execute(&mut **tx)
176    .await?
177    .rows_affected();
178    moved += sqlx::query!(
179        "UPDATE mcp_sessions SET user_id = $1 WHERE user_id = $2",
180        target,
181        source
182    )
183    .execute(&mut **tx)
184    .await?
185    .rows_affected();
186    moved += sqlx::query!(
187        "UPDATE logs SET user_id = $1 WHERE user_id = $2",
188        target,
189        source
190    )
191    .execute(&mut **tx)
192    .await?
193    .rows_affected();
194    Ok(moved)
195}
196
197async fn transfer_content_rows(
198    tx: &mut Transaction<'_, Postgres>,
199    source: &str,
200    target: &str,
201) -> Result<u64> {
202    let mut moved = 0;
203    moved += sqlx::query!(
204        "UPDATE ai_requests SET user_id = $1 WHERE user_id = $2",
205        target,
206        source
207    )
208    .execute(&mut **tx)
209    .await?
210    .rows_affected();
211    moved += sqlx::query!(
212        "UPDATE engagement_events SET user_id = $1 WHERE user_id = $2",
213        target,
214        source
215    )
216    .execute(&mut **tx)
217    .await?
218    .rows_affected();
219    moved += sqlx::query!(
220        "UPDATE analytics_events SET user_id = $1 WHERE user_id = $2",
221        target,
222        source
223    )
224    .execute(&mut **tx)
225    .await?
226    .rows_affected();
227    moved += sqlx::query!(
228        "UPDATE event_outbox SET user_id = $1 WHERE user_id = $2",
229        target,
230        source
231    )
232    .execute(&mut **tx)
233    .await?
234    .rows_affected();
235    moved += sqlx::query!(
236        "UPDATE files SET user_id = $1 WHERE user_id = $2",
237        target,
238        source
239    )
240    .execute(&mut **tx)
241    .await?
242    .rows_affected();
243    moved += sqlx::query!(
244        "UPDATE link_clicks SET user_id = $1 WHERE user_id = $2",
245        target,
246        source
247    )
248    .execute(&mut **tx)
249    .await?
250    .rows_affected();
251    Ok(moved)
252}
systemprompt_users/repository/user/merge.rs

systemprompt_users/repository/user/
merge.rs
