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

systemprompt_users/repository/user/
merge.rs
