meguri/src/identity.rs

//! Instance identity: configuration sync, group-derived access, SSO accounts, bootstrap
//! accounts and contact linking. Authorization stays on the server; the browser only
//! displays what these functions decide.
use crate::{
    access::{Access, Config, Role, WorkspaceConfig, bootstrap_password, resolve_access},
    domain::*,
    error::{Error, Result, ensure},
    models::*,
    service,
    store::{self, Store, WorkspaceRecord},
};
use chrono::{DateTime, Utc};
use cot::db::{DatabaseError, Model, query};
use serde_json::json;
use std::collections::BTreeSet;
use uuid::Uuid;

fn config_error(message: impl Into<String>) -> Error {
    Error::Validation(format!("Invalid configuration: {}", message.into()))
}

pub fn normalize_email(value: &str) -> Option<String> {
    let email = value.trim().to_lowercase();
    (email.len() <= 254 && email_address::EmailAddress::is_valid(&email)).then_some(email)
}

pub fn new_user_row(username: String, display_name: String, hash: Option<String>) -> UserRow {
    UserRow {
        language: None,
        demo_deleted_at: None,
        id: Uuid::new_v4().to_string(),
        username,
        display_name,
        country: "CY".into(),
        timezone: None,
        password_hash: hash,
        created_at: Utc::now(),
        email: None,
        instance_admin: false,
        bootstrap: false,
        groups: "[]".into(),
    }
}

async fn insert_user(db: &Store, row: &UserRow) -> Result<bool> {
    match row.clone().insert(&db.db).await {
        Ok(_) => Ok(true),
        Err(DatabaseError::UniqueViolation) => Ok(false),
        Err(e) => Err(e.into()),
    }
}

// ---- workspaces from configuration ---------------------------------------------------

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WorkspaceChange {
    Created,
    Adopted,
    Enabled,
    Renamed,
}

#[derive(Debug, Clone)]
pub struct EnsuredWorkspace {
    pub id: Uuid,
    pub changes: Vec<WorkspaceChange>,
}

const CREATE_CONFIGURED_WORKSPACE: &str = r#"
WITH workspace AS (
 INSERT INTO meguri__workspaces(id,key,document,revision,created_at) VALUES($1,$2,$3,1,now()) RETURNING *
), history AS (
 INSERT INTO meguri__workspace_history(id,workspace_id,revision,document,effective_at,actor_id) SELECT id||':1',id,1,document,clock_timestamp(),NULL FROM workspace
), jobs AS (
 INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt) SELECT id,1,'pending',0,now() FROM workspace
)
INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at) SELECT id,NULL,'workspace.created',$4,now() FROM workspace
"#;

const ENABLE_WORKSPACE: &str = r#"
WITH enabled AS (
 UPDATE meguri__workspaces SET disabled_at=NULL WHERE id=$1 AND disabled_at IS NOT NULL AND demo_deleted_at IS NULL RETURNING id,revision
), jobs AS (
 INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt) SELECT id,revision,'pending',0,now() FROM enabled
 ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision,state='pending',attempts=0,next_attempt=now(),last_error=NULL
)
INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at) SELECT id,NULL,'workspace.enabled','{}',now() FROM enabled
"#;

const DISABLE_WORKSPACE: &str = r#"
WITH disabled AS (
 UPDATE meguri__workspaces SET disabled_at=now() WHERE id=$1 AND disabled_at IS NULL AND demo_deleted_at IS NULL RETURNING id
), jobs AS (
 DELETE FROM meguri__sync_jobs WHERE workspace_id IN(SELECT id FROM disabled)
), members AS (
 DELETE FROM meguri__workspace_members WHERE workspace_id IN(SELECT id FROM disabled)
)
INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at) SELECT id,NULL,'workspace.disabled','{}',now() FROM disabled
"#;

/// Active (not demo-expired) workspace rows, including disabled ones.
async fn live_rows(db: &Store) -> Result<Vec<WorkspaceRow>> {
    Ok(WorkspaceRow::objects()
        .all(&db.db)
        .await?
        .into_iter()
        .filter(|r| r.demo_deleted_at.is_none())
        .collect())
}

fn record_of(row: &WorkspaceRow) -> Result<WorkspaceRecord> {
    Ok(WorkspaceRecord {
        id: store::uuid(&row.id)?,
        revision: row.revision,
        document: serde_json::from_str(&row.document)?,
    })
}

fn ambiguous_adoption(config: &WorkspaceConfig, matches: usize) -> Error {
    config_error(format!(
        "workspace {:?} matches {matches} existing workspaces named {:?}; rename or delete the extras so the key can adopt exactly one",
        config.key, config.name
    ))
}

/// Everything that can be refused from configuration alone is refused before the first write,
/// so a rejected configuration never leaves a half-applied instance behind.
async fn preflight(db: &Store, config: &Config, allow_disable_all: bool) -> Result<()> {
    let rows = live_rows(db).await?;
    if config.workspaces.is_empty() && !allow_disable_all {
        let active = rows.iter().filter(|r| r.disabled_at.is_none()).count();
        ensure(
            active == 0,
            format!(
                "Refusing to disable {active} existing workspaces because the configuration lists none. Fix the configuration or set MEGURI_ALLOW_DISABLE_ALL=true to disable them all."
            ),
        )?;
    }
    for workspace in &config.workspaces {
        if rows
            .iter()
            .any(|r| r.key.as_deref() == Some(workspace.key.as_str()))
        {
            continue;
        }
        let mut matches = 0;
        for row in rows.iter().filter(|r| r.key.is_none()) {
            if record_of(row)?.document.name == workspace.name {
                matches += 1;
            }
        }
        if matches > 1 {
            return Err(ambiguous_adoption(workspace, matches));
        }
    }
    for username in config.bootstrap_usernames() {
        let name = username.clone();
        if let Some(row) = query!(UserRow,$username==name).get(&db.db).await? {
            let id = row.id;
            if query!(ExternalIdentityRow,$user_id==id)
                .exists(&db.db)
                .await?
            {
                return Err(config_error(format!(
                    "bootstrap username {username:?} belongs to an SSO account; choose another name"
                )));
            }
        }
    }
    Ok(())
}

/// Creates, adopts (by exact name), enables or renames the workspace declared by `config`.
/// Every step is idempotent, so replicas that run it concurrently converge.
pub async fn ensure_workspace(db: &Store, config: &WorkspaceConfig) -> Result<EnsuredWorkspace> {
    for _ in 0..4 {
        let rows = live_rows(db).await?;
        if let Some(row) = rows
            .iter()
            .find(|r| r.key.as_deref() == Some(config.key.as_str()))
        {
            let id = store::uuid(&row.id)?;
            let mut changes = vec![];
            if row.disabled_at.is_some() {
                let result = db.db.raw_with(ENABLE_WORKSPACE, &[&row.id]).await?;
                if result.rows_affected().0 > 0 {
                    changes.push(WorkspaceChange::Enabled);
                }
            }
            let mut record = record_of(row)?;
            if record.document.name != config.name {
                let before = record.document.name.clone();
                record.document.name = config.name.clone();
                match db
                    .system_commit(
                        &mut record,
                        None,
                        "workspace.renamed",
                        json!({"from":before,"to":config.name}),
                    )
                    .await
                {
                    Ok(()) => changes.push(WorkspaceChange::Renamed),
                    Err(Error::Conflict(_)) => continue,
                    Err(e) => return Err(e),
                }
            }
            return Ok(EnsuredWorkspace { id, changes });
        }
        let mut candidates = vec![];
        for row in rows.iter().filter(|r| r.key.is_none()) {
            if record_of(row)?.document.name == config.name {
                candidates.push(row);
            }
        }
        if candidates.len() > 1 {
            return Err(ambiguous_adoption(config, candidates.len()));
        }
        if let Some(row) = candidates.first() {
            let adopted = db
                .db
                .raw_with(
                    "UPDATE meguri__workspaces SET key=$2 WHERE id=$1 AND key IS NULL",
                    &[&row.id, &config.key],
                )
                .await?;
            if adopted.rows_affected().0 == 0 {
                continue;
            }
            let mut changes = vec![WorkspaceChange::Adopted];
            if row.disabled_at.is_some() {
                db.db.raw_with(ENABLE_WORKSPACE, &[&row.id]).await?;
                changes.push(WorkspaceChange::Enabled);
            }
            return Ok(EnsuredWorkspace {
                id: store::uuid(&row.id)?,
                changes,
            });
        }
        let id = Uuid::new_v4();
        let document = serde_json::to_string(&WorkspaceDocument::new(config.name.clone()))?;
        let details = json!({"key":config.key}).to_string();
        match db
            .db
            .raw_with(
                CREATE_CONFIGURED_WORKSPACE,
                &[&id.to_string(), &config.key, &document, &details],
            )
            .await
        {
            Ok(_) => {
                return Ok(EnsuredWorkspace {
                    id,
                    changes: vec![WorkspaceChange::Created],
                });
            }
            Err(DatabaseError::UniqueViolation) => continue,
            Err(e) => return Err(e.into()),
        }
    }
    Err(Error::Internal(format!(
        "workspace {:?} kept changing while it was being created",
        config.key
    )))
}

/// The enabled workspace declared with `key`, if it exists.
pub async fn find_workspace(db: &Store, key: &str) -> Result<Option<WorkspaceRecord>> {
    live_rows(db)
        .await?
        .iter()
        .find(|r| r.key.as_deref() == Some(key) && r.disabled_at.is_none())
        .map(record_of)
        .transpose()
}

pub async fn disable_workspace(db: &Store, id: &str) -> Result<bool> {
    Ok(db
        .db
        .raw_with(DISABLE_WORKSPACE, &[&id])
        .await?
        .rows_affected()
        .0
        > 0)
}

// ---- group-derived access -------------------------------------------------------------

const APPLY_ACCESS: &str = r#"
WITH changed AS (
 UPDATE meguri__users SET groups=$2,instance_admin=$3 WHERE id=$1 RETURNING id
), desired AS (
 SELECT w.id AS workspace_id,d.is_owner FROM jsonb_to_recordset($4::jsonb) AS d(key text,is_owner boolean)
 JOIN meguri__workspaces w ON w.key=d.key AND w.demo_deleted_at IS NULL AND w.disabled_at IS NULL
), removed AS (
 DELETE FROM meguri__workspace_members m WHERE m.user_id=$1 AND EXISTS(SELECT 1 FROM changed)
 AND NOT EXISTS(SELECT 1 FROM desired d WHERE d.workspace_id=m.workspace_id)
)
INSERT INTO meguri__workspace_members(id,workspace_id,user_id,permissions,is_owner)
SELECT d.workspace_id||':'||$1,d.workspace_id,$1,
 CASE WHEN d.is_owner THEN $5 ELSE COALESCE((SELECT w.document::jsonb->'member_permissions' FROM meguri__workspaces w WHERE w.id=d.workspace_id)::text,$6) END,
 d.is_owner
FROM desired d WHERE EXISTS(SELECT 1 FROM changed)
ON CONFLICT(id) DO UPDATE SET permissions=EXCLUDED.permissions,is_owner=EXCLUDED.is_owner
"#;

pub fn stored_groups(row: &UserRow) -> Result<Vec<String>> {
    Ok(serde_json::from_str(&row.groups)?)
}

/// The one place that turns groups into memberships and the instance-admin flag. SSO login,
/// the administrator's password-user editor and startup reconciliation all end here.
/// `groups` replaces the stored list when given.
pub async fn apply_access(
    db: &Store,
    config: &Config,
    user: Uuid,
    groups: Option<Vec<String>>,
) -> Result<Access> {
    let id = user.to_string();
    let row = query!(UserRow,$id==id.clone())
        .get(&db.db)
        .await?
        .ok_or(Error::NotFound)?;
    let mut groups = match groups {
        Some(groups) => groups,
        None => stored_groups(&row)?,
    };
    groups.sort();
    groups.dedup();
    let access = resolve_access(config, &groups, row.bootstrap);
    let desired: Vec<_> = access
        .workspaces
        .iter()
        .map(|(key, role)| json!({"key":key,"is_owner":*role == Role::Admin}))
        .collect();
    db.db
        .raw_with(
            APPLY_ACCESS,
            &[
                &id,
                &serde_json::to_string(&groups)?,
                &access.instance_admin,
                &serde_json::to_string(&desired)?,
                &serde_json::to_string(&Permission::all())?,
                &serde_json::to_string(&Permission::employee_default())?,
            ],
        )
        .await?;
    Ok(access)
}

pub async fn reconcile_all(db: &Store, config: &Config) -> Result<usize> {
    let mut count = 0;
    for row in UserRow::objects().all(&db.db).await? {
        if row.demo_deleted_at.is_some() {
            continue;
        }
        apply_access(db, config, store::uuid(&row.id)?, None).await?;
        count += 1;
    }
    Ok(count)
}

// ---- bootstrap accounts ---------------------------------------------------------------

pub enum BootstrapPasswords {
    /// HMAC of a database-held secret and the current UTC hour: identical across replicas.
    Derived,
    /// One generated password for a single-process demonstration run.
    Fixed(String),
}

async fn instance_secret(db: &Store) -> Result<Vec<u8>> {
    let candidate = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
    db.db
        .raw_with(
            "INSERT INTO meguri__instance_state(key,value) VALUES('bootstrap_secret',$1) ON CONFLICT(key) DO NOTHING",
            &[&candidate],
        )
        .await?;
    let secret = query!(InstanceStateRow,$key=="bootstrap_secret".to_owned())
        .get(&db.db)
        .await?
        .ok_or_else(|| Error::Internal("bootstrap secret is missing".into()))?;
    Ok(secret.value.into_bytes())
}

/// Creates or refreshes one bootstrap account. Returns whether the password changed.
async fn ensure_bootstrap_account(db: &Store, username: &str, password: &str) -> Result<bool> {
    for _ in 0..3 {
        let name = username.to_owned();
        match query!(UserRow,$username==name).get(&db.db).await? {
            Some(row) => {
                let id = row.id.clone();
                if query!(ExternalIdentityRow,$user_id==id)
                    .exists(&db.db)
                    .await?
                {
                    return Err(config_error(format!(
                        "bootstrap username {username:?} belongs to an SSO account; choose another name"
                    )));
                }
                if row.bootstrap
                    && let Some(hash) = row.password_hash.clone()
                    && service::password_matches(hash, password.to_owned()).await?
                {
                    return Ok(false);
                }
                let hash = service::hash_password(password.to_owned()).await?;
                db.db
                    .raw_with(
                        "UPDATE meguri__users SET password_hash=$2,bootstrap=true,demo_deleted_at=NULL WHERE id=$1",
                        &[&row.id, &hash],
                    )
                    .await?;
                db.db
                    .raw_with("DELETE FROM meguri__sessions WHERE user_id=$1", &[&row.id])
                    .await?;
                return Ok(true);
            }
            None => {
                let hash = service::hash_password(password.to_owned()).await?;
                let mut row = new_user_row(username.to_owned(), username.to_owned(), Some(hash));
                row.bootstrap = true;
                if insert_user(db, &row).await? {
                    return Ok(true);
                }
            }
        }
    }
    Err(Error::Internal(format!(
        "could not create bootstrap account {username:?}"
    )))
}

/// Accounts that stopped being bootstrap accounts lose their password and their sessions.
async fn release_bootstrap_accounts(db: &Store, keep: &[String]) -> Result<Vec<String>> {
    let mut released = vec![];
    for row in query!(UserRow,$bootstrap==true).all(&db.db).await? {
        if keep.contains(&row.username) {
            continue;
        }
        db.db
            .raw_with(
                "UPDATE meguri__users SET bootstrap=false,password_hash=NULL WHERE id=$1",
                &[&row.id],
            )
            .await?;
        db.db
            .raw_with("DELETE FROM meguri__sessions WHERE user_id=$1", &[&row.id])
            .await?;
        released.push(row.username);
    }
    Ok(released)
}

// ---- synchronization ----------------------------------------------------------------

#[derive(Debug, Default, Clone)]
pub struct SyncReport {
    pub created: Vec<String>,
    pub adopted: Vec<String>,
    pub enabled: Vec<String>,
    pub renamed: Vec<String>,
    pub disabled: Vec<String>,
    pub bootstrap_rotated: Vec<String>,
    pub bootstrap_released: Vec<String>,
    /// Current username/password pairs; the caller decides where to show them.
    pub bootstrap_credentials: Vec<(String, String)>,
}

pub struct SyncOptions {
    pub passwords: BootstrapPasswords,
    pub now: DateTime<Utc>,
    /// Allows a configuration without workspaces to disable every existing one.
    pub allow_disable_all: bool,
}

/// Makes the database match the declarative configuration: configured workspaces exist and are
/// enabled, other workspaces are disabled (never deleted), bootstrap accounts match the
/// configured list, and every user's memberships follow their stored groups.
pub async fn sync_config(db: &Store, config: &Config, options: SyncOptions) -> Result<SyncReport> {
    config.validate()?;
    preflight(db, config, options.allow_disable_all).await?;
    let mut report = SyncReport::default();
    let mut keys = BTreeSet::new();
    for workspace in &config.workspaces {
        keys.insert(workspace.key.clone());
        let ensured = ensure_workspace(db, workspace).await?;
        for change in &ensured.changes {
            let (list, action) = match change {
                WorkspaceChange::Created => (&mut report.created, "workspace.created"),
                WorkspaceChange::Adopted => (&mut report.adopted, "workspace.adopted"),
                WorkspaceChange::Enabled => (&mut report.enabled, "workspace.enabled"),
                WorkspaceChange::Renamed => (&mut report.renamed, "workspace.renamed"),
            };
            list.push(workspace.key.clone());
            db.instance_event(
                None,
                action,
                json!({"key":workspace.key,"id":ensured.id,"name":workspace.name}),
            )
            .await?;
        }
    }
    for row in live_rows(db).await? {
        let configured = row.key.as_ref().is_some_and(|k| keys.contains(k));
        if !configured && row.disabled_at.is_none() && disable_workspace(db, &row.id).await? {
            let name = record_of(&row)?.document.name;
            report
                .disabled
                .push(row.key.clone().unwrap_or_else(|| name.clone()));
            db.instance_event(
                None,
                "workspace.disabled",
                json!({"key":row.key,"id":row.id,"name":name}),
            )
            .await?;
        }
    }
    let usernames = config.bootstrap_usernames();
    report.bootstrap_released = release_bootstrap_accounts(db, usernames).await?;
    if !usernames.is_empty() {
        let secret = match options.passwords {
            BootstrapPasswords::Derived => Some(instance_secret(db).await?),
            BootstrapPasswords::Fixed(_) => None,
        };
        for username in usernames {
            let password = match (&options.passwords, &secret) {
                (BootstrapPasswords::Fixed(password), _) => password.clone(),
                (BootstrapPasswords::Derived, Some(secret)) => {
                    bootstrap_password(secret, username, options.now)
                }
                _ => unreachable!("derived passwords always load the instance secret"),
            };
            if ensure_bootstrap_account(db, username, &password).await? {
                report.bootstrap_rotated.push(username.clone());
                db.instance_event(None, "bootstrap.password_set", json!({"username":username}))
                    .await?;
            }
            report
                .bootstrap_credentials
                .push((username.clone(), password));
        }
    }
    reconcile_all(db, config).await?;
    for (list, label) in [
        (&report.created, "created"),
        (&report.adopted, "adopted"),
        (&report.enabled, "enabled"),
        (&report.renamed, "renamed"),
        (&report.disabled, "disabled"),
    ] {
        if !list.is_empty() {
            tracing::info!(workspaces = ?list, "workspaces {label} from configuration");
        }
    }
    Ok(report)
}

// ---- SSO accounts -------------------------------------------------------------------

#[derive(Debug, Clone)]
pub struct SsoProfile {
    pub issuer: String,
    pub subject: String,
    pub email: Option<String>,
    pub username_hint: Option<String>,
    pub display_name: Option<String>,
    pub groups: Vec<String>,
}

#[derive(Debug)]
pub enum SsoOutcome {
    /// The provider's groups map to nothing; no account is created or kept enabled.
    NoAccess,
    SignedIn {
        user: Box<User>,
        token: String,
        csrf: String,
    },
}

fn username_candidates(profile: &SsoProfile) -> Vec<String> {
    let clean = |value: &str| -> String {
        value
            .trim()
            .to_lowercase()
            .chars()
            .map(|c| {
                if c.is_ascii_lowercase() || c.is_ascii_digit() || "._-".contains(c) {
                    c
                } else {
                    '-'
                }
            })
            .collect::<String>()
            .trim_matches('-')
            .chars()
            .take(56)
            .collect()
    };
    let mut bases: Vec<String> = [
        profile.username_hint.as_deref(),
        profile.email.as_deref().and_then(|e| e.split('@').next()),
    ]
    .into_iter()
    .flatten()
    .map(clean)
    .filter(|b| b.len() >= 3)
    .collect();
    if bases.is_empty() {
        bases.push("user".into());
    }
    let suffix = &service::token_hash(&format!("{}\n{}", profile.issuer, profile.subject))[..6];
    let mut candidates = vec![];
    for base in &bases {
        candidates.push(base.clone());
        candidates.push(format!("{base}-{suffix}"));
    }
    candidates
}

async fn find_identity(db: &Store, profile: &SsoProfile) -> Result<Option<String>> {
    let (issuer, subject) = (profile.issuer.clone(), profile.subject.clone());
    Ok(
        query!(ExternalIdentityRow,$issuer==issuer && $subject==subject)
            .get(&db.db)
            .await?
            .map(|i| i.user_id),
    )
}

async fn create_sso_user(db: &Store, profile: &SsoProfile) -> Result<String> {
    for username in username_candidates(profile) {
        let display = profile
            .display_name
            .as_deref()
            .map(str::trim)
            .filter(|n| !n.is_empty() && n.len() <= 120)
            .unwrap_or(&username)
            .to_owned();
        let mut row = new_user_row(username, display, None);
        row.email = profile.email.clone();
        if !insert_user(db, &row).await? {
            continue;
        }
        let mut identity = ExternalIdentityRow {
            id: Uuid::new_v4().to_string(),
            issuer: profile.issuer.clone(),
            subject: profile.subject.clone(),
            user_id: row.id.clone(),
        };
        match identity.insert(&db.db).await {
            Ok(_) => return Ok(row.id),
            Err(DatabaseError::UniqueViolation) => {
                db.db
                    .raw_with("DELETE FROM meguri__users WHERE id=$1", &[&row.id])
                    .await?;
                return find_identity(db, profile).await?.ok_or_else(|| {
                    Error::Internal("SSO identity disappeared during sign-in".into())
                });
            }
            Err(e) => return Err(e.into()),
        }
    }
    Err(Error::Internal(
        "could not derive a unique username for the SSO account".into(),
    ))
}

/// Signs in a verified SSO identity. Groups are applied before anything else, so access that
/// was revoked in the provider disappears at this login and an unmapped user stays out.
pub async fn sso_sign_in(db: &Store, config: &Config, profile: SsoProfile) -> Result<SsoOutcome> {
    let existing = find_identity(db, &profile).await?;
    let user_id = match existing {
        Some(id) => id,
        None => {
            if resolve_access(config, &profile.groups, false).is_empty() {
                return Ok(SsoOutcome::NoAccess);
            }
            create_sso_user(db, &profile).await?
        }
    };
    let id = store::uuid(&user_id)?;
    let access = apply_access(db, config, id, Some(profile.groups.clone())).await?;
    if access.is_empty() {
        return Ok(SsoOutcome::NoAccess);
    }
    if let Some(email) = &profile.email {
        db.db
            .raw_with(
                "UPDATE meguri__users SET email=$2 WHERE id=$1 AND email IS DISTINCT FROM $2",
                &[&user_id, email],
            )
            .await?;
    }
    if let Err(error) = link_contacts(db, id).await {
        tracing::warn!(user = %id, %error, "could not link the sign-in email to workspace contacts");
    }
    let row = query!(UserRow,$id==user_id)
        .get(&db.db)
        .await?
        .ok_or(Error::NotFound)?;
    let user = store::user(row)?;
    let (token, csrf) = service::create_session(db, user.id).await?;
    Ok(SsoOutcome::SignedIn {
        user: Box::new(user),
        token,
        csrf,
    })
}

// ---- contact linking ---------------------------------------------------------------

/// In every workspace the user belongs to, binds the user's email to the scheduling person
/// that carries it, so people planned by email are ready at first sign-in:
/// * a contact with the same email becomes the user (the contact's old ID is kept as an alias
///   and every reference is rewritten, preserving schedules, requests and variables);
/// * otherwise the user's own person gains the email, or is added with it.
///
/// Ambiguous or conflicting data is skipped, never guessed. Access is never granted by
/// email: only group membership creates workspace access.
pub async fn link_contacts(db: &Store, user: Uuid) -> Result<()> {
    let id = user.to_string();
    let row = query!(UserRow,$id==id.clone())
        .get(&db.db)
        .await?
        .ok_or(Error::NotFound)?;
    let Some(email) = row.email.as_deref().and_then(normalize_email) else {
        return Ok(());
    };
    for membership in query!(WorkspaceMemberRow,$user_id==id).all(&db.db).await? {
        let workspace = store::uuid(&membership.workspace_id)?;
        for _ in 0..3 {
            let Ok(mut record) = db.workspace(workspace).await else {
                break;
            };
            let members = db.member_ids(workspace).await?;
            let Some((document, details)) = linked_document(&record.document, &row, user, &email)
            else {
                break;
            };
            record.document = document;
            if record.document.validate(&members).is_err() {
                tracing::warn!(%workspace, %user, "contact linking skipped: the result would be invalid");
                break;
            }
            match db
                .system_commit(&mut record, Some(user), "person.linked", details)
                .await
            {
                Ok(()) => break,
                Err(Error::Conflict(_)) => continue,
                Err(e) => return Err(e),
            }
        }
    }
    Ok(())
}

/// Pure transformation for [`link_contacts`]. `None` means nothing to change.
pub fn linked_document(
    document: &WorkspaceDocument,
    user: &UserRow,
    user_id: Uuid,
    email: &str,
) -> Option<(WorkspaceDocument, serde_json::Value)> {
    let contacts: Vec<usize> = document
        .people
        .iter()
        .enumerate()
        .filter(|(_, p)| p.email.as_deref() == Some(email) && p.id != user_id)
        .map(|(i, _)| i)
        .collect();
    if contacts.len() > 1 {
        return None;
    }
    if let Some(&index) = contacts.first() {
        let contact = &document.people[index];
        if contact.user_id.is_some() {
            return None;
        }
        if document
            .people
            .iter()
            .any(|p| p.id == user_id && p.email.as_deref().is_some_and(|e| e != email))
        {
            return None;
        }
        let old = contact.id;
        let mut working = document.clone();
        working.people.retain(|p| p.id != user_id);
        let text = serde_json::to_string(&working)
            .ok()?
            .replace(&old.to_string(), &user_id.to_string());
        let mut linked: WorkspaceDocument = serde_json::from_str(&text).ok()?;
        let person = linked.people.iter_mut().find(|p| p.id == user_id)?;
        person.user_id = Some(user_id);
        if !person.aliases.contains(&old) {
            person.aliases.push(old);
        }
        return Some((
            linked,
            json!({"user_id":user_id,"contact_id":old,"merged":true}),
        ));
    }
    match document.people.iter().position(|p| p.id == user_id) {
        Some(index) => {
            let person = &document.people[index];
            if person.email.is_some() && person.user_id == Some(user_id) {
                return None;
            }
            if person.email.as_deref().is_some_and(|e| e != email) {
                return None;
            }
            let mut linked = document.clone();
            linked.people[index].email = Some(email.to_owned());
            linked.people[index].user_id = Some(user_id);
            Some((linked, json!({"user_id":user_id,"merged":false})))
        }
        None => {
            let mut linked = document.clone();
            linked.people.push(Person {
                id: user_id,
                user_id: Some(user_id),
                display_name: user.display_name.clone(),
                email: Some(email.to_owned()),
                country: Some(user.country.clone()),
                aliases: vec![],
            });
            Some((linked, json!({"user_id":user_id,"merged":false})))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn row() -> UserRow {
        new_user_row("dana".into(), "Dana".into(), None)
    }
    fn contact(email: &str) -> Person {
        Person {
            id: Uuid::new_v4(),
            user_id: None,
            display_name: "Contact".into(),
            email: Some(email.into()),
            country: Some("AM".into()),
            aliases: vec![],
        }
    }

    #[test]
    fn a_matching_contact_becomes_the_user_everywhere_without_duplicates() {
        let user = Uuid::new_v4();
        let existing_self = Person {
            id: user,
            user_id: Some(user),
            display_name: "Dana".into(),
            email: None,
            country: Some("CY".into()),
            aliases: vec![],
        };
        let mut doc = WorkspaceDocument::new("Team".into());
        let c = contact("dana@corp.test");
        let old = c.id;
        doc.people = vec![existing_self, c];
        doc.generation.variables = vec![];
        doc.absences.push(Absence {
            id: Uuid::new_v4(),
            user_id: old,
            start: "2026-11-01".parse().unwrap(),
            end: "2026-11-02".parse().unwrap(),
        });
        let (linked, details) = linked_document(&doc, &row(), user, "dana@corp.test").unwrap();
        assert_eq!(linked.people.len(), 1, "no second entry for the same user");
        let person = &linked.people[0];
        assert_eq!((person.id, person.user_id), (user, Some(user)));
        assert_eq!(person.aliases, vec![old]);
        assert_eq!(person.display_name, "Contact");
        assert_eq!(linked.absences[0].user_id, user);
        assert_eq!(details["contact_id"], json!(old));
        assert_eq!(
            serde_json::to_string(&linked)
                .unwrap()
                .matches(&old.to_string())
                .count(),
            1,
            "the old id survives only as the alias"
        );
        // Applying the result again is a no-op.
        assert!(linked_document(&linked, &row(), user, "dana@corp.test").is_none());
    }

    #[test]
    fn ambiguous_or_conflicting_data_is_never_guessed() {
        let user = Uuid::new_v4();
        let mut doc = WorkspaceDocument::new("Team".into());
        // Two contacts with one email cannot validate, and must not be merged either.
        doc.people = vec![contact("dana@corp.test"), contact("dana@corp.test")];
        assert!(linked_document(&doc, &row(), user, "dana@corp.test").is_none());
        // A contact that already belongs to another login is not stolen.
        let mut taken = contact("dana@corp.test");
        taken.user_id = Some(Uuid::new_v4());
        doc.people = vec![taken];
        assert!(linked_document(&doc, &row(), user, "dana@corp.test").is_none());
        // The user's own person carries a different email: leave both alone.
        doc.people = vec![
            Person {
                id: user,
                user_id: Some(user),
                display_name: "Dana".into(),
                email: Some("other@corp.test".into()),
                country: None,
                aliases: vec![],
            },
            contact("dana@corp.test"),
        ];
        assert!(linked_document(&doc, &row(), user, "dana@corp.test").is_none());
    }

    #[test]
    fn users_without_a_contact_gain_a_person_with_their_email() {
        let user = Uuid::new_v4();
        let doc = WorkspaceDocument::new("Team".into());
        let (linked, _) = linked_document(&doc, &row(), user, "dana@corp.test").unwrap();
        assert_eq!(linked.people[0].email.as_deref(), Some("dana@corp.test"));
        assert_eq!(linked.people[0].user_id, Some(user));
        let mut with_person = doc.clone();
        with_person.people.push(Person {
            id: user,
            user_id: Some(user),
            display_name: "Dana".into(),
            email: None,
            country: None,
            aliases: vec![],
        });
        let (linked, _) = linked_document(&with_person, &row(), user, "dana@corp.test").unwrap();
        assert_eq!(linked.people.len(), 1);
        assert_eq!(linked.people[0].email.as_deref(), Some("dana@corp.test"));
    }

    #[test]
    fn usernames_are_clean_unique_candidates() {
        let profile = |hint: Option<&str>, email: Option<&str>| SsoProfile {
            issuer: "https://idp.test".into(),
            subject: "sub-1".into(),
            email: email.map(str::to_owned),
            username_hint: hint.map(str::to_owned),
            display_name: None,
            groups: vec![],
        };
        let candidates = username_candidates(&profile(Some("Jane Doe!"), Some("jd@corp.test")));
        assert_eq!(candidates[0], "jane-doe");
        assert!(
            candidates[1].starts_with("jane-doe-") && candidates[1].len() == "jane-doe-".len() + 6
        );
        // Too-short hints fall back to the email name, and nothing at all to a neutral base.
        assert_eq!(
            username_candidates(&profile(Some("ab"), Some("Lena.Ivanova@corp.test")))[0],
            "lena.ivanova"
        );
        assert_eq!(username_candidates(&profile(None, None))[0], "user");
        assert!(candidates.iter().all(|c| crate::access::valid_username(c)));
        let long = "x".repeat(200);
        assert!(
            username_candidates(&profile(Some(&long), None))
                .iter()
                .all(|c| crate::access::valid_username(c))
        );
    }
}