systemprompt_users/repository/user/
merge.rs

1//! The users-owned half of an account merge.
2//!
3//! Every other domain's rows are moved by its own
4//! [`OwnerReassignment`](systemprompt_traits::OwnerReassignment) before this
5//! runs; what is left is this crate's own: the source's sessions move to the
6//! target, the merge is recorded as a governance decision, and the source user
7//! is deleted — in one transaction, so a failure leaves the source in place
8//! for a rerun.
9//!
10//! Copyright (c) systemprompt.io — Business Source License 1.1.
11//! See <https://systemprompt.io> for licensing details.
12
13use sqlx::{Acquire, Postgres, Transaction};
14use systemprompt_identifiers::{Actor, ContextId, SessionId, UserId};
15use systemprompt_security::authz::types::DecisionTag;
16use systemprompt_security::authz::{GovernanceDecisionRecord, insert_governance_decision};
17
18use crate::error::Result;
19use crate::repository::UserRepository;
20
21const MERGE_TOOL_NAME: &str = "users.merge";
22const MERGE_POLICY: &str = "account_merge";
23
24#[derive(Debug, Clone, Copy)]
25pub struct MergeResult {
26    pub sessions: u64,
27    pub tasks: u64,
28    pub total_rows: u64,
29}
30
31pub const MERGE_EXCLUDED_SECURITY_TABLES: &[&str] = &[
32    "oauth_auth_codes",
33    "oauth_refresh_tokens",
34    "oauth_clients",
35    "webauthn_credentials",
36    "webauthn_challenges",
37    "webauthn_setup_tokens",
38    "user_api_keys",
39    "user_device_certs",
40    "bridge_sessions",
41    "bridge_exchange_codes",
42    "federated_identities",
43];
44
45impl UserRepository {
46    pub async fn complete_merge(&self, source_id: &UserId, target_id: &UserId) -> Result<u64> {
47        let mut conn = self.write_pool.acquire().await?;
48        let mut tx = conn.begin().await?;
49        Self::lock_unless_held(&mut tx, source_id).await?;
50
51        let sessions = sqlx::query!(
52            "UPDATE user_sessions SET user_id = $1 WHERE user_id = $2",
53            target_id.as_str(),
54            source_id.as_str()
55        )
56        .execute(&mut *tx)
57        .await?
58        .rows_affected();
59
60        record_merge_attribution(&mut tx, source_id, target_id).await?;
61
62        sqlx::query!("DELETE FROM users WHERE id = $1", source_id.as_str())
63            .execute(&mut *tx)
64            .await?;
65
66        tx.commit().await?;
67        Ok(sessions)
68    }
69}
70
71// Why: governance_decisions is append-only — a decision is evidence of what was
72// authorised for whom at the time, so the merge is recorded as a new decision
73// rather than by re-attributing the source user's history to the target. A
74// reader following the target's trail finds this row and the source id in it.
75async fn record_merge_attribution(
76    tx: &mut Transaction<'_, Postgres>,
77    source_id: &UserId,
78    target_id: &UserId,
79) -> Result<()> {
80    let id = uuid::Uuid::new_v4().to_string();
81    let session_id = SessionId::new(id.clone());
82    let context_id = ContextId::derived_from_session(&session_id);
83    let actor = Actor::system(target_id.clone());
84    let reason = format!("account merge: {source_id} merged into {target_id}");
85    let evaluated_rules = serde_json::json!([]);
86    let record = GovernanceDecisionRecord {
87        id: &id,
88        actor: &actor,
89        session_id: Some(&session_id),
90        tool_name: MERGE_TOOL_NAME,
91        agent_id: None,
92        agent_scope: None,
93        decision: DecisionTag::Allow,
94        policy: MERGE_POLICY,
95        reason: &reason,
96        evaluated_rules: &evaluated_rules,
97        plugin_id: None,
98        act_chain: &[],
99        context_id: &context_id,
100        task_id: None,
101        trace_id: None,
102        client_id: None,
103        tool_use_id: None,
104    };
105    insert_governance_decision(&mut **tx, &record).await?;
106    Ok(())
107}
systemprompt_users/repository/user/merge.rs

systemprompt_users/repository/user/
merge.rs
