//! The SDK's store: one SQLite file per client.

use crate::{Share, ShareState};
use anyhow::Result;
use rusqlite_migration::Migrations;
use std::fs;
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use wire::{AppKind, AppProxyMode, ConfigHash, ShareInfo, SharedApp};
use wispers_access_wire as wire;

pub struct DB {
    conn: Mutex<rusqlite::Connection>,
}

/// Device-local ID of a share.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct ShareId(String);

impl ShareId {
    pub(crate) fn mint() -> Self {
        ShareId(uuid::Uuid::new_v4().to_string())
    }

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

uniffi::custom_type!(ShareId, String, {
    lower: |id| id.0,
    try_lift: |s| Ok(ShareId(s)),
});

impl std::fmt::Display for ShareId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl DB {
    /// A store that lives in memory, for tests.
    #[cfg(test)]
    pub(crate) fn in_memory() -> Arc<Self> {
        let conn = prepare(rusqlite::Connection::open_in_memory().unwrap()).unwrap();
        Arc::new(DB {
            conn: Mutex::new(conn),
        })
    }

    /// Opens `state.db` under `dir`, creating both as needed.
    pub fn open(dir: &Path) -> Result<Arc<Self>> {
        let conn = open_db(dir)?;
        let db = DB {
            conn: Mutex::new(conn),
        };
        Ok(Arc::new(db))
    }

    pub fn new_row(self: &Arc<Self>) -> Result<Row> {
        let t = now_millis();
        let id: i64;
        {
            let conn = self.conn.lock().expect("unpoisoned db lock");
            conn.execute(
                "INSERT INTO shares (share_id, created_at) VALUES (?1, ?2)",
                rusqlite::params![ShareId::mint().as_str(), t],
            )?;
            id = conn.last_insert_rowid();
        }
        Ok(Row {
            db: self.clone(),
            id,
        })
    }

    /// Looks a share up by its hostname label (or share id).
    pub fn find_row(self: &Arc<Self>, key: &str) -> Result<Option<Row>> {
        use rusqlite::OptionalExtension;
        let conn = self.conn.lock().expect("unpoisoned db lock");
        let id = conn
            .query_row(
                "SELECT id FROM shares
                 WHERE complete = TRUE AND (hostname = ?1 OR share_id = ?1)",
                [key],
                |r| r.get::<_, i64>(0),
            )
            .optional()?;
        Ok(id.map(|id| Row {
            db: self.clone(),
            id,
        }))
    }

    pub fn get_all_rows(self: &Arc<Self>) -> Result<Vec<Row>> {
        let conn = self.conn.lock().expect("unpoisoned db lock");
        let mut stmt = conn.prepare("SELECT id FROM shares WHERE complete = TRUE")?;
        let rows = stmt
            .query_map([], |r| {
                Ok(Row {
                    db: self.clone(),
                    id: r.get(0)?,
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }
}

/// One joined share.
#[derive(Clone)]
pub struct Row {
    db: Arc<DB>,
    id: i64,
}

impl Row {
    pub fn share_id(&self) -> Result<ShareId> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let id = conn.query_row(
            "SELECT share_id FROM shares WHERE id = ?1",
            [self.id],
            |r| r.get::<_, String>(0),
        )?;
        Ok(ShareId(id))
    }

    pub fn write_display_name(&self, name: &str) -> Result<()> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        conn.execute(
            "UPDATE shares SET display_name = ?1 WHERE id = ?2",
            rusqlite::params![name, self.id],
        )?;
        Ok(())
    }

    /// Claims `hostname` as this share's label, or `hostname-2`, `-3`, … if
    /// taken; falls back to the share id.
    pub fn write_deduped_hostname(&self, hostname: &str) -> Result<String> {
        let share_id = self.share_id()?;
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        for n in 1.. {
            let candidate: String = if n == 1 {
                hostname.to_owned()
            } else {
                format!("{}-{}", hostname, n)
            };
            if candidate.len() > 63 {
                // Max DNS label exceeded.
                break;
            }
            match conn.execute(
                "UPDATE shares SET hostname = ?1 WHERE id = ?2",
                rusqlite::params![candidate, self.id],
            ) {
                Ok(_) => return Ok(candidate),
                Err(e) if is_unique_violation(&e) => continue,
                Err(e) => return Err(e.into()),
            }
        }
        // Normal deduping has failed. Just use the share id.
        conn.execute(
            "UPDATE shares SET hostname = ?1 WHERE id = ?2",
            rusqlite::params![share_id.as_str(), self.id],
        )?;
        Ok(share_id.to_string())
    }

    /// (share id, display name, hostname label)
    pub fn read_names(&self) -> Result<(ShareId, String, String)> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let row = conn.query_row(
            "SELECT share_id, display_name, hostname
                 FROM shares
                 WHERE id = ?1",
            [self.id],
            |row| {
                Ok((
                    ShareId(row.get::<_, String>(0)?),
                    row.get::<_, Option<String>>(1)?.unwrap_or("".to_owned()),
                    row.get::<_, Option<String>>(2)?.unwrap_or("".to_owned()),
                ))
            },
        )?;
        Ok(row)
    }

    /// Persists the self-hosted hub a Wispers Connect share uses, if any.
    pub fn write_wispers_connect_backend(&self, backend: Option<&str>) -> Result<()> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        conn.execute(
            "UPDATE shares SET wispers_connect_backend = ?1 WHERE id = ?2",
            rusqlite::params![backend, self.id],
        )?;
        Ok(())
    }

    /// Records that this share rides iroh, and which endpoint its host node
    /// is. The device's key for it lives in the secret store.
    pub fn write_iroh_endpoint_id(&self, host_endpoint_id: &str) -> Result<()> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        conn.execute(
            "UPDATE shares SET transport = 'iroh', iroh_endpoint_id = ?1 WHERE id = ?2",
            rusqlite::params![host_endpoint_id, self.id],
        )?;
        Ok(())
    }

    /// The host node's endpoint ID, in hex, for an iroh share.
    pub fn read_iroh_endpoint_id(&self) -> Result<Option<String>> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let host = conn.query_row(
            "SELECT iroh_endpoint_id FROM shares WHERE id = ?1",
            [self.id],
            |r| r.get::<_, Option<String>>(0),
        )?;
        Ok(host)
    }

    /// The transport this share rides.
    pub fn read_transport_kind(&self) -> Result<wire::Transport> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let name: String = conn.query_row(
            "SELECT transport FROM shares WHERE id = ?1",
            [self.id],
            |r| r.get(0),
        )?;
        Ok(name.parse()?)
    }

    pub fn read_wispers_connect_backend(&self) -> Result<Option<String>> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let backend = conn.query_row(
            "SELECT wispers_connect_backend FROM shares WHERE id = ?1",
            [self.id],
            |r| r.get::<_, Option<String>>(0),
        )?;
        Ok(backend)
    }

    pub fn write_share_info(&self, info: &ShareInfo) -> Result<()> {
        let mut conn = self.db.conn.lock().expect("unpoisoned db lock");
        let tx = conn.transaction()?;
        tx.execute(
            "UPDATE shares SET display_name = ?1, config_hash = ?2, ca_certificates = ?3
             WHERE id = ?4",
            rusqlite::params![
                info.name,
                info.config_hash.to_string(),
                serde_json::to_string(&info.ca_certificates)?,
                self.id
            ],
        )?;
        tx.execute("DELETE FROM apps WHERE share_id = ?1", [self.id])?;
        for (position, app) in info.apps.iter().enumerate() {
            tx.execute(
                "INSERT INTO apps
                     (share_id, position, app_id, name, kind, mode, start_url, proxied_origins)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
                rusqlite::params![
                    self.id,
                    position as i64,
                    app.id,
                    app.name,
                    app.kind.as_str(),
                    app.mode.as_str(),
                    app.start_url,
                    serde_json::to_string(&app.proxied_origins)?,
                ],
            )?;
        }
        tx.commit()?;
        Ok(())
    }

    /// The config hash the stored app list came with, for a conditional
    /// refetch. `None` before the first fetch.
    pub fn read_share_config_hash(&self) -> Result<Option<ConfigHash>> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let hash = conn.query_row(
            "SELECT config_hash FROM shares WHERE id = ?1",
            [self.id],
            |r| r.get::<_, Option<String>>(0),
        )?;
        Ok(hash.and_then(|h| h.parse().ok()))
    }

    /// The apps as last fetched, in the host node's order.
    pub fn read_apps(&self) -> Result<Vec<SharedApp>> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let mut stmt = conn.prepare(
            "SELECT app_id, name, kind, mode, start_url, proxied_origins
             FROM apps WHERE share_id = ?1 ORDER BY position",
        )?;
        let apps = stmt
            .query_map([self.id], |r| {
                Ok(SharedApp {
                    id: r.get(0)?,
                    name: r.get(1)?,
                    kind: AppKind::parse_or_web(&r.get::<_, String>(2)?),
                    mode: AppProxyMode::parse(&r.get::<_, String>(3)?),
                    start_url: r.get(4)?,
                    // Written by write_share_info, so always a JSON list.
                    proxied_origins: serde_json::from_str(&r.get::<_, String>(5)?)
                        .unwrap_or_default(),
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(apps)
    }

    pub fn mark_complete(&self) -> Result<()> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        conn.execute("UPDATE shares SET complete = TRUE WHERE id = ?1", [self.id])?;
        Ok(())
    }

    /// Records that the hub definitively rejected this share's node. One-way:
    /// a terminal share is never dialed again, only removed.
    pub fn write_terminal_state(&self, state: &str) -> Result<()> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        conn.execute(
            "UPDATE shares SET terminal_state = ?1 WHERE id = ?2",
            rusqlite::params![state, self.id],
        )?;
        Ok(())
    }

    /// Records whether the host node told our guest node it's too old. Returns
    /// `true` if this write changed the value.
    pub fn write_outdated(&self, outdated: bool) -> Result<bool> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let changed = conn.execute(
            "UPDATE shares SET outdated = ?1 WHERE id = ?2 AND outdated != ?1",
            rusqlite::params![outdated, self.id],
        )?;
        Ok(changed > 0)
    }

    /// When this device joined the share.
    pub fn read_joined_at(&self) -> Result<SystemTime> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let millis: i64 = conn.query_row(
            "SELECT created_at FROM shares WHERE id = ?1",
            [self.id],
            |r| r.get(0),
        )?;
        Ok(UNIX_EPOCH + Duration::from_millis(millis.max(0) as u64))
    }

    /// The share as the SDK's API presents it.
    pub fn read_share(&self) -> Result<Share> {
        let (id, name, label) = self.read_names()?;
        let state = match self
            .read_terminal_state()?
            .as_deref()
            .and_then(crate::transports::TerminalState::parse)
        {
            None if self.read_outdated()? => ShareState::Outdated,
            None => ShareState::Live,
            Some(crate::transports::TerminalState::Removed) => ShareState::Removed,
            Some(crate::transports::TerminalState::Revoked) => ShareState::Revoked,
        };
        Ok(Share {
            id,
            name,
            label,
            transport: self.read_transport_kind()?,
            apps: self.read_apps()?,
            ca_certificates: self.read_ca_certificates()?,
            state,
            joined_at: self.read_joined_at()?,
        })
    }

    /// The share's CA certificates, as PEM, as last fetched.
    pub fn read_ca_certificates(&self) -> Result<Vec<String>> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let json: String = conn.query_row(
            "SELECT ca_certificates FROM shares WHERE id = ?1",
            [self.id],
            |r| r.get(0),
        )?;
        Ok(serde_json::from_str(&json)?)
    }

    pub fn read_terminal_state(&self) -> Result<Option<String>> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let state = conn.query_row(
            "SELECT terminal_state FROM shares WHERE id = ?1",
            [self.id],
            |r| r.get::<_, Option<String>>(0),
        )?;
        Ok(state)
    }

    /// Whether the host node last turned this device away as too old.
    pub fn read_outdated(&self) -> Result<bool> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let outdated: bool = conn.query_row(
            "SELECT outdated FROM shares WHERE id = ?1",
            [self.id],
            |r| r.get(0),
        )?;
        Ok(outdated)
    }

    /// The port a `PerApp` proxy last served this app on.
    pub fn read_app_port(&self, app_id: &str) -> Result<Option<u16>> {
        use rusqlite::OptionalExtension;
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let port = conn
            .query_row(
                "SELECT port FROM app_ports WHERE share_id = ?1 AND app_id = ?2",
                rusqlite::params![self.id, app_id],
                |r| r.get::<_, u16>(0),
            )
            .optional()?;
        Ok(port)
    }

    pub fn write_app_port(&self, app_id: &str, port: u16) -> Result<()> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        conn.execute(
            "INSERT INTO app_ports (share_id, app_id, port) VALUES (?1, ?2, ?3)
             ON CONFLICT (share_id, app_id) DO UPDATE SET port = excluded.port",
            rusqlite::params![self.id, app_id, port],
        )?;
        Ok(())
    }

    pub fn read_forward_proxy_port(&self) -> Result<Option<u16>> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        let port = conn.query_row(
            "SELECT forward_proxy_port FROM shares WHERE id = ?1",
            [self.id],
            |r| r.get::<_, Option<u16>>(0),
        )?;
        Ok(port)
    }

    pub fn write_forward_proxy_port(&self, port: u16) -> Result<()> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        conn.execute(
            "UPDATE shares SET forward_proxy_port = ?2 WHERE id = ?1",
            rusqlite::params![self.id, port],
        )?;
        Ok(())
    }

    /// Deletes the share and its apps outright. Its secrets are the secret
    /// store's to delete.
    pub fn delete_row(&self) -> Result<()> {
        let conn = self.db.conn.lock().expect("unpoisoned db lock");
        conn.execute("DELETE FROM shares WHERE id = ?1", [self.id])?;
        Ok(())
    }
}

fn is_unique_violation(e: &rusqlite::Error) -> bool {
    use rusqlite::{Error, ErrorCode};
    matches!(
        e,
        Error::SqliteFailure(
            rusqlite::ffi::Error {
                code: ErrorCode::ConstraintViolation,
                ..
            },
            _
        )
    )
}

fn open_db(dir: &Path) -> Result<rusqlite::Connection> {
    let db_path = dir.join("state.db");
    fs::create_dir_all(dir)?;
    let conn = rusqlite::Connection::open(db_path)?;
    prepare(conn)
}

/// Everything a freshly opened connection needs before use: the pragmas,
/// the schema at its latest version, and the sweep of rows a failed `join`
/// left behind. Shared with the in-memory database the tests use.
fn prepare(mut conn: rusqlite::Connection) -> Result<rusqlite::Connection> {
    conn.pragma_update(None, "foreign_keys", "ON")?;
    migrations().to_latest(&mut conn)?;
    clean_up_incomplete_rows(&mut conn)?;
    Ok(conn)
}

fn migrations() -> Migrations<'static> {
    use rusqlite_migration::M;

    Migrations::new(vec![
        // v1 — one row per joined share (which transport, how to reach the
        // host node, and what it said the share is), one row per app in it.
        // Key material is not here but in the secret store. The schema of
        // the unreleased waclient builds that kept keys in this file was
        // dropped without a migration.
        M::up(
            "CREATE TABLE shares (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
                 share_id TEXT NOT NULL UNIQUE,
                 display_name TEXT,
                 hostname TEXT UNIQUE,
                 transport TEXT NOT NULL DEFAULT 'wispers-connect',
                 wispers_connect_backend TEXT,
                 iroh_endpoint_id TEXT,
                 config_hash TEXT,
                 created_at INTEGER NOT NULL,
                 complete INTEGER,
                 terminal_state TEXT
             ) STRICT;
             CREATE TABLE apps (
                 share_id INTEGER NOT NULL REFERENCES shares(id) ON DELETE CASCADE,
                 position INTEGER NOT NULL,
                 app_id TEXT NOT NULL,
                 name TEXT NOT NULL,
                 kind TEXT NOT NULL,
                 PRIMARY KEY (share_id, app_id)
             ) STRICT;
             CREATE TABLE app_ports (
                 share_id INTEGER NOT NULL REFERENCES shares(id) ON DELETE CASCADE,
                 app_id TEXT NOT NULL,
                 port INTEGER NOT NULL,
                 PRIMARY KEY (share_id, app_id)
             ) STRICT;",
        ),
        // v2 — record when the guest node was told it's outdated
        // (guest_node.rs, `ClientTooOldError`).
        M::up("ALTER TABLE shares ADD COLUMN outdated INTEGER NOT NULL DEFAULT 0;"),
        // v3 — forward-proxy mode. The lists are JSON. Earlier versions of
        // this crate dropped the share info fields they didn't know, so an
        // app they stored may be a forward-proxy app without its mode and
        // origins. Forgetting the config hashes makes the next share info
        // fetch unconditional, which fills them in.
        M::up(
            "ALTER TABLE shares ADD COLUMN ca_certificates TEXT NOT NULL DEFAULT '[]';
             ALTER TABLE apps ADD COLUMN mode TEXT NOT NULL DEFAULT 'reverse-proxy';
             ALTER TABLE apps ADD COLUMN start_url TEXT;
             ALTER TABLE apps ADD COLUMN proxied_origins TEXT NOT NULL DEFAULT '[]';
             UPDATE shares SET config_hash = NULL;",
        ),
        // v4 — the share's port on the desktop's forward proxy, kept so that
        // the share's running browser profile still finds it after a restart.
        M::up("ALTER TABLE shares ADD COLUMN forward_proxy_port INTEGER;"),
    ])
}

fn clean_up_incomplete_rows(conn: &mut rusqlite::Connection) -> Result<()> {
    conn.execute("DELETE FROM shares WHERE complete = FALSE", [])?;
    Ok(())
}

fn now_millis() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("system clock is before the Unix epoch")
        .as_millis() as i64
}

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

    #[test]
    fn migration_set_is_valid() {
        migrations().validate().unwrap();
    }

    #[test]
    fn upgrading_to_modes_forgets_config_hashes() {
        let mut conn = rusqlite::Connection::open_in_memory().unwrap();
        migrations().to_version(&mut conn, 2).unwrap();
        conn.execute(
            "INSERT INTO shares (share_id, config_hash, created_at) VALUES ('s', '7', 0)",
            [],
        )
        .unwrap();
        migrations().to_latest(&mut conn).unwrap();
        let hash: Option<String> = conn
            .query_row("SELECT config_hash FROM shares", [], |r| r.get(0))
            .unwrap();
        assert_eq!(hash, None);
    }

    #[test]
    fn share_info_round_trips_and_replaces() {
        let db = DB::in_memory();
        let row = db.new_row().unwrap();
        assert!(row.read_share_config_hash().unwrap().is_none());
        assert!(row.read_apps().unwrap().is_empty());

        let info = ShareInfo {
            config_hash: ConfigHash(7),
            name: "Family".into(),
            transport: "wispers-connect".into(),
            apps: vec![
                SharedApp {
                    id: "jf".into(),
                    name: "Jellyfin".into(),
                    kind: AppKind::Jellyfin,
                    mode: AppProxyMode::ReverseProxy,
                    start_url: None,
                    proxied_origins: Vec::new(),
                },
                SharedApp {
                    id: "dev".into(),
                    name: "Dev".into(),
                    kind: AppKind::Web,
                    mode: AppProxyMode::ForwardProxy,
                    start_url: Some("http://localhost:4200/".into()),
                    proxied_origins: vec!["http://localhost:4200".into()],
                },
            ],
            ca_certificates: vec!["-----BEGIN CERTIFICATE-----\n…".into()],
        };
        row.write_share_info(&info).unwrap();
        assert_eq!(row.read_share_config_hash().unwrap(), Some(ConfigHash(7)));
        assert_eq!(row.read_apps().unwrap(), info.apps);
        assert_eq!(row.read_names().unwrap().1, "Family");

        // A later fetch replaces the list wholesale, order included.
        let later = ShareInfo {
            config_hash: ConfigHash(8),
            apps: vec![info.apps[1].clone()],
            ..info
        };
        row.write_share_info(&later).unwrap();
        assert_eq!(row.read_share_config_hash().unwrap(), Some(ConfigHash(8)));
        assert_eq!(row.read_apps().unwrap(), later.apps);

        // Deleting the share takes its apps with it.
        row.write_deduped_hostname("family").unwrap();
        row.mark_complete().unwrap();
        assert!(db.find_row("family").unwrap().is_some());
        row.delete_row().unwrap();
        assert!(db.find_row("family").unwrap().is_none());
        let orphans: i64 = db
            .conn
            .lock()
            .unwrap()
            .query_row("SELECT count(*) FROM apps", [], |r| r.get(0))
            .unwrap();
        assert_eq!(orphans, 0);
    }

    #[test]
    fn hostnames_dedupe() {
        let db = DB::in_memory();
        let a = db.new_row().unwrap();
        let b = db.new_row().unwrap();
        assert_ne!(a.share_id().unwrap(), b.share_id().unwrap());
        assert_eq!(a.write_deduped_hostname("family").unwrap(), "family");
        assert_eq!(b.write_deduped_hostname("family").unwrap(), "family-2");
        // Lookup by share id, once the row is complete.
        let id = a.share_id().unwrap();
        assert!(db.find_row(id.as_str()).unwrap().is_none());
        a.mark_complete().unwrap();
        assert!(db.find_row(id.as_str()).unwrap().is_some());
    }
}