//! SQLite-backed control-plane store.

use std::sync::Mutex;

use rusqlite::{params, Connection, OptionalExtension};

use crate::error::ControlPlaneError;
use crate::migrations;
use crate::model::{
    password_hash_v1, share_token_hash_v1, AiPolicyRecord, AiRuntimeEventRecord,
    AiRuntimeRunRecord, AppendAiRuntimeEventInput, AppendAuditInput, AuditLogRecord,
    AuthSessionRecord, AuthenticatePasswordInput, CreatePasswordIdentityInput, CreateSessionInput,
    CreateShareLinkInput, CreatedShareLink, DirectoryGrantInput, DirectoryGrantLookup,
    DirectoryGrantRecord, OutboxEventInput, OutboxEventRecord, ResolvedAccess, ResolvedAuthSession,
    ShareLinkRecord, SyncStateRecord, UpsertAiPolicyInput, UpsertAiRuntimeRunInput,
    UpsertSyncStateInput, UpsertUserInput, UserRecord, WorkspaceRecord,
};
use crate::store::ControlPlaneStore;

pub struct SqliteControlPlaneStore {
    conn: Mutex<Connection>,
}

impl SqliteControlPlaneStore {
    pub fn open(path: impl AsRef<std::path::Path>) -> Result<Self, ControlPlaneError> {
        let conn = Connection::open(path)?;
        configure_connection(&conn)?;
        migrations::run_migrations(&conn)?;
        Ok(Self {
            conn: Mutex::new(conn),
        })
    }

    pub fn in_memory() -> Result<Self, ControlPlaneError> {
        let conn = Connection::open_in_memory()?;
        configure_connection(&conn)?;
        migrations::run_migrations(&conn)?;
        Ok(Self {
            conn: Mutex::new(conn),
        })
    }
}

fn configure_connection(conn: &Connection) -> Result<(), ControlPlaneError> {
    conn.execute_batch(
        "PRAGMA journal_mode = WAL;
         PRAGMA synchronous = NORMAL;
         PRAGMA foreign_keys = ON;
         PRAGMA busy_timeout = 5000;",
    )?;
    Ok(())
}

fn now_text() -> String {
    chrono::Utc::now().to_rfc3339()
}

fn new_id(prefix: &str) -> String {
    format!("{prefix}_{}", uuid::Uuid::new_v4().simple())
}

fn capabilities_json(capabilities: &[String]) -> Result<String, ControlPlaneError> {
    serde_json::to_string(capabilities).map_err(ControlPlaneError::from)
}

fn file_uri_to_legacy_path(value: &str) -> String {
    value
        .trim()
        .strip_prefix("file://")
        .unwrap_or("")
        .to_string()
}

fn row_to_user(row: &rusqlite::Row<'_>) -> rusqlite::Result<UserRecord> {
    Ok(UserRecord {
        id: row.get(0)?,
        email: row.get(1)?,
        username: row.get(2)?,
        display_name: row.get(3)?,
        role: row.get(4)?,
        status: row.get(5)?,
        created_at: row.get(6)?,
        updated_at: row.get(7)?,
        revision: row.get(8)?,
    })
}

fn row_to_auth_session(row: &rusqlite::Row<'_>) -> rusqlite::Result<AuthSessionRecord> {
    Ok(AuthSessionRecord {
        id: row.get(0)?,
        user_id: row.get(1)?,
        token_hash: row.get(2)?,
        user_agent: row.get(3)?,
        ip_hash: row.get(4)?,
        created_at: row.get(5)?,
        expires_at: row.get(6)?,
        revoked_at: row.get(7)?,
        last_seen_at: row.get(8)?,
    })
}

fn row_to_workspace(row: &rusqlite::Row<'_>) -> rusqlite::Result<WorkspaceRecord> {
    Ok(WorkspaceRecord {
        id: row.get(0)?,
        owner_user_id: row.get(1)?,
        name: row.get(2)?,
        kind: row.get(3)?,
        root_uri: row.get(4)?,
        root_path: row.get(5)?,
        source_kind: row.get(6)?,
        status: row.get(7)?,
        created_at: row.get(8)?,
        updated_at: row.get(9)?,
        revision: row.get(10)?,
    })
}

fn row_to_grant(row: &rusqlite::Row<'_>) -> rusqlite::Result<DirectoryGrantRecord> {
    Ok(DirectoryGrantRecord {
        id: row.get(0)?,
        user_id: row.get(1)?,
        workspace_id: row.get(2)?,
        root_uri: row.get(3)?,
        root_path: row.get(4)?,
        permission: row.get(5)?,
        recursive: row.get::<_, i64>(6)? != 0,
        capabilities_json: row.get(7)?,
        source: row.get(8)?,
        status: row.get(9)?,
        created_by: row.get(10)?,
        created_at: row.get(11)?,
        updated_at: row.get(12)?,
        revision: row.get(13)?,
    })
}

fn row_to_outbox(row: &rusqlite::Row<'_>) -> rusqlite::Result<OutboxEventRecord> {
    Ok(OutboxEventRecord {
        id: row.get(0)?,
        topic: row.get(1)?,
        event_type: row.get(2)?,
        payload_json: row.get(3)?,
        created_at: row.get(4)?,
        delivered_at: row.get(5)?,
        attempts: row.get(6)?,
    })
}

fn row_to_share_link(row: &rusqlite::Row<'_>) -> rusqlite::Result<ShareLinkRecord> {
    Ok(ShareLinkRecord {
        id: row.get(0)?,
        workspace_id: row.get(1)?,
        resource_kind: row.get(2)?,
        resource_id: row.get(3)?,
        token_hash: row.get(4)?,
        permission: row.get(5)?,
        created_by: row.get(6)?,
        expires_at: row.get(7)?,
        revoked_at: row.get(8)?,
        created_at: row.get(9)?,
        updated_at: row.get(10)?,
        revision: row.get(11)?,
    })
}

fn row_to_audit_log(row: &rusqlite::Row<'_>) -> rusqlite::Result<AuditLogRecord> {
    Ok(AuditLogRecord {
        id: row.get(0)?,
        actor_user_id: row.get(1)?,
        action: row.get(2)?,
        target_kind: row.get(3)?,
        target_id: row.get(4)?,
        metadata_json: row.get(5)?,
        created_at: row.get(6)?,
    })
}

fn row_to_ai_policy(row: &rusqlite::Row<'_>) -> rusqlite::Result<AiPolicyRecord> {
    Ok(AiPolicyRecord {
        id: row.get(0)?,
        user_id: row.get(1)?,
        workspace_id: row.get(2)?,
        allowed_roots_json: row.get(3)?,
        model_policy_json: row.get(4)?,
        quota_json: row.get(5)?,
        created_at: row.get(6)?,
        updated_at: row.get(7)?,
        revision: row.get(8)?,
    })
}

fn row_to_sync_state(row: &rusqlite::Row<'_>) -> rusqlite::Result<SyncStateRecord> {
    Ok(SyncStateRecord {
        id: row.get(0)?,
        workspace_id: row.get(1)?,
        remote_kind: row.get(2)?,
        remote_id: row.get(3)?,
        cursor: row.get(4)?,
        last_synced_at: row.get(5)?,
        status: row.get(6)?,
        error_json: row.get(7)?,
        updated_at: row.get(8)?,
        revision: row.get(9)?,
    })
}

fn row_to_ai_runtime_run(row: &rusqlite::Row<'_>) -> rusqlite::Result<AiRuntimeRunRecord> {
    Ok(AiRuntimeRunRecord {
        id: row.get(0)?,
        user_id: row.get(1)?,
        workspace_id: row.get(2)?,
        document_id: row.get(3)?,
        session_id: row.get(4)?,
        run_id: row.get(5)?,
        title: row.get(6)?,
        profile: row.get(7)?,
        acp_runtime: row.get(8)?,
        trace_id: row.get(9)?,
        status: row.get(10)?,
        runtime_json: row.get(11)?,
        payload_json: row.get(12)?,
        deleted_at: row.get(13)?,
        created_at: row.get(14)?,
        updated_at: row.get(15)?,
        revision: row.get(16)?,
    })
}

fn row_to_ai_runtime_event(row: &rusqlite::Row<'_>) -> rusqlite::Result<AiRuntimeEventRecord> {
    Ok(AiRuntimeEventRecord {
        id: row.get(0)?,
        user_id: row.get(1)?,
        workspace_id: row.get(2)?,
        document_id: row.get(3)?,
        session_id: row.get(4)?,
        run_id: row.get(5)?,
        profile: row.get(6)?,
        acp_runtime: row.get(7)?,
        event_type: row.get(8)?,
        payload_json: row.get(9)?,
        created_at: row.get(10)?,
    })
}

fn derive_ai_runtime_title(payload_json: &str, fallback: &str) -> String {
    let title = serde_json::from_str::<serde_json::Value>(payload_json)
        .ok()
        .and_then(|payload| {
            payload
                .get("title")
                .or_else(|| payload.get("message"))
                .and_then(serde_json::Value::as_str)
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(ToOwned::to_owned)
        })
        .unwrap_or_else(|| fallback.to_string());
    title.chars().take(80).collect()
}

impl ControlPlaneStore for SqliteControlPlaneStore {
    fn upsert_user(&self, input: UpsertUserInput) -> Result<UserRecord, ControlPlaneError> {
        if input.username.trim().is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "username 不能为空".to_string(),
            ));
        }
        if input.display_name.trim().is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "display_name 不能为空".to_string(),
            ));
        }

        let conn = self.conn.lock().unwrap();
        let now = now_text();
        let id = input.id.unwrap_or_else(|| new_id("usr"));
        let role = input.role.unwrap_or_else(|| "user".to_string());

        let existing = conn
            .query_row(
                "SELECT id, email, username, display_name, role, status, created_at, updated_at, revision
                 FROM users WHERE id = ?1",
                params![id],
                row_to_user,
            )
            .optional()?;

        if let Some(existing) = existing {
            let revision = existing.revision + 1;
            conn.execute(
                "UPDATE users
                 SET email = ?1, username = ?2, display_name = ?3, role = ?4, updated_at = ?5, revision = ?6
                 WHERE id = ?7",
                params![
                    input.email,
                    input.username,
                    input.display_name,
                    role,
                    now,
                    revision,
                    existing.id
                ],
            )?;
            return Ok(UserRecord {
                id: existing.id,
                email: input.email,
                username: input.username,
                display_name: input.display_name,
                role,
                status: existing.status,
                created_at: existing.created_at,
                updated_at: now,
                revision,
            });
        }

        conn.execute(
            "INSERT INTO users (id, email, username, display_name, role, status, created_at, updated_at, revision)
             VALUES (?1, ?2, ?3, ?4, ?5, 'active', ?6, ?7, 1)",
            params![id, input.email, input.username, input.display_name, role, now, now],
        )?;

        Ok(UserRecord {
            id,
            email: input.email,
            username: input.username,
            display_name: input.display_name,
            role,
            status: "active".to_string(),
            created_at: now.clone(),
            updated_at: now,
            revision: 1,
        })
    }

    fn create_password_identity(
        &self,
        input: CreatePasswordIdentityInput,
    ) -> Result<(), ControlPlaneError> {
        let username = input.username.trim();
        if username.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "username 不能为空".to_string(),
            ));
        }
        if input.password.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "password 不能为空".to_string(),
            ));
        }

        let conn = self.conn.lock().unwrap();
        let now = now_text();
        let user_id = input.user_id;
        let password_hash = password_hash_v1(&input.password);
        conn.execute(
            "INSERT INTO auth_identities (id, user_id, provider, provider_subject, password_hash, password_version, created_at, updated_at)
             VALUES (?1, ?2, 'password_username', ?3, ?4, 1, ?5, ?6)
             ON CONFLICT(provider, provider_subject)
             DO UPDATE SET user_id = excluded.user_id, password_hash = excluded.password_hash, updated_at = excluded.updated_at",
            params![
                new_id("ident"),
                user_id,
                username,
                password_hash,
                now,
                now
            ],
        )?;
        if let Some(email) = input
            .email
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty())
        {
            conn.execute(
                "INSERT INTO auth_identities (id, user_id, provider, provider_subject, password_hash, password_version, created_at, updated_at)
                 VALUES (?1, ?2, 'password_email', ?3, ?4, 1, ?5, ?6)
                 ON CONFLICT(provider, provider_subject)
                 DO UPDATE SET user_id = excluded.user_id, password_hash = excluded.password_hash, updated_at = excluded.updated_at",
                params![
                    new_id("ident"),
                    user_id,
                    email,
                    password_hash_v1(&input.password),
                    now,
                    now
                ],
            )?;
        }
        Ok(())
    }

    fn authenticate_password(
        &self,
        input: AuthenticatePasswordInput,
    ) -> Result<ResolvedAuthSession, ControlPlaneError> {
        let account = input.account.trim();
        if account.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "account 不能为空".to_string(),
            ));
        }

        let conn = self.conn.lock().unwrap();
        let provider = if account.contains('@') {
            "password_email"
        } else {
            "password_username"
        };
        let expected_hash = password_hash_v1(&input.password);
        let user = conn
            .query_row(
                "SELECT u.id, u.email, u.username, u.display_name, u.role, u.status, u.created_at, u.updated_at, u.revision
                 FROM auth_identities i
                 JOIN users u ON u.id = i.user_id
                 WHERE i.provider = ?1
                   AND i.provider_subject = ?2
                   AND i.password_hash = ?3
                   AND u.status = 'active'
                 LIMIT 1",
                params![provider, account, expected_hash],
                row_to_user,
            )
            .optional()?
            .ok_or_else(|| ControlPlaneError::Unauthorized("账号或密码错误".to_string()))?;
        drop(conn);

        let session = self.create_session(CreateSessionInput {
            id: input.session_id,
            user_id: user.id.clone(),
            token_hash: input.token_hash,
            user_agent: input.user_agent,
            ip_hash: input.ip_hash,
            expires_at: input.expires_at,
        })?;
        Ok(ResolvedAuthSession { session, user })
    }

    fn create_session(
        &self,
        input: CreateSessionInput,
    ) -> Result<AuthSessionRecord, ControlPlaneError> {
        let token_hash = input.token_hash.trim();
        if token_hash.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "token_hash 不能为空".to_string(),
            ));
        }

        let conn = self.conn.lock().unwrap();
        let now = now_text();
        let expires_at = input
            .expires_at
            .unwrap_or_else(|| (chrono::Utc::now() + chrono::Duration::days(30)).to_rfc3339());
        let record = AuthSessionRecord {
            id: input.id.unwrap_or_else(|| new_id("sess")),
            user_id: input.user_id,
            token_hash: token_hash.to_string(),
            user_agent: input.user_agent,
            ip_hash: input.ip_hash,
            created_at: now.clone(),
            expires_at,
            revoked_at: None,
            last_seen_at: now.clone(),
        };
        conn.execute(
            "INSERT INTO auth_sessions (id, user_id, token_hash, user_agent, ip_hash, created_at, expires_at, revoked_at, last_seen_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8)",
            params![
                record.id,
                record.user_id,
                record.token_hash,
                record.user_agent,
                record.ip_hash,
                record.created_at,
                record.expires_at,
                record.last_seen_at
            ],
        )?;
        Ok(record)
    }

    fn get_session_by_token_hash(
        &self,
        token_hash: &str,
    ) -> Result<Option<ResolvedAuthSession>, ControlPlaneError> {
        let token_hash = token_hash.trim();
        if token_hash.is_empty() {
            return Ok(None);
        }

        let conn = self.conn.lock().unwrap();
        let now = now_text();
        let resolved = conn
            .query_row(
                "SELECT
                   s.id, s.user_id, s.token_hash, s.user_agent, s.ip_hash, s.created_at, s.expires_at, s.revoked_at, s.last_seen_at,
                   u.id, u.email, u.username, u.display_name, u.role, u.status, u.created_at, u.updated_at, u.revision
                 FROM auth_sessions s
                 JOIN users u ON u.id = s.user_id
                 WHERE s.token_hash = ?1
                   AND s.revoked_at IS NULL
                   AND s.expires_at > ?2
                   AND u.status = 'active'
                 LIMIT 1",
                params![token_hash, now],
                |row| {
                    Ok(ResolvedAuthSession {
                        session: row_to_auth_session(row)?,
                        user: UserRecord {
                            id: row.get(9)?,
                            email: row.get(10)?,
                            username: row.get(11)?,
                            display_name: row.get(12)?,
                            role: row.get(13)?,
                            status: row.get(14)?,
                            created_at: row.get(15)?,
                            updated_at: row.get(16)?,
                            revision: row.get(17)?,
                        },
                    })
                },
            )
            .optional()?;

        if let Some(resolved) = &resolved {
            conn.execute(
                "UPDATE auth_sessions SET last_seen_at = ?1 WHERE id = ?2",
                params![now, resolved.session.id],
            )?;
        }

        Ok(resolved)
    }

    fn revoke_session(&self, session_id: &str) -> Result<(), ControlPlaneError> {
        let session_id = session_id.trim();
        if session_id.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "session_id 不能为空".to_string(),
            ));
        }
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE auth_sessions SET revoked_at = ?1 WHERE id = ?2 AND revoked_at IS NULL",
            params![now_text(), session_id],
        )?;
        Ok(())
    }

    fn ensure_default_workspace(
        &self,
        actor_id: &str,
    ) -> Result<WorkspaceRecord, ControlPlaneError> {
        let conn = self.conn.lock().unwrap();
        let existing = conn
            .query_row(
                "SELECT id, owner_user_id, name, kind, root_uri, root_path, source_kind, status, created_at, updated_at, revision
                 FROM workspaces
                 WHERE owner_user_id = ?1 AND kind = 'personal' AND status = 'active'
                 LIMIT 1",
                params![actor_id],
                row_to_workspace,
            )
            .optional()?;
        if let Some(workspace) = existing {
            return Ok(workspace);
        }

        let user = conn
            .query_row(
                "SELECT id, email, username, display_name, role, status, created_at, updated_at, revision
                 FROM users WHERE id = ?1",
                params![actor_id],
                row_to_user,
            )
            .optional()?
            .ok_or_else(|| ControlPlaneError::NotFound(format!("user not found: {actor_id}")))?;

        let now = now_text();
        let workspace = WorkspaceRecord {
            id: new_id("ws"),
            owner_user_id: user.id.clone(),
            name: format!("{} 的空间", user.display_name),
            kind: "personal".to_string(),
            root_uri: format!("local://users/{}/workspaces/my-space", user.id),
            root_path: format!(
                "/mnt/Data1T/Mnote_data/users/{}/workspaces/my-space",
                user.id
            ),
            source_kind: "local_folder".to_string(),
            status: "active".to_string(),
            created_at: now.clone(),
            updated_at: now.clone(),
            revision: 1,
        };

        conn.execute(
            "INSERT INTO workspaces (id, owner_user_id, name, kind, root_uri, root_path, source_kind, status, created_at, updated_at, revision)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
            params![
                workspace.id,
                workspace.owner_user_id,
                workspace.name,
                workspace.kind,
                workspace.root_uri,
                workspace.root_path,
                workspace.source_kind,
                workspace.status,
                workspace.created_at,
                workspace.updated_at,
                workspace.revision
            ],
        )?;
        conn.execute(
            "INSERT INTO workspace_members (id, workspace_id, user_id, role, status, created_at, updated_at, revision)
             VALUES (?1, ?2, ?3, 'owner', 'active', ?4, ?5, 1)",
            params![new_id("wsm"), workspace.id, user.id, now, now],
        )?;
        conn.execute(
            "INSERT INTO directory_grants (id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision)
             VALUES (?1, ?2, ?3, ?4, ?5, 'write', 1, ?6, 'auto', 'active', ?7, ?8, ?9, 1)",
            params![
                new_id("grant"),
                workspace.owner_user_id,
                workspace.id,
                workspace.root_uri,
                workspace.root_path,
                capabilities_json(&["ai".to_string(), "share".to_string()])?,
                workspace.owner_user_id,
                now,
                now
            ],
        )?;

        Ok(workspace)
    }

    fn grant_directory_access(
        &self,
        input: DirectoryGrantInput,
    ) -> Result<DirectoryGrantRecord, ControlPlaneError> {
        let conn = self.conn.lock().unwrap();
        let now = now_text();
        let record = DirectoryGrantRecord {
            id: new_id("grant"),
            user_id: input.user_id,
            workspace_id: input.workspace_id,
            root_uri: input.root_uri,
            root_path: input.root_path,
            permission: input.permission,
            recursive: input.recursive,
            capabilities_json: capabilities_json(&input.capabilities)?,
            source: input.source,
            status: "active".to_string(),
            created_by: input.created_by,
            created_at: now.clone(),
            updated_at: now,
            revision: 1,
        };
        conn.execute(
            "INSERT INTO directory_grants (id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
            params![
                record.id,
                record.user_id,
                record.workspace_id,
                record.root_uri,
                record.root_path,
                record.permission,
                if record.recursive { 1 } else { 0 },
                record.capabilities_json,
                record.source,
                record.status,
                record.created_by,
                record.created_at,
                record.updated_at,
                record.revision
            ],
        )?;
        Ok(record)
    }

    fn list_directory_grants(&self) -> Result<Vec<DirectoryGrantRecord>, ControlPlaneError> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision
             FROM directory_grants
             WHERE status = 'active'
             ORDER BY created_at ASC",
        )?;
        let grants = stmt
            .query_map([], row_to_grant)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(ControlPlaneError::from)?;
        Ok(grants)
    }

    fn list_directory_grants_for_actor(
        &self,
        actor_id: &str,
    ) -> Result<Vec<DirectoryGrantRecord>, ControlPlaneError> {
        let actor_id = actor_id.trim();
        if actor_id.is_empty() {
            return Ok(Vec::new());
        }
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision
             FROM directory_grants
             WHERE user_id = ?1 AND status = 'active'
             ORDER BY created_at ASC",
        )?;
        let grants = stmt
            .query_map(params![actor_id], row_to_grant)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(ControlPlaneError::from)?;
        Ok(grants)
    }

    fn find_directory_grants(
        &self,
        lookup: DirectoryGrantLookup,
    ) -> Result<Vec<DirectoryGrantRecord>, ControlPlaneError> {
        let conn = self.conn.lock().unwrap();
        let grant_id = lookup
            .grant_id
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty());
        let user_id = lookup
            .user_id
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty());
        let root_uri = lookup
            .root_uri
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty());
        let status = if lookup.include_revoked {
            "%"
        } else {
            "active"
        };
        let mut stmt = conn.prepare(
            "SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision
             FROM directory_grants
             WHERE (?1 IS NULL OR id = ?1)
               AND (?2 IS NULL OR user_id = ?2)
               AND (?3 IS NULL OR root_uri = ?3)
               AND status LIKE ?4
             ORDER BY created_at ASC",
        )?;
        let grants = stmt
            .query_map(params![grant_id, user_id, root_uri, status], row_to_grant)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(ControlPlaneError::from)?;
        Ok(grants)
    }

    fn revoke_directory_grant(
        &self,
        grant_id: &str,
        expected_revision: Option<i64>,
    ) -> Result<(), ControlPlaneError> {
        let grant_id = grant_id.trim();
        if grant_id.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "grant_id 不能为空".to_string(),
            ));
        }
        let conn = self.conn.lock().unwrap();
        let existing_revision = conn
            .query_row(
                "SELECT revision FROM directory_grants WHERE id = ?1 AND status = 'active'",
                params![grant_id],
                |row| row.get::<_, i64>(0),
            )
            .optional()?
            .ok_or_else(|| ControlPlaneError::NotFound(format!("grant not found: {grant_id}")))?;
        if let Some(expected_revision) = expected_revision {
            if expected_revision != existing_revision {
                return Err(ControlPlaneError::Conflict(
                    "目录授权 revision 已变化".to_string(),
                ));
            }
        }
        conn.execute(
            "UPDATE directory_grants
             SET status = 'revoked', updated_at = ?1, revision = revision + 1
             WHERE id = ?2 AND status = 'active'",
            params![now_text(), grant_id],
        )?;
        Ok(())
    }

    fn resolve_access(
        &self,
        actor_id: &str,
        root_uri: &str,
    ) -> Result<ResolvedAccess, ControlPlaneError> {
        let conn = self.conn.lock().unwrap();
        let root_path = file_uri_to_legacy_path(root_uri);
        let mut stmt = conn.prepare(
            "SELECT id, user_id, workspace_id, root_uri, root_path, permission, recursive, capabilities_json, source, status, created_by, created_at, updated_at, revision
             FROM directory_grants
             WHERE user_id = ?1
               AND status = 'active'
               AND (
                 ?2 = root_uri
                 OR (recursive = 1 AND ?2 LIKE root_uri || '%')
                 OR (?3 != '' AND ?3 = root_path)
                 OR (?3 != '' AND recursive = 1 AND ?3 LIKE root_path || '/%')
               )",
        )?;
        let grants = stmt
            .query_map(params![actor_id, root_uri, root_path], row_to_grant)?
            .collect::<Result<Vec<_>, _>>()?;
        let mut permission = "none".to_string();
        for grant in &grants {
            permission = max_permission(&permission, &grant.permission).to_string();
        }
        Ok(ResolvedAccess {
            user_id: actor_id.to_string(),
            root_uri: root_uri.to_string(),
            permission,
            grant_ids: grants.into_iter().map(|grant| grant.id).collect(),
        })
    }

    fn create_share_link(
        &self,
        input: CreateShareLinkInput,
    ) -> Result<CreatedShareLink, ControlPlaneError> {
        let workspace_id = input.workspace_id.trim().to_string();
        if workspace_id.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "workspace_id 不能为空".to_string(),
            ));
        }
        let resource_kind = input.resource_kind.trim().to_string();
        let resource_id = input.resource_id.trim().to_string();
        let permission = input.permission.trim().to_string();
        if resource_kind.is_empty() || resource_id.is_empty() || permission.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "share_link 资源与权限不能为空".to_string(),
            ));
        }
        let token = input
            .token
            .filter(|value| !value.trim().is_empty())
            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
        let token_hash = share_token_hash_v1(&token);
        let conn = self.conn.lock().unwrap();
        let now = now_text();
        let link = ShareLinkRecord {
            id: input.id.unwrap_or_else(|| new_id("share")),
            workspace_id,
            resource_kind,
            resource_id,
            token_hash: token_hash.clone(),
            permission,
            created_by: input.created_by,
            expires_at: input.expires_at,
            revoked_at: None,
            created_at: now.clone(),
            updated_at: now.clone(),
            revision: 1,
        };
        conn.execute(
            "INSERT INTO share_links (id, workspace_id, resource_kind, resource_id, token_hash, permission, created_by, expires_at, revoked_at, created_at, updated_at, revision)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL, ?9, ?10, 1)",
            params![
                link.id,
                link.workspace_id,
                link.resource_kind,
                link.resource_id,
                link.token_hash,
                link.permission,
                link.created_by,
                link.expires_at,
                link.created_at,
                link.updated_at
            ],
        )?;
        Ok(CreatedShareLink { link, token })
    }

    fn list_share_links(
        &self,
        workspace_id: &str,
    ) -> Result<Vec<ShareLinkRecord>, ControlPlaneError> {
        let workspace_id = workspace_id.trim();
        if workspace_id.is_empty() {
            return Ok(Vec::new());
        }
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, workspace_id, resource_kind, resource_id, token_hash, permission, created_by, expires_at, revoked_at, created_at, updated_at, revision
             FROM share_links
             WHERE workspace_id = ?1
             ORDER BY created_at ASC",
        )?;
        let links = stmt
            .query_map(params![workspace_id], row_to_share_link)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(ControlPlaneError::from)?;
        Ok(links)
    }

    fn resolve_share_link(
        &self,
        token_hash: &str,
    ) -> Result<Option<ShareLinkRecord>, ControlPlaneError> {
        let token_hash = token_hash.trim();
        if token_hash.is_empty() {
            return Ok(None);
        }
        let conn = self.conn.lock().unwrap();
        let link = conn
            .query_row(
                "SELECT id, workspace_id, resource_kind, resource_id, token_hash, permission, created_by, expires_at, revoked_at, created_at, updated_at, revision
                 FROM share_links
                 WHERE token_hash = ?1 AND revoked_at IS NULL
                 LIMIT 1",
                params![token_hash],
                row_to_share_link,
            )
            .optional()?;
        Ok(link)
    }

    fn revoke_share_link(&self, link_id: &str) -> Result<(), ControlPlaneError> {
        let link_id = link_id.trim();
        if link_id.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "link_id 不能为空".to_string(),
            ));
        }
        let conn = self.conn.lock().unwrap();
        let existed = conn.execute(
            "UPDATE share_links
                 SET revoked_at = ?1, updated_at = ?2, revision = revision + 1
                 WHERE id = ?3 AND revoked_at IS NULL",
            params![now_text(), now_text(), link_id],
        )?;
        if existed == 0 {
            return Err(ControlPlaneError::NotFound(format!(
                "share link not found: {link_id}"
            )));
        }
        Ok(())
    }

    fn append_audit(&self, input: AppendAuditInput) -> Result<(), ControlPlaneError> {
        let action = input.action.trim();
        let target_kind = input.target_kind.trim();
        if action.is_empty() || target_kind.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "audit action/target_kind 不能为空".to_string(),
            ));
        }
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "INSERT INTO audit_log (id, actor_user_id, action, target_kind, target_id, metadata_json, created_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
            params![
                new_id("audit"),
                input.actor_user_id,
                action,
                target_kind,
                input.target_id,
                input.metadata_json,
                now_text()
            ],
        )?;
        Ok(())
    }

    fn list_audit_log(&self, limit: usize) -> Result<Vec<AuditLogRecord>, ControlPlaneError> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, actor_user_id, action, target_kind, target_id, metadata_json, created_at
             FROM audit_log
             ORDER BY created_at DESC
             LIMIT ?1",
        )?;
        let rows = stmt
            .query_map(params![limit as i64], row_to_audit_log)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(ControlPlaneError::from)?;
        Ok(rows)
    }

    fn append_outbox(
        &self,
        input: OutboxEventInput,
    ) -> Result<OutboxEventRecord, ControlPlaneError> {
        let conn = self.conn.lock().unwrap();
        let record = OutboxEventRecord {
            id: new_id("evt"),
            topic: input.topic,
            event_type: input.event_type,
            payload_json: input.payload_json,
            created_at: now_text(),
            delivered_at: None,
            attempts: 0,
        };
        conn.execute(
            "INSERT INTO outbox_events (id, topic, event_type, payload_json, created_at, delivered_at, attempts)
             VALUES (?1, ?2, ?3, ?4, ?5, NULL, 0)",
            params![
                record.id,
                record.topic,
                record.event_type,
                record.payload_json,
                record.created_at
            ],
        )?;
        Ok(record)
    }

    fn drain_outbox(&self, limit: usize) -> Result<Vec<OutboxEventRecord>, ControlPlaneError> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, topic, event_type, payload_json, created_at, delivered_at, attempts
             FROM outbox_events
             WHERE delivered_at IS NULL
             ORDER BY created_at ASC
             LIMIT ?1",
        )?;
        let rows = stmt
            .query_map(params![limit as i64], row_to_outbox)?
            .collect::<Result<Vec<_>, _>>()?;
        let delivered_at = now_text();
        for row in &rows {
            conn.execute(
                "UPDATE outbox_events SET delivered_at = ?1, attempts = attempts + 1 WHERE id = ?2",
                params![delivered_at, row.id],
            )?;
        }
        Ok(rows)
    }

    fn mark_outbox_delivered(&self, event_id: &str) -> Result<(), ControlPlaneError> {
        let event_id = event_id.trim();
        if event_id.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "event_id 不能为空".to_string(),
            ));
        }
        let conn = self.conn.lock().unwrap();
        let updated = conn.execute(
            "UPDATE outbox_events
             SET delivered_at = COALESCE(delivered_at, ?1), attempts = attempts + 1
             WHERE id = ?2",
            params![now_text(), event_id],
        )?;
        if updated == 0 {
            return Err(ControlPlaneError::NotFound(format!(
                "outbox event not found: {event_id}"
            )));
        }
        Ok(())
    }

    fn upsert_ai_policy(
        &self,
        input: UpsertAiPolicyInput,
    ) -> Result<AiPolicyRecord, ControlPlaneError> {
        if input.user_id.is_none() && input.workspace_id.is_none() {
            return Err(ControlPlaneError::InvalidInput(
                "ai_policy 必须绑定 user_id 或 workspace_id".to_string(),
            ));
        }
        serde_json::from_str::<serde_json::Value>(&input.allowed_roots_json)?;
        serde_json::from_str::<serde_json::Value>(&input.model_policy_json)?;
        serde_json::from_str::<serde_json::Value>(&input.quota_json)?;

        let conn = self.conn.lock().unwrap();
        let now = now_text();
        let existing = conn
            .query_row(
                "SELECT id, user_id, workspace_id, allowed_roots_json, model_policy_json, quota_json, created_at, updated_at, revision
                 FROM ai_policies
                 WHERE (?1 IS NULL OR user_id = ?1)
                   AND (?2 IS NULL OR workspace_id = ?2)
                 ORDER BY created_at DESC
                 LIMIT 1",
                params![input.user_id, input.workspace_id],
                row_to_ai_policy,
            )
            .optional()?;

        if let Some(existing) = existing {
            let revision = existing.revision + 1;
            conn.execute(
                "UPDATE ai_policies
                 SET allowed_roots_json = ?1, model_policy_json = ?2, quota_json = ?3, updated_at = ?4, revision = ?5
                 WHERE id = ?6",
                params![
                    input.allowed_roots_json,
                    input.model_policy_json,
                    input.quota_json,
                    now,
                    revision,
                    existing.id
                ],
            )?;
            return Ok(AiPolicyRecord {
                id: existing.id,
                user_id: existing.user_id,
                workspace_id: existing.workspace_id,
                allowed_roots_json: input.allowed_roots_json,
                model_policy_json: input.model_policy_json,
                quota_json: input.quota_json,
                created_at: existing.created_at,
                updated_at: now,
                revision,
            });
        }

        let record = AiPolicyRecord {
            id: input.id.unwrap_or_else(|| new_id("aip")),
            user_id: input.user_id,
            workspace_id: input.workspace_id,
            allowed_roots_json: input.allowed_roots_json,
            model_policy_json: input.model_policy_json,
            quota_json: input.quota_json,
            created_at: now.clone(),
            updated_at: now,
            revision: 1,
        };
        conn.execute(
            "INSERT INTO ai_policies (id, user_id, workspace_id, allowed_roots_json, model_policy_json, quota_json, created_at, updated_at, revision)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 1)",
            params![
                record.id,
                record.user_id,
                record.workspace_id,
                record.allowed_roots_json,
                record.model_policy_json,
                record.quota_json,
                record.created_at,
                record.updated_at
            ],
        )?;
        Ok(record)
    }

    fn get_ai_policy(
        &self,
        actor_id: &str,
        workspace_id: Option<&str>,
    ) -> Result<Option<AiPolicyRecord>, ControlPlaneError> {
        let actor_id = actor_id.trim();
        let workspace_id = workspace_id
            .map(str::trim)
            .filter(|value| !value.is_empty());
        if actor_id.is_empty() && workspace_id.is_none() {
            return Ok(None);
        }
        let conn = self.conn.lock().unwrap();
        if let Some(workspace_id) = workspace_id {
            let workspace_policy = conn
                .query_row(
                    "SELECT id, user_id, workspace_id, allowed_roots_json, model_policy_json, quota_json, created_at, updated_at, revision
                     FROM ai_policies
                     WHERE workspace_id = ?1
                     ORDER BY updated_at DESC
                     LIMIT 1",
                    params![workspace_id],
                    row_to_ai_policy,
                )
                .optional()?;
            if workspace_policy.is_some() {
                return Ok(workspace_policy);
            }
        }
        if actor_id.is_empty() {
            return Ok(None);
        }
        conn.query_row(
            "SELECT id, user_id, workspace_id, allowed_roots_json, model_policy_json, quota_json, created_at, updated_at, revision
             FROM ai_policies
             WHERE user_id = ?1 AND workspace_id IS NULL
             ORDER BY updated_at DESC
             LIMIT 1",
            params![actor_id],
            row_to_ai_policy,
        )
        .optional()
        .map_err(ControlPlaneError::from)
    }

    fn upsert_sync_state(
        &self,
        input: UpsertSyncStateInput,
    ) -> Result<SyncStateRecord, ControlPlaneError> {
        let workspace_id = input.workspace_id.trim().to_string();
        let remote_kind = input.remote_kind.trim().to_string();
        let status = input.status.trim().to_string();
        if workspace_id.is_empty() || remote_kind.is_empty() || status.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "sync_state workspace_id/remote_kind/status 不能为空".to_string(),
            ));
        }
        if let Some(error_json) = &input.error_json {
            serde_json::from_str::<serde_json::Value>(error_json)?;
        }

        let conn = self.conn.lock().unwrap();
        let now = now_text();
        let existing = conn
            .query_row(
                "SELECT id, workspace_id, remote_kind, remote_id, cursor, last_synced_at, status, error_json, updated_at, revision
                 FROM sync_state
                 WHERE workspace_id = ?1 AND remote_kind = ?2
                 LIMIT 1",
                params![workspace_id, remote_kind],
                row_to_sync_state,
            )
            .optional()?;
        if let Some(existing) = existing {
            let revision = existing.revision + 1;
            conn.execute(
                "UPDATE sync_state
                 SET remote_id = ?1, cursor = ?2, last_synced_at = ?3, status = ?4, error_json = ?5, updated_at = ?6, revision = ?7
                 WHERE id = ?8",
                params![
                    input.remote_id,
                    input.cursor,
                    input.last_synced_at,
                    status,
                    input.error_json,
                    now,
                    revision,
                    existing.id
                ],
            )?;
            return Ok(SyncStateRecord {
                id: existing.id,
                workspace_id: existing.workspace_id,
                remote_kind: existing.remote_kind,
                remote_id: input.remote_id,
                cursor: input.cursor,
                last_synced_at: input.last_synced_at,
                status,
                error_json: input.error_json,
                updated_at: now,
                revision,
            });
        }

        let record = SyncStateRecord {
            id: input.id.unwrap_or_else(|| new_id("sync")),
            workspace_id,
            remote_kind,
            remote_id: input.remote_id,
            cursor: input.cursor,
            last_synced_at: input.last_synced_at,
            status,
            error_json: input.error_json,
            updated_at: now,
            revision: 1,
        };
        conn.execute(
            "INSERT INTO sync_state (id, workspace_id, remote_kind, remote_id, cursor, last_synced_at, status, error_json, updated_at, revision)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 1)",
            params![
                record.id,
                record.workspace_id,
                record.remote_kind,
                record.remote_id,
                record.cursor,
                record.last_synced_at,
                record.status,
                record.error_json,
                record.updated_at
            ],
        )?;
        Ok(record)
    }

    fn get_sync_state(
        &self,
        workspace_id: &str,
        remote_kind: &str,
    ) -> Result<Option<SyncStateRecord>, ControlPlaneError> {
        let workspace_id = workspace_id.trim();
        let remote_kind = remote_kind.trim();
        if workspace_id.is_empty() || remote_kind.is_empty() {
            return Ok(None);
        }
        let conn = self.conn.lock().unwrap();
        conn.query_row(
            "SELECT id, workspace_id, remote_kind, remote_id, cursor, last_synced_at, status, error_json, updated_at, revision
             FROM sync_state
             WHERE workspace_id = ?1 AND remote_kind = ?2
             LIMIT 1",
            params![workspace_id, remote_kind],
            row_to_sync_state,
        )
        .optional()
        .map_err(ControlPlaneError::from)
    }

    fn upsert_ai_runtime_run(
        &self,
        input: UpsertAiRuntimeRunInput,
    ) -> Result<AiRuntimeRunRecord, ControlPlaneError> {
        if input.user_id.trim().is_empty()
            || input.session_id.trim().is_empty()
            || input.run_id.trim().is_empty()
        {
            return Err(ControlPlaneError::InvalidInput(
                "ai runtime run user_id/session_id/run_id 不能为空".to_string(),
            ));
        }
        serde_json::from_str::<serde_json::Value>(&input.runtime_json)?;
        serde_json::from_str::<serde_json::Value>(&input.payload_json)?;

        let conn = self.conn.lock().unwrap();
        let now = now_text();
        let existing = conn
            .query_row(
                "SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision
                 FROM ai_runtime_runs
                 WHERE run_id = ?1
                 LIMIT 1",
                params![input.run_id],
                row_to_ai_runtime_run,
            )
            .optional()?;

        if let Some(existing) = existing {
            let revision = existing.revision + 1;
            conn.execute(
                "UPDATE ai_runtime_runs
                 SET title = ?1, profile = ?2, acp_runtime = ?3, trace_id = ?4, status = ?5, runtime_json = ?6, payload_json = ?7, deleted_at = NULL, updated_at = ?8, revision = ?9
                 WHERE id = ?10",
                params![
                    input.title,
                    input.profile,
                    input.acp_runtime,
                    input.trace_id,
                    input.status,
                    input.runtime_json,
                    input.payload_json,
                    now,
                    revision,
                    existing.id
                ],
            )?;
            return Ok(AiRuntimeRunRecord {
                id: existing.id,
                user_id: existing.user_id,
                workspace_id: existing.workspace_id,
                document_id: existing.document_id,
                session_id: existing.session_id,
                run_id: existing.run_id,
                title: input.title,
                profile: input.profile,
                acp_runtime: input.acp_runtime,
                trace_id: input.trace_id,
                status: input.status,
                runtime_json: input.runtime_json,
                payload_json: input.payload_json,
                deleted_at: None,
                created_at: existing.created_at,
                updated_at: now,
                revision,
            });
        }

        let record = AiRuntimeRunRecord {
            id: input.id.unwrap_or_else(|| new_id("acr")),
            user_id: input.user_id,
            workspace_id: input.workspace_id,
            document_id: input.document_id,
            session_id: input.session_id,
            run_id: input.run_id,
            title: input.title,
            profile: input.profile,
            acp_runtime: input.acp_runtime,
            trace_id: input.trace_id,
            status: input.status,
            runtime_json: input.runtime_json,
            payload_json: input.payload_json,
            deleted_at: None,
            created_at: now.clone(),
            updated_at: now,
            revision: 1,
        };
        conn.execute(
            "INSERT INTO ai_runtime_runs (id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, NULL, ?14, ?15, 1)",
            params![
                record.id,
                record.user_id,
                record.workspace_id,
                record.document_id,
                record.session_id,
                record.run_id,
                record.title,
                record.profile,
                record.acp_runtime,
                record.trace_id,
                record.status,
                record.runtime_json,
                record.payload_json,
                record.created_at,
                record.updated_at
            ],
        )?;
        Ok(record)
    }

    fn list_ai_runtime_runs(
        &self,
        user_id: &str,
        workspace_id: Option<&str>,
        document_id: Option<&str>,
        session_id: Option<&str>,
        limit: usize,
    ) -> Result<Vec<AiRuntimeRunRecord>, ControlPlaneError> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision
             FROM ai_runtime_runs
             WHERE user_id = ?1
               AND (?2 IS NULL OR workspace_id = ?2)
               AND (?3 IS NULL OR document_id = ?3)
               AND (?4 IS NULL OR session_id = ?4)
               AND deleted_at IS NULL
             ORDER BY updated_at DESC
             LIMIT ?5",
        )?;
        let rows = stmt
            .query_map(
                params![user_id, workspace_id, document_id, session_id, limit as i64],
                row_to_ai_runtime_run,
            )?
            .collect::<Result<Vec<_>, _>>()
            .map_err(ControlPlaneError::from)?;
        Ok(rows)
    }

    fn append_ai_runtime_event(
        &self,
        input: AppendAiRuntimeEventInput,
    ) -> Result<AiRuntimeEventRecord, ControlPlaneError> {
        if input.user_id.trim().is_empty()
            || input.session_id.trim().is_empty()
            || input.run_id.trim().is_empty()
        {
            return Err(ControlPlaneError::InvalidInput(
                "ai runtime event user_id/session_id/run_id 不能为空".to_string(),
            ));
        }
        serde_json::from_str::<serde_json::Value>(&input.payload_json)?;

        let conn = self.conn.lock().unwrap();
        let record = AiRuntimeEventRecord {
            id: input.id.unwrap_or_else(|| new_id("are")),
            user_id: input.user_id,
            workspace_id: input.workspace_id,
            document_id: input.document_id,
            session_id: input.session_id,
            run_id: input.run_id,
            profile: input.profile,
            acp_runtime: input.acp_runtime,
            event_type: input.event_type,
            payload_json: input.payload_json,
            created_at: now_text(),
        };
        conn.execute(
            "INSERT INTO ai_runtime_events (id, user_id, workspace_id, document_id, session_id, run_id, profile, acp_runtime, event_type, payload_json, created_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
            params![
                record.id,
                record.user_id,
                record.workspace_id,
                record.document_id,
                record.session_id,
                record.run_id,
                record.profile,
                record.acp_runtime,
                record.event_type,
                record.payload_json,
                record.created_at
            ],
        )?;
        Ok(record)
    }

    fn list_ai_runtime_events(
        &self,
        user_id: &str,
        run_id: &str,
        limit: usize,
    ) -> Result<Vec<AiRuntimeEventRecord>, ControlPlaneError> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, user_id, workspace_id, document_id, session_id, run_id, profile, acp_runtime, event_type, payload_json, created_at
             FROM ai_runtime_events
             WHERE user_id = ?1 AND run_id = ?2
             ORDER BY created_at ASC
             LIMIT ?3",
        )?;
        let rows = stmt
            .query_map(
                params![user_id, run_id, limit as i64],
                row_to_ai_runtime_event,
            )?
            .collect::<Result<Vec<_>, _>>()
            .map_err(ControlPlaneError::from)?;
        Ok(rows)
    }

    fn rename_ai_runtime_session(
        &self,
        user_id: &str,
        session_id: &str,
        workspace_id: Option<&str>,
        title: &str,
    ) -> Result<Vec<AiRuntimeRunRecord>, ControlPlaneError> {
        let user_id = user_id.trim();
        let session_id = session_id.trim();
        let title = title.trim();
        if user_id.is_empty() || session_id.is_empty() || title.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "ai runtime session user_id/session_id/title 不能为空".to_string(),
            ));
        }
        let conn = self.conn.lock().unwrap();
        let now = now_text();
        conn.execute(
            "UPDATE ai_runtime_runs
             SET title = ?1, updated_at = ?2, revision = revision + 1
             WHERE user_id = ?3
               AND session_id = ?4
               AND (?5 IS NULL OR workspace_id = ?5)
               AND deleted_at IS NULL",
            params![title, now, user_id, session_id, workspace_id],
        )?;
        let mut stmt = conn.prepare(
            "SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision
             FROM ai_runtime_runs
             WHERE user_id = ?1
               AND session_id = ?2
               AND (?3 IS NULL OR workspace_id = ?3)
               AND deleted_at IS NULL
             ORDER BY updated_at DESC",
        )?;
        let rows = stmt
            .query_map(
                params![user_id, session_id, workspace_id],
                row_to_ai_runtime_run,
            )?
            .collect::<Result<Vec<_>, _>>()
            .map_err(ControlPlaneError::from)?;
        Ok(rows)
    }

    fn auto_title_ai_runtime_session(
        &self,
        user_id: &str,
        session_id: &str,
        workspace_id: Option<&str>,
    ) -> Result<Option<AiRuntimeRunRecord>, ControlPlaneError> {
        let user_id = user_id.trim();
        let session_id = session_id.trim();
        if user_id.is_empty() || session_id.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "ai runtime session user_id/session_id 不能为空".to_string(),
            ));
        }
        let conn = self.conn.lock().unwrap();
        let existing = conn
            .query_row(
                "SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision
                 FROM ai_runtime_runs
                 WHERE user_id = ?1
                   AND session_id = ?2
                   AND (?3 IS NULL OR workspace_id = ?3)
                   AND deleted_at IS NULL
                 ORDER BY updated_at DESC
                 LIMIT 1",
                params![user_id, session_id, workspace_id],
                row_to_ai_runtime_run,
            )
            .optional()?;
        let Some(existing) = existing else {
            return Ok(None);
        };
        let title = existing
            .title
            .clone()
            .filter(|value| !value.trim().is_empty())
            .unwrap_or_else(|| {
                derive_ai_runtime_title(&existing.payload_json, &existing.session_id)
            });
        let now = now_text();
        let revision = existing.revision + 1;
        conn.execute(
            "UPDATE ai_runtime_runs
             SET title = ?1, updated_at = ?2, revision = ?3
             WHERE id = ?4",
            params![title, now, revision, existing.id],
        )?;
        Ok(Some(AiRuntimeRunRecord {
            title: Some(title),
            updated_at: now,
            revision,
            ..existing
        }))
    }

    fn delete_ai_runtime_session(
        &self,
        user_id: &str,
        session_id: &str,
        workspace_id: Option<&str>,
    ) -> Result<usize, ControlPlaneError> {
        let user_id = user_id.trim();
        let session_id = session_id.trim();
        if user_id.is_empty() || session_id.is_empty() {
            return Err(ControlPlaneError::InvalidInput(
                "ai runtime session user_id/session_id 不能为空".to_string(),
            ));
        }
        let conn = self.conn.lock().unwrap();
        let changed = conn.execute(
            "UPDATE ai_runtime_runs
             SET deleted_at = ?1, updated_at = ?1, revision = revision + 1
             WHERE user_id = ?2
               AND session_id = ?3
               AND (?4 IS NULL OR workspace_id = ?4)
               AND deleted_at IS NULL",
            params![now_text(), user_id, session_id, workspace_id],
        )?;
        Ok(changed)
    }
}

fn max_permission<'a>(left: &'a str, right: &'a str) -> &'a str {
    if permission_rank(right) > permission_rank(left) {
        right
    } else {
        left
    }
}

fn permission_rank(permission: &str) -> u8 {
    match permission {
        "admin" => 3,
        "write" => 2,
        "read" => 1,
        _ => 0,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::model::{
        password_hash_v1, session_token_hash, AppendAiRuntimeEventInput, AuthenticatePasswordInput,
        CreatePasswordIdentityInput, UpsertAiPolicyInput, UpsertAiRuntimeRunInput,
        UpsertSyncStateInput,
    };

    fn store() -> SqliteControlPlaneStore {
        SqliteControlPlaneStore::in_memory().expect("in-memory control plane")
    }

    fn create_user(store: &SqliteControlPlaneStore, id: &str) -> UserRecord {
        store
            .upsert_user(UpsertUserInput {
                id: Some(id.to_string()),
                email: Some(format!("{id}@example.com")),
                username: id.to_string(),
                display_name: id.to_string(),
                role: None,
                password_hash: None,
            })
            .expect("upsert user")
    }

    fn create_password_identity(
        store: &SqliteControlPlaneStore,
        username: &str,
        email: &str,
        password: &str,
    ) -> UserRecord {
        let user = create_user(store, username);
        store
            .create_password_identity(CreatePasswordIdentityInput {
                user_id: user.id.clone(),
                email: Some(email.to_string()),
                username: username.to_string(),
                password: password.to_string(),
            })
            .expect("create password identity");
        user
    }

    #[test]
    fn upsert_user_inserts_and_updates_revision() {
        let store = store();
        let user = create_user(&store, "alice");
        assert_eq!(user.revision, 1);

        let updated = store
            .upsert_user(UpsertUserInput {
                id: Some("alice".to_string()),
                email: Some("alice2@example.com".to_string()),
                username: "alice".to_string(),
                display_name: "Alice".to_string(),
                role: Some("admin".to_string()),
                password_hash: None,
            })
            .expect("update user");

        assert_eq!(updated.email.as_deref(), Some("alice2@example.com"));
        assert_eq!(updated.role, "admin");
        assert_eq!(updated.revision, 2);
    }

    #[test]
    fn ensure_default_workspace_creates_owner_grant() {
        let store = store();
        create_user(&store, "shujuan");

        let workspace = store
            .ensure_default_workspace("shujuan")
            .expect("default workspace");
        assert_eq!(workspace.name, "shujuan 的空间");

        let access = store
            .resolve_access("shujuan", &workspace.root_uri)
            .expect("resolve access");
        assert_eq!(access.permission, "write");
        assert_eq!(access.grant_ids.len(), 1);
    }

    #[test]
    fn explicit_write_grant_out_ranks_read_grant() {
        let store = store();
        create_user(&store, "bob");
        store
            .grant_directory_access(DirectoryGrantInput {
                user_id: "bob".to_string(),
                workspace_id: None,
                root_uri: "local://shared".to_string(),
                root_path: "/tmp/shared".to_string(),
                permission: "read".to_string(),
                recursive: true,
                capabilities: vec![],
                source: "admin".to_string(),
                created_by: None,
            })
            .expect("read grant");
        store
            .grant_directory_access(DirectoryGrantInput {
                user_id: "bob".to_string(),
                workspace_id: None,
                root_uri: "local://shared".to_string(),
                root_path: "/tmp/shared".to_string(),
                permission: "write".to_string(),
                recursive: true,
                capabilities: vec!["ai".to_string()],
                source: "admin".to_string(),
                created_by: None,
            })
            .expect("write grant");

        let access = store
            .resolve_access("bob", "local://shared/page.md")
            .expect("resolve access");
        assert_eq!(access.permission, "write");
        assert_eq!(access.grant_ids.len(), 2);
    }

    #[test]
    fn list_and_revoke_directory_grant_updates_resolved_access() {
        let store = store();
        create_user(&store, "reader");
        let grant = store
            .grant_directory_access(DirectoryGrantInput {
                user_id: "reader".to_string(),
                workspace_id: None,
                root_uri: "local://shared".to_string(),
                root_path: "/tmp/shared".to_string(),
                permission: "read".to_string(),
                recursive: true,
                capabilities: vec![],
                source: "admin".to_string(),
                created_by: None,
            })
            .expect("read grant");

        let grants = store
            .list_directory_grants_for_actor("reader")
            .expect("list grants");
        assert_eq!(grants.len(), 1);
        assert_eq!(grants[0].id, grant.id);

        store
            .revoke_directory_grant(&grant.id, None)
            .expect("revoke grant");
        let access = store
            .resolve_access("reader", "local://shared/page.md")
            .expect("resolve revoked access");
        assert_eq!(access.permission, "none");
    }

    #[test]
    fn resolve_access_supports_legacy_directory_grants_with_plain_path_root_uri() {
        let store = store();
        create_user(&store, "legacy_reader");
        store
            .grant_directory_access(DirectoryGrantInput {
                user_id: "legacy_reader".to_string(),
                workspace_id: None,
                root_uri: "/tmp/shared".to_string(),
                root_path: "/tmp/shared".to_string(),
                permission: "read".to_string(),
                recursive: true,
                capabilities: vec![],
                source: "legacy".to_string(),
                created_by: None,
            })
            .expect("legacy grant");

        let access = store
            .resolve_access("legacy_reader", "file:///tmp/shared/page.md")
            .expect("resolve access");
        assert_eq!(access.permission, "read");
        assert_eq!(access.grant_ids.len(), 1);
    }

    #[test]
    fn share_link_token_is_hashed_and_revocable() {
        let store = store();
        create_user(&store, "owner");
        create_user(&store, "ws_owner");
        let workspace = store
            .ensure_default_workspace("ws_owner")
            .expect("workspace");
        let created = store
            .create_share_link(CreateShareLinkInput {
                id: None,
                workspace_id: workspace.id,
                resource_kind: "page".to_string(),
                resource_id: "page_1".to_string(),
                token: Some("plain-token".to_string()),
                permission: "read".to_string(),
                created_by: "ws_owner".to_string(),
                expires_at: None,
            })
            .expect("create share link");

        assert_eq!(created.token, "plain-token");
        assert!(created.link.token_hash.starts_with("sha256-v1:"));
        assert_ne!(created.link.token_hash, "plain-token");

        let resolved = store
            .resolve_share_link(&created.link.token_hash)
            .expect("resolve share link")
            .expect("active share link");
        assert_eq!(resolved.id, created.link.id);

        let listed = store
            .list_share_links(&created.link.workspace_id)
            .expect("list links");
        assert_eq!(listed.len(), 1);

        store
            .revoke_share_link(&created.link.id)
            .expect("revoke share");
        assert!(store
            .resolve_share_link(&created.link.token_hash)
            .expect("resolve revoked share")
            .is_none());
    }

    #[test]
    fn audit_log_appends_and_lists_latest_first() {
        let store = store();
        create_user(&store, "owner");
        store
            .append_audit(AppendAuditInput {
                actor_user_id: Some("owner".to_string()),
                action: "control.share.created".to_string(),
                target_kind: "share_link".to_string(),
                target_id: Some("share_1".to_string()),
                metadata_json: "{}".to_string(),
            })
            .expect("append audit");
        store
            .append_audit(AppendAuditInput {
                actor_user_id: Some("owner".to_string()),
                action: "control.share.revoked".to_string(),
                target_kind: "share_link".to_string(),
                target_id: Some("share_1".to_string()),
                metadata_json: "{\"reason\":\"test\"}".to_string(),
            })
            .expect("append audit 2");

        let rows = store.list_audit_log(10).expect("list audit");
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0].action, "control.share.revoked");
        assert_eq!(rows[1].action, "control.share.created");
    }

    #[test]
    fn drain_outbox_marks_events_delivered() {
        let store = store();
        store
            .append_outbox(OutboxEventInput {
                topic: "control".to_string(),
                event_type: "control.workspace.updated".to_string(),
                payload_json: "{}".to_string(),
            })
            .expect("append event");

        let first = store.drain_outbox(10).expect("first drain");
        assert_eq!(first.len(), 1);

        let second = store.drain_outbox(10).expect("second drain");
        assert!(second.is_empty());
    }

    #[test]
    fn mark_outbox_delivered_marks_one_pending_event() {
        let store = store();
        let first = store
            .append_outbox(OutboxEventInput {
                topic: "control".to_string(),
                event_type: "control.grant.created".to_string(),
                payload_json: "{}".to_string(),
            })
            .expect("first event");
        let second = store
            .append_outbox(OutboxEventInput {
                topic: "control".to_string(),
                event_type: "control.share.created".to_string(),
                payload_json: "{}".to_string(),
            })
            .expect("second event");

        store
            .mark_outbox_delivered(&first.id)
            .expect("mark first delivered");

        let pending = store.drain_outbox(10).expect("drain pending");
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].id, second.id);
    }

    #[test]
    fn ai_policy_upsert_returns_workspace_policy_before_user_policy() {
        let store = store();
        create_user(&store, "ai_user");
        let workspace = store
            .ensure_default_workspace("ai_user")
            .expect("default workspace");

        let user_policy = store
            .upsert_ai_policy(UpsertAiPolicyInput {
                id: None,
                user_id: Some("ai_user".to_string()),
                workspace_id: None,
                allowed_roots_json: "[\"file:///user-root\"]".to_string(),
                model_policy_json: "{\"default\":\"local\"}".to_string(),
                quota_json: "{}".to_string(),
            })
            .expect("user policy");
        assert_eq!(user_policy.revision, 1);

        let workspace_policy = store
            .upsert_ai_policy(UpsertAiPolicyInput {
                id: None,
                user_id: Some("ai_user".to_string()),
                workspace_id: Some(workspace.id.clone()),
                allowed_roots_json: "[\"file:///workspace-root\"]".to_string(),
                model_policy_json: "{\"default\":\"workspace\"}".to_string(),
                quota_json: "{}".to_string(),
            })
            .expect("workspace policy");

        let resolved = store
            .get_ai_policy("ai_user", Some(&workspace.id))
            .expect("get workspace policy")
            .expect("workspace policy exists");
        assert_eq!(resolved.id, workspace_policy.id);
        assert!(resolved.allowed_roots_json.contains("workspace-root"));

        let user_only = store
            .get_ai_policy("ai_user", None)
            .expect("get user policy")
            .expect("user policy exists");
        assert_eq!(user_only.id, user_policy.id);
    }

    #[test]
    fn sync_state_upsert_tracks_cursor_status_and_revision() {
        let store = store();
        create_user(&store, "sync_user");
        let workspace = store
            .ensure_default_workspace("sync_user")
            .expect("default workspace");

        let first = store
            .upsert_sync_state(UpsertSyncStateInput {
                id: None,
                workspace_id: workspace.id.clone(),
                remote_kind: "local_worker".to_string(),
                remote_id: Some("worker-1".to_string()),
                cursor: Some("cursor-1".to_string()),
                last_synced_at: None,
                status: "idle".to_string(),
                error_json: None,
            })
            .expect("first sync state");
        assert_eq!(first.revision, 1);

        let second = store
            .upsert_sync_state(UpsertSyncStateInput {
                id: None,
                workspace_id: workspace.id.clone(),
                remote_kind: "local_worker".to_string(),
                remote_id: Some("worker-1".to_string()),
                cursor: Some("cursor-2".to_string()),
                last_synced_at: Some("2026-05-22T00:00:00Z".to_string()),
                status: "error".to_string(),
                error_json: Some("{\"code\":\"test\"}".to_string()),
            })
            .expect("second sync state");

        assert_eq!(second.id, first.id);
        assert_eq!(second.revision, 2);
        assert_eq!(second.cursor.as_deref(), Some("cursor-2"));

        let loaded = store
            .get_sync_state(&workspace.id, "local_worker")
            .expect("get sync state")
            .expect("sync state exists");
        assert_eq!(loaded.status, "error");
        assert_eq!(loaded.error_json.as_deref(), Some("{\"code\":\"test\"}"));
    }

    #[test]
    fn ai_runtime_run_upsert_lists_updates_and_events() {
        let store = store();
        create_user(&store, "ai_runtime_user");

        let first = store
            .upsert_ai_runtime_run(UpsertAiRuntimeRunInput {
                id: None,
                user_id: "ai_runtime_user".to_string(),
                workspace_id: Some("ws_1".to_string()),
                document_id: Some("doc_1".to_string()),
                session_id: "sess_1".to_string(),
                run_id: "run_1".to_string(),
                title: Some("初始标题".to_string()),
                profile: "reasonix".to_string(),
                acp_runtime: "reasonix".to_string(),
                trace_id: Some("trace_1".to_string()),
                status: "running".to_string(),
                runtime_json: "{\"status\":\"running\"}".to_string(),
                payload_json: "{\"message\":\"读取当前页面\"}".to_string(),
            })
            .expect("insert runtime run");
        assert_eq!(first.revision, 1);

        let updated = store
            .upsert_ai_runtime_run(UpsertAiRuntimeRunInput {
                id: None,
                user_id: "ai_runtime_user".to_string(),
                workspace_id: Some("ws_1".to_string()),
                document_id: Some("doc_1".to_string()),
                session_id: "sess_1".to_string(),
                run_id: "run_1".to_string(),
                title: Some("更新标题".to_string()),
                profile: "reasonix".to_string(),
                acp_runtime: "reasonix".to_string(),
                trace_id: Some("trace_2".to_string()),
                status: "completed".to_string(),
                runtime_json: "{\"status\":\"completed\"}".to_string(),
                payload_json: "{\"message\":\"完成\"}".to_string(),
            })
            .expect("update runtime run");
        assert_eq!(updated.id, first.id);
        assert_eq!(updated.revision, 2);

        let runs = store
            .list_ai_runtime_runs(
                "ai_runtime_user",
                Some("ws_1"),
                Some("doc_1"),
                Some("sess_1"),
                10,
            )
            .expect("list runtime runs");
        assert_eq!(runs.len(), 1);
        assert_eq!(runs[0].title.as_deref(), Some("更新标题"));

        let event = store
            .append_ai_runtime_event(AppendAiRuntimeEventInput {
                id: None,
                user_id: "ai_runtime_user".to_string(),
                workspace_id: Some("ws_1".to_string()),
                document_id: Some("doc_1".to_string()),
                session_id: "sess_1".to_string(),
                run_id: "run_1".to_string(),
                profile: "reasonix".to_string(),
                acp_runtime: "reasonix".to_string(),
                event_type: "message.delta".to_string(),
                payload_json: "{\"text\":\"hello\"}".to_string(),
            })
            .expect("append runtime event");
        assert_eq!(event.event_type, "message.delta");

        let events = store
            .list_ai_runtime_events("ai_runtime_user", "run_1", 10)
            .expect("list runtime events");
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].payload_json, "{\"text\":\"hello\"}");
    }

    #[test]
    fn ai_runtime_session_management_renames_auto_titles_and_soft_deletes() {
        let store = store();
        create_user(&store, "ai_runtime_user");
        store
            .upsert_ai_runtime_run(UpsertAiRuntimeRunInput {
                id: None,
                user_id: "ai_runtime_user".to_string(),
                workspace_id: Some("ws_1".to_string()),
                document_id: Some("doc_1".to_string()),
                session_id: "sess_1".to_string(),
                run_id: "run_1".to_string(),
                title: None,
                profile: "reasonix".to_string(),
                acp_runtime: "reasonix".to_string(),
                trace_id: None,
                status: "completed".to_string(),
                runtime_json: "{\"status\":\"completed\"}".to_string(),
                payload_json: "{\"message\":\"请总结这篇文档\"}".to_string(),
            })
            .expect("insert runtime run");

        let titled = store
            .auto_title_ai_runtime_session("ai_runtime_user", "sess_1", Some("ws_1"))
            .expect("auto title")
            .expect("titled run");
        assert_eq!(titled.title.as_deref(), Some("请总结这篇文档"));

        let renamed = store
            .rename_ai_runtime_session("ai_runtime_user", "sess_1", Some("ws_1"), "人工标题")
            .expect("rename session");
        assert_eq!(renamed.len(), 1);
        assert_eq!(renamed[0].title.as_deref(), Some("人工标题"));

        let deleted = store
            .delete_ai_runtime_session("ai_runtime_user", "sess_1", Some("ws_1"))
            .expect("delete session");
        assert_eq!(deleted, 1);

        let runs = store
            .list_ai_runtime_runs("ai_runtime_user", Some("ws_1"), None, Some("sess_1"), 10)
            .expect("list after delete");
        assert!(runs.is_empty());
    }

    #[test]
    fn session_lookup_resolves_active_user_by_token_hash() {
        let store = store();
        create_user(&store, "shujuan");

        let session = store
            .create_session(CreateSessionInput {
                id: None,
                user_id: "shujuan".to_string(),
                token_hash: session_token_hash("raw-session-token"),
                user_agent: Some("mnote-test".to_string()),
                ip_hash: None,
                expires_at: None,
            })
            .expect("create session");

        let resolved = store
            .get_session_by_token_hash(&session.token_hash)
            .expect("lookup session")
            .expect("active session");

        assert_eq!(resolved.session.user_id, "shujuan");
        assert_eq!(resolved.user.id, "shujuan");
        assert_eq!(resolved.user.email.as_deref(), Some("shujuan@example.com"));
    }

    #[test]
    fn revoked_session_is_not_resolved() {
        let store = store();
        create_user(&store, "shujuan");

        let session = store
            .create_session(CreateSessionInput {
                id: None,
                user_id: "shujuan".to_string(),
                token_hash: session_token_hash("revoked-token"),
                user_agent: None,
                ip_hash: None,
                expires_at: None,
            })
            .expect("create session");
        store.revoke_session(&session.id).expect("revoke session");

        let resolved = store
            .get_session_by_token_hash(&session.token_hash)
            .expect("lookup session");
        assert!(resolved.is_none());
    }

    #[test]
    fn password_hash_v1_is_sha256_hex_placeholder() {
        assert!(password_hash_v1("secret").starts_with("sha256-v1:"));
    }

    #[test]
    fn authenticate_password_accepts_email_login_and_creates_session() {
        let store = store();
        let test_password = ["correct", "horse"].join(" ");
        create_password_identity(&store, "alice", "alice@example.com", &test_password);

        let resolved = store
            .authenticate_password(AuthenticatePasswordInput {
                account: "alice@example.com".to_string(),
                password: test_password,
                session_id: None,
                token_hash: session_token_hash("alice-email-session"),
                user_agent: Some("mnote-test".to_string()),
                ip_hash: None,
                expires_at: None,
            })
            .expect("authenticate by email");

        assert_eq!(resolved.user.username, "alice");
        assert_eq!(
            resolved.session.token_hash,
            session_token_hash("alice-email-session")
        );

        let lookup = store
            .get_session_by_token_hash(&resolved.session.token_hash)
            .expect("lookup created session")
            .expect("created session");
        assert_eq!(lookup.user.id, resolved.user.id);
    }

    #[test]
    fn authenticate_password_accepts_username_login() {
        let store = store();
        let test_password = ["sword", "fish"].join("");
        create_password_identity(&store, "bob", "bob@example.com", &test_password);

        let resolved = store
            .authenticate_password(AuthenticatePasswordInput {
                account: "bob".to_string(),
                password: test_password,
                session_id: None,
                token_hash: session_token_hash("bob-username-session"),
                user_agent: None,
                ip_hash: None,
                expires_at: None,
            })
            .expect("authenticate by username");

        assert_eq!(resolved.user.email.as_deref(), Some("bob@example.com"));
        assert_eq!(
            resolved.session.token_hash,
            session_token_hash("bob-username-session")
        );
    }

    #[test]
    fn authenticate_password_rejects_wrong_password() {
        let store = store();
        let right_password = ["right", "password"].join("-");
        let wrong_password = ["wrong", "password"].join("-");
        create_password_identity(&store, "chris", "chris@example.com", &right_password);

        let err = store
            .authenticate_password(AuthenticatePasswordInput {
                account: "chris@example.com".to_string(),
                password: wrong_password,
                session_id: None,
                token_hash: session_token_hash("chris-failed-session"),
                user_agent: None,
                ip_hash: None,
                expires_at: None,
            })
            .expect_err("wrong password should fail");

        assert!(matches!(err, ControlPlaneError::Unauthorized(_)));
        assert!(store
            .get_session_by_token_hash(&session_token_hash("chris-failed-session"))
            .expect("lookup failed session")
            .is_none());
    }
}