kcode-telegram-identity 0.1.0

//! Persistent Telegram identity authorization and Kennedy Kmap-root directory.

use std::{collections::HashSet, path::Path, sync::Mutex};

use chrono::Utc;
use kcode_kweb_db::NodeId;
use kcode_tg_kennedy_bot::{AddUserOutcome, IdentityObservation, IdentitySink, WhitelistSnapshot};
use rusqlite::{Connection, OptionalExtension, params};
use serde::{Deserialize, Serialize};

const IDENTITY_MIGRATION: &str = include_str!("../migrations/001_initial.sql");

/// Durable identity authorization and user/group root mappings.
pub struct Directory {
    database: Mutex<Connection>,
}

/// One whitelisted Telegram identity and its Kennedy root assignment.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct User {
    pub handle: String,
    pub telegram_user_id: Option<i64>,
    pub current_username: Option<String>,
    pub display_name: Option<String>,
    pub root_node_id: Option<String>,
    pub root_ready: bool,
    pub can_add_users: bool,
}

/// One opaque Telegram group and its Kennedy root assignment.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Group {
    pub group_id: String,
    pub root_node_id: Option<String>,
    pub root_ready: bool,
}

/// Stable categories callers can map to their own transport errors.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ErrorKind {
    InvalidInput,
    NotFound,
    Conflict,
    Storage,
}

/// A typed identity-directory failure.
#[derive(Debug, thiserror::Error)]
#[error("{message}")]
pub struct Error {
    kind: ErrorKind,
    message: String,
}

impl Error {
    pub fn kind(&self) -> ErrorKind {
        self.kind
    }

    pub fn message(&self) -> &str {
        &self.message
    }

    fn invalid(message: impl Into<String>) -> Self {
        Self {
            kind: ErrorKind::InvalidInput,
            message: message.into(),
        }
    }

    fn not_found() -> Self {
        Self {
            kind: ErrorKind::NotFound,
            message: "Telegram directory entry not found.".into(),
        }
    }

    fn conflict(message: impl Into<String>) -> Self {
        Self {
            kind: ErrorKind::Conflict,
            message: message.into(),
        }
    }

    fn storage(error: impl std::fmt::Display) -> Self {
        Self {
            kind: ErrorKind::Storage,
            message: error.to_string(),
        }
    }
}

pub type Result<T> = std::result::Result<T, Error>;

impl Directory {
    /// Open the directory and ensure the normalized bootstrap handle exists.
    pub fn open(path: &Path, bootstrap_handle: &str) -> Result<Self> {
        let connection = Connection::open(path)
            .map_err(|error| Error::storage(format!("opening {}: {error}", path.display())))?;
        connection
            .execute_batch(
                "PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000; PRAGMA foreign_keys=ON;",
            )
            .map_err(Error::storage)?;
        connection
            .execute_batch(IDENTITY_MIGRATION)
            .map_err(Error::storage)?;
        let directory = Self {
            database: Mutex::new(connection),
        };
        directory.seed_bootstrap_user(bootstrap_handle)?;
        Ok(directory)
    }

    fn seed_bootstrap_user(&self, handle: &str) -> Result<()> {
        let handle = normalize_username(handle);
        if handle.is_empty() {
            return Err(Error::invalid(
                "Telegram bootstrap handle must not be empty",
            ));
        }
        let database = self.lock()?;
        let now = Utc::now().to_rfc3339();
        database
            .execute(
                "INSERT INTO whitelist_entries(handle,can_add_users,whitelisted_at,updated_at)
                 VALUES(?1,1,?2,?2)
                 ON CONFLICT(handle) DO UPDATE SET can_add_users=1,updated_at=excluded.updated_at",
                params![handle, now],
            )
            .map_err(Error::storage)?;
        Ok(())
    }

    fn lock(&self) -> Result<std::sync::MutexGuard<'_, Connection>> {
        self.database
            .lock()
            .map_err(|_| Error::storage("locking Telegram identity directory"))
    }

    /// List whitelisted users whose Kweb root assignment is incomplete.
    pub fn provisioning_users(&self) -> Result<Vec<User>> {
        let database = self.lock()?;
        let mut statement = database
            .prepare(
                "SELECT handle,telegram_user_id,current_username,display_name,root_node_id,root_ready,can_add_users
                 FROM whitelist_entries WHERE root_ready=0 ORDER BY whitelisted_at,handle",
            )
            .map_err(Error::storage)?;
        statement
            .query_map([], row_user)
            .map_err(Error::storage)?
            .collect::<std::result::Result<Vec<_>, _>>()
            .map_err(Error::storage)
    }

    /// List observed groups whose Kweb root assignment is incomplete.
    pub fn provisioning_groups(&self) -> Result<Vec<Group>> {
        let database = self.lock()?;
        let mut statement = database
            .prepare(
                "SELECT group_id,root_node_id,root_ready FROM telegram_group_roots
                 WHERE root_ready=0 ORDER BY datetime(created_at),group_id",
            )
            .map_err(Error::storage)?;
        statement
            .query_map([], row_group)
            .map_err(Error::storage)?
            .collect::<std::result::Result<Vec<_>, _>>()
            .map_err(Error::storage)
    }

    /// Look up one whitelisted identity by its stable numeric Telegram ID.
    pub fn user(&self, telegram_user_id: i64) -> Result<User> {
        let database = self.lock()?;
        directory_user_by_id(&database, telegram_user_id)?.ok_or_else(Error::not_found)
    }

    /// Look up one previously observed opaque Telegram group.
    pub fn group(&self, group_id: &str) -> Result<Group> {
        let database = self.lock()?;
        directory_group_by_id(&database, group_id)?.ok_or_else(Error::not_found)
    }

    /// Complete the root assignment for a preauthorized handle.
    pub fn complete_handle_root(&self, handle: &str, root_node_id: NodeId) -> Result<User> {
        let handle = normalize_username(handle);
        let database = self.lock()?;
        let current = directory_user_by_handle(&database, &handle)?.ok_or_else(Error::not_found)?;
        ensure_user_root_compatible(&current, root_node_id, "whitelisted handle")?;
        database
            .execute(
                "UPDATE whitelist_entries SET root_node_id=?1,root_ready=1,updated_at=?2
                 WHERE handle=?3",
                params![root_node_id.to_string(), Utc::now().to_rfc3339(), handle],
            )
            .map_err(Error::storage)?;
        directory_user_by_handle(&database, &handle)?.ok_or_else(Error::not_found)
    }

    /// Complete the root assignment for a bound numeric Telegram identity.
    pub fn complete_user_root(&self, telegram_user_id: i64, root_node_id: NodeId) -> Result<User> {
        let database = self.lock()?;
        let current =
            directory_user_by_id(&database, telegram_user_id)?.ok_or_else(Error::not_found)?;
        ensure_user_root_compatible(&current, root_node_id, "Telegram identity")?;
        database
            .execute(
                "UPDATE whitelist_entries SET root_node_id=?1,root_ready=1,updated_at=?2
                 WHERE telegram_user_id=?3",
                params![
                    root_node_id.to_string(),
                    Utc::now().to_rfc3339(),
                    telegram_user_id
                ],
            )
            .map_err(Error::storage)?;
        directory_user_by_id(&database, telegram_user_id)?.ok_or_else(Error::not_found)
    }

    /// Complete the root assignment for an observed opaque Telegram group.
    pub fn complete_group_root(&self, group_id: &str, root_node_id: NodeId) -> Result<Group> {
        let database = self.lock()?;
        let current = directory_group_by_id(&database, group_id)?.ok_or_else(Error::not_found)?;
        let root_node_id = root_node_id.to_string();
        if current.root_ready && current.root_node_id.as_deref() != Some(&root_node_id) {
            return Err(Error::conflict(
                "This Telegram group already has a different root node.",
            ));
        }
        database
            .execute(
                "UPDATE telegram_group_roots SET root_node_id=?1,root_ready=1,updated_at=?2
                 WHERE group_id=?3",
                params![root_node_id, Utc::now().to_rfc3339(), group_id],
            )
            .map_err(Error::storage)?;
        directory_group_by_id(&database, group_id)?.ok_or_else(Error::not_found)
    }
}

impl IdentitySink for Directory {
    fn observe_identity(&self, observation: &IdentityObservation) -> anyhow::Result<()> {
        let database = self.lock()?;
        observe_identity(&database, observation)?;
        Ok(())
    }

    fn whitelist(&self) -> anyhow::Result<WhitelistSnapshot> {
        let database = self.lock()?;
        let telegram_user_ids = database
            .prepare(
                "SELECT telegram_user_id FROM whitelist_entries
                 WHERE telegram_user_id IS NOT NULL ORDER BY telegram_user_id",
            )?
            .query_map([], |row| row.get::<_, i64>(0))?
            .collect::<std::result::Result<HashSet<_>, _>>()?;
        Ok(WhitelistSnapshot { telegram_user_ids })
    }

    fn request_add_user(
        &self,
        requested_by_telegram_user_id: i64,
        handle: &str,
    ) -> anyhow::Result<AddUserOutcome> {
        let database = self.lock()?;
        let can_add = directory_user_by_id(&database, requested_by_telegram_user_id)?
            .is_some_and(|user| user.can_add_users);
        if !can_add {
            return Ok(AddUserOutcome::Forbidden);
        }
        let user = whitelist_handle(&database, handle, requested_by_telegram_user_id)?;
        Ok(AddUserOutcome::Whitelisted {
            handle: user.handle,
            telegram_user_id: user.telegram_user_id,
        })
    }

    fn observe_group(&self, group_id: &str) -> anyhow::Result<()> {
        let database = self.lock()?;
        let now = Utc::now().to_rfc3339();
        database.execute(
            "INSERT INTO telegram_group_roots(group_id,created_at,updated_at)
             VALUES(?1,?2,?2) ON CONFLICT(group_id) DO NOTHING",
            params![group_id, now],
        )?;
        Ok(())
    }
}

fn normalize_username(value: &str) -> String {
    value.trim().trim_start_matches('@').to_ascii_lowercase()
}

fn directory_user_by_clause(
    database: &Connection,
    clause: &str,
    value: &dyn rusqlite::ToSql,
) -> Result<Option<User>> {
    database
        .query_row(
            &format!(
                "SELECT handle,telegram_user_id,current_username,display_name,root_node_id,root_ready,can_add_users
                 FROM whitelist_entries WHERE {clause}"
            ),
            [value],
            row_user,
        )
        .optional()
        .map_err(Error::storage)
}

fn directory_user_by_id(database: &Connection, telegram_user_id: i64) -> Result<Option<User>> {
    directory_user_by_clause(database, "telegram_user_id=?1", &telegram_user_id)
}

fn directory_user_by_handle(database: &Connection, handle: &str) -> Result<Option<User>> {
    directory_user_by_clause(database, "handle=?1", &handle)
}

fn directory_group_by_id(database: &Connection, group_id: &str) -> Result<Option<Group>> {
    database
        .query_row(
            "SELECT group_id,root_node_id,root_ready FROM telegram_group_roots WHERE group_id=?1",
            [group_id],
            row_group,
        )
        .optional()
        .map_err(Error::storage)
}

fn row_user(row: &rusqlite::Row<'_>) -> rusqlite::Result<User> {
    let root_node_id = canonical_root(row.get(4)?, 4)?;
    Ok(User {
        handle: row.get(0)?,
        telegram_user_id: row.get(1)?,
        current_username: row.get(2)?,
        display_name: row.get(3)?,
        root_node_id,
        root_ready: row.get::<_, i64>(5)? != 0,
        can_add_users: row.get::<_, i64>(6)? != 0,
    })
}

fn row_group(row: &rusqlite::Row<'_>) -> rusqlite::Result<Group> {
    let root_node_id = canonical_root(row.get(1)?, 1)?;
    Ok(Group {
        group_id: row.get(0)?,
        root_node_id,
        root_ready: row.get::<_, i64>(2)? != 0,
    })
}

fn canonical_root(value: Option<String>, column: usize) -> rusqlite::Result<Option<String>> {
    value
        .map(|value| {
            value
                .parse::<NodeId>()
                .map(|id| id.to_string())
                .map_err(|error| {
                    rusqlite::Error::FromSqlConversionFailure(
                        column,
                        rusqlite::types::Type::Text,
                        Box::new(error),
                    )
                })
        })
        .transpose()
}

fn observe_identity(database: &Connection, observation: &IdentityObservation) -> Result<()> {
    let now = Utc::now().to_rfc3339();
    let normalized = observation
        .username
        .as_deref()
        .map(normalize_username)
        .filter(|value| !value.is_empty());
    database
        .execute(
            "INSERT INTO observed_identities(telegram_user_id,current_username,display_name,first_seen_at,last_seen_at)
             VALUES(?1,?2,?3,?4,?4)
             ON CONFLICT(telegram_user_id) DO UPDATE SET
                 current_username=excluded.current_username,
                 display_name=excluded.display_name,last_seen_at=excluded.last_seen_at",
            params![
                observation.telegram_user_id,
                normalized,
                observation.display_name,
                now
            ],
        )
        .map_err(Error::storage)?;
    if directory_user_by_id(database, observation.telegram_user_id)?.is_some() {
        database
            .execute(
                "UPDATE whitelist_entries SET current_username=?1,display_name=?2,updated_at=?3
                 WHERE telegram_user_id=?4",
                params![
                    normalized,
                    observation.display_name,
                    now,
                    observation.telegram_user_id
                ],
            )
            .map_err(Error::storage)?;
        return Ok(());
    }
    let Some(handle) = normalized else {
        return Ok(());
    };
    let Some(entry) = directory_user_by_handle(database, &handle)? else {
        return Ok(());
    };
    if entry.telegram_user_id.is_some() {
        return Ok(());
    }
    database
        .execute(
            "UPDATE whitelist_entries SET telegram_user_id=?1,current_username=?2,display_name=?3,
                 resolved_at=?4,updated_at=?4 WHERE handle=?2 AND telegram_user_id IS NULL",
            params![
                observation.telegram_user_id,
                handle,
                observation.display_name,
                now
            ],
        )
        .map_err(Error::storage)?;
    Ok(())
}

fn whitelist_handle(database: &Connection, handle: &str, added_by: i64) -> Result<User> {
    let handle = normalize_username(handle.trim_matches(['\'', '"']));
    if handle.is_empty() {
        return Err(Error::invalid("the Telegram handle must not be empty"));
    }
    let now = Utc::now().to_rfc3339();
    database
        .execute(
            "INSERT INTO whitelist_entries(handle,current_username,added_by_telegram_user_id,whitelisted_at,updated_at)
             VALUES(?1,?1,?2,?3,?3)
             ON CONFLICT(handle) DO UPDATE SET updated_at=excluded.updated_at",
            params![handle, added_by, now],
        )
        .map_err(Error::storage)?;
    directory_user_by_handle(database, &handle)?.ok_or_else(Error::not_found)
}

fn ensure_user_root_compatible(user: &User, root: NodeId, label: &str) -> Result<()> {
    let root = root.to_string();
    if user.root_ready && user.root_node_id.as_deref() != Some(&root) {
        return Err(Error::conflict(format!(
            "This {label} already has a different root node."
        )));
    }
    Ok(())
}

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

    fn directory() -> Directory {
        let database = Connection::open_in_memory().unwrap();
        database
            .execute_batch(
                "CREATE TABLE kmap_system_roots(
                     role TEXT PRIMARY KEY CHECK(role IN ('user','kennedy')),
                     root_node_id TEXT NOT NULL UNIQUE CHECK(length(root_node_id)=8),
                     created_at TEXT NOT NULL
                 );
                 INSERT INTO kmap_system_roots VALUES(
                     'user','AAAAAAAB','2026-01-01T00:00:00Z'
                 );",
            )
            .unwrap();
        database.execute_batch(IDENTITY_MIGRATION).unwrap();
        let directory = Directory {
            database: Mutex::new(database),
        };
        directory.seed_bootstrap_user("@taek42").unwrap();
        directory
    }

    #[test]
    fn opens_against_the_identity_schema_created_by_kmap_startup() {
        let directory = std::env::temp_dir().join(format!(
            "kennedy-telegram-identity-startup-test-{}",
            uuid::Uuid::new_v4()
        ));
        std::fs::create_dir_all(&directory).unwrap();
        let user_database = directory.join("users.sqlite3");
        let kmap_schema = Connection::open(&user_database).unwrap();
        kmap_schema
            .execute_batch(
                "CREATE TABLE kmap_system_roots(
                     role TEXT PRIMARY KEY CHECK(role IN ('user','kennedy')),
                     root_node_id TEXT NOT NULL UNIQUE CHECK(length(root_node_id)=8),
                     created_at TEXT NOT NULL
                 );
                 INSERT INTO kmap_system_roots VALUES(
                     'user','AAAAAAAB','2026-01-01T00:00:00Z'
                 );",
            )
            .unwrap();
        drop(kmap_schema);

        let identity = Directory::open(&user_database, "@taek42").unwrap();
        assert!(
            identity
                .lock()
                .unwrap()
                .query_row(
                    "SELECT EXISTS(SELECT 1 FROM whitelist_entries WHERE handle='taek42')",
                    [],
                    |row| row.get::<_, i64>(0),
                )
                .unwrap()
                != 0
        );
        drop(identity);
        std::fs::remove_dir_all(directory).unwrap();
    }

    #[test]
    fn tofu_is_owned_by_kennedy_and_numeric_ids_remain_authoritative() {
        let directory = directory();
        directory
            .observe_identity(&IdentityObservation {
                telegram_user_id: 42,
                username: Some("TaEk42".into()),
                display_name: "David".into(),
            })
            .unwrap();
        assert!(
            directory
                .whitelist()
                .unwrap()
                .telegram_user_ids
                .contains(&42)
        );
        directory
            .observe_identity(&IdentityObservation {
                telegram_user_id: 43,
                username: Some("taek42".into()),
                display_name: "Other".into(),
            })
            .unwrap();
        assert!(
            !directory
                .whitelist()
                .unwrap()
                .telegram_user_ids
                .contains(&43)
        );
    }

    #[test]
    fn identity_migration_removes_legacy_anonymous_group_pseudo_user() {
        let directory = directory();
        let database = directory.lock().unwrap();
        database
            .execute(
                "INSERT INTO observed_identities(
                     telegram_user_id,current_username,display_name,first_seen_at,last_seen_at
                 ) VALUES(1087968824,'GroupAnonymousBot','Group',?1,?1)",
                [Utc::now().to_rfc3339()],
            )
            .unwrap();
        database.execute_batch(IDENTITY_MIGRATION).unwrap();
        assert_eq!(
            database
                .query_row(
                    "SELECT COUNT(*) FROM observed_identities WHERE telegram_user_id=1087968824",
                    [],
                    |row| row.get::<_, i64>(0),
                )
                .unwrap(),
            0
        );
    }

    #[test]
    fn add_user_capability_and_group_roots_stay_in_kennedy() {
        let directory = directory();
        directory
            .observe_identity(&IdentityObservation {
                telegram_user_id: 42,
                username: Some("taek42".into()),
                display_name: "David".into(),
            })
            .unwrap();
        assert!(matches!(
            directory.request_add_user(77, "@friend").unwrap(),
            AddUserOutcome::Forbidden
        ));
        assert!(matches!(
            directory.request_add_user(42, "@friend").unwrap(),
            AddUserOutcome::Whitelisted { .. }
        ));
        directory.observe_group("opaque-group").unwrap();
        let database = directory.lock().unwrap();
        let group = directory_group_by_id(&database, "opaque-group")
            .unwrap()
            .unwrap();
        assert_eq!(group.root_node_id, None);
        assert!(!group.root_ready);
    }

    #[test]
    fn root_completion_is_owned_and_conflict_checked_by_kennedy() {
        let directory = directory();
        directory
            .observe_identity(&IdentityObservation {
                telegram_user_id: 42,
                username: Some("taek42".into()),
                display_name: "David".into(),
            })
            .unwrap();
        directory.observe_group("opaque-group").unwrap();

        let user_root = "AAAAAAAC".parse().unwrap();
        let user = directory.complete_user_root(42, user_root).unwrap();
        assert!(user.root_ready);
        assert_eq!(user.root_node_id.as_deref(), Some("AAAAAAAC"));

        let group_root = "AAAAAAAD".parse().unwrap();
        let group = directory
            .complete_group_root("opaque-group", group_root)
            .unwrap();
        assert!(group.root_ready);
        assert_eq!(group.root_node_id.as_deref(), Some("AAAAAAAD"));

        let mismatch = directory
            .complete_group_root("opaque-group", "AAAAAAAE".parse().unwrap())
            .unwrap_err();
        assert_eq!(mismatch.kind(), ErrorKind::Conflict);
    }
}