//! What repositories share for others to use, kept by kind, with the
//! repositories that use each: the directory the pages show
//! (`dev.knotless.listSteps`). Shared steps are the one kind yet: a
//! release of the package `@step/<name>` whose one file is a step file
//! ([`knotless::uses`]). Each such release is kept as it is indexed, its
//! file read after ([`crate::shared::follow`]), and shown once it is
//! found to be one. Who uses each is read from the lock on each public
//! repository's default branch, as what it is written in is counted
//! ([`crate::languages`]).

use super::*;

/// The kinds of thing shared, as `shared.kind` names them.
pub const STEP: &str = "step";

/// The tables: each release of something shared, by its record's URI,
/// with its file as read, as JSON, once it is (`state` 0 until then, 1
/// for one that is what its package says, 2 for one that is not); and
/// each repository's uses of them, a row for each thing it uses, at
/// whatever version.
pub(super) const TABLES: &str = "CREATE TABLE IF NOT EXISTS shared (
      uri TEXT PRIMARY KEY, kind TEXT NOT NULL, did TEXT NOT NULL, repo_key TEXT NOT NULL,
      name TEXT NOT NULL, created_at TEXT NOT NULL, state INTEGER NOT NULL, value TEXT);
    CREATE INDEX IF NOT EXISTS shared_name ON shared (kind, state, name, did, repo_key);
    CREATE INDEX IF NOT EXISTS shared_repo ON shared (did, repo_key, kind, state, name);
    CREATE INDEX IF NOT EXISTS shared_unread ON shared (state) WHERE state = 0;
    CREATE TABLE IF NOT EXISTS shared_use (
      did TEXT NOT NULL, key TEXT NOT NULL, kind TEXT NOT NULL, of_did TEXT NOT NULL,
      of_key TEXT NOT NULL, name TEXT NOT NULL,
      PRIMARY KEY (did, key, kind, of_did, of_key, name));
    CREATE INDEX IF NOT EXISTS shared_use_of ON shared_use (kind, of_did, of_key, name);";

/// The releases of steps an index from before the table holds, once, as
/// the table is made: kept to be read, as new ones are.
pub(super) const UNREAD: &str = "INSERT OR IGNORE INTO shared
    SELECT uri, 'step', did, repo_key, substr(json_extract(value, '$.package'), 7),
      created_at, 0, NULL
    FROM release
    -- scans: release, once, as the table of what is shared is made
    WHERE substr(json_extract(value, '$.package'), 1, 6) = '@step/'";

/// The kind and name of what `package` releases, if it is shared: a step,
/// `@step/<name>`, with a name a step may have.
pub fn kind_of(package: &str) -> Option<(&'static str, &str)> {
    let name = package.strip_prefix("@step/")?;
    knotless::workflow::valid_name(name).then_some((STEP, name))
}

/// A thing shared, by its kind, the owner and key of its repository, and
/// its name.
#[derive(Debug, Clone, PartialEq)]
pub struct Shared {
    pub kind: String,
    pub did: String,
    pub key: String,
    pub name: String,
}

/// A release of a thing shared, read: its record's URI, the record, and
/// what was read of its file.
pub type Version = (String, ReleaseRecord, Value);

/// Where a page of things shared starts: after the one it names, by
/// name, owner and key, as `cursor` gives them.
fn after(cursor: Option<&str>, prefix: &str) -> (String, String, String) {
    let mut parts = cursor.unwrap_or_default().splitn(3, ' ');
    match (parts.next(), parts.next(), parts.next()) {
        (Some(name), Some(did), Some(key)) => (name.into(), did.into(), key.into()),
        _ => (prefix.into(), String::new(), String::new()),
    }
}

/// The cursor a page ending at `last` gives.
pub fn cursor(last: &Shared) -> String {
    format!("{} {} {}", last.name, last.did, last.key)
}

impl Index {
    /// Keeps the release at `uri` of the thing `package` names, if it is
    /// shared, to have its file read; and asks for it to be read. Called
    /// as the release is indexed, with its database.
    pub(super) fn share(
        &self,
        db: &Connection,
        uri: &str,
        did: &str,
        release: &ReleaseRecord,
    ) -> rusqlite::Result<()> {
        let Some((kind, name)) = kind_of(&release.package) else {
            return Ok(());
        };
        let kept = db.execute(
            "INSERT OR IGNORE INTO shared VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, NULL)",
            params![uri, kind, did, release.repo, name, release.created_at],
        );
        let _ = self.read.send(uri.to_string());
        kept.map(|_| ())
    }

    /// What [`Index::share`] asks to be read, for the one that reads it:
    /// none after the first take.
    pub fn reading(&self) -> Option<mpsc::UnboundedReceiver<String>> {
        self.reading.lock().unwrap().take()
    }

    /// The releases of things shared whose files are yet to be read.
    pub fn unread(&self) -> rusqlite::Result<Vec<String>> {
        let db = self.db.lock().unwrap();
        let mut stmt = db.prepare_cached("SELECT uri FROM shared WHERE state = 0")?;
        let rows = stmt.query_map([], |row| row.get(0))?;
        rows.collect()
    }

    /// The release at `uri` of a thing shared, with the thing, if the
    /// index has both.
    pub fn shared_release(&self, uri: &str) -> rusqlite::Result<Option<(Shared, ReleaseRecord)>> {
        let db = self.db.lock().unwrap();
        let mut stmt = db.prepare_cached(
            "SELECT s.kind, s.did, s.repo_key, s.name, r.value FROM shared s
             JOIN release r ON r.uri = s.uri WHERE s.uri = ?1",
        )?;
        let found = stmt.query_row([uri], |row| {
            let text: String = row.get(4)?;
            let thing = Shared {
                kind: row.get(0)?,
                did: row.get(1)?,
                key: row.get(2)?,
                name: row.get(3)?,
            };
            Ok((thing, text))
        });
        let found = found.optional()?;
        Ok(found.and_then(|(thing, text)| Some((thing, serde_json::from_str(&text).ok()?))))
    }

    /// Keeps what was read of the release at `uri`'s file: what it is, or
    /// none for one that is not what its package says.
    pub fn set_shared(&self, uri: &str, value: Option<&Value>) -> rusqlite::Result<()> {
        let db = self.db.lock().unwrap();
        let (state, text) = match value {
            Some(v) => (1, Some(v.to_string())),
            None => (2, None),
        };
        let mut stmt =
            db.prepare_cached("UPDATE shared SET state = ?2, value = ?3 WHERE uri = ?1")?;
        stmt.execute(params![uri, state, text]).map(|_| ())
    }

    /// Keeps `used` as what `did`'s repository with `key` uses, in place of
    /// what it did.
    pub fn set_uses(&self, did: &str, key: &str, used: &[Shared]) -> rusqlite::Result<()> {
        let mut db = self.db.lock().unwrap();
        let tx = db.transaction()?;
        tx.execute(
            "DELETE FROM shared_use WHERE did = ?1 AND key = ?2",
            [did, key],
        )?;
        for s in used {
            tx.execute(
                "INSERT OR IGNORE INTO shared_use VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![did, key, s.kind, s.did, s.key, s.name],
            )?;
        }
        tx.commit()
    }

    /// The key of `did`'s repository named `name`, if the index has it.
    pub fn key_named(&self, did: &str, name: &str) -> rusqlite::Result<Option<String>> {
        let db = self.db.lock().unwrap();
        let mut stmt =
            db.prepare_cached("SELECT key FROM repo WHERE did = ?1 AND name = ?2 LIMIT 1")?;
        stmt.query_row([did, name], |row| row.get(0)).optional()
    }

    /// A page of at most `limit` things of `kind` shared, in the order of
    /// their names: those of `did`'s repository `key`, or everyone's, whose
    /// names start with `prefix`, after the cursor an earlier page gave;
    /// and the cursor for the next, if any. One of an account, repository
    /// or release taken down, or whose repository the index lacks, is left
    /// out.
    pub fn shared(
        &self,
        kind: &str,
        repo: Option<(&str, &str)>,
        prefix: &str,
        limit: u32,
        from: Option<&str>,
    ) -> rusqlite::Result<(Vec<Shared>, Option<String>)> {
        let (name, did, key) = after(from, prefix);
        let (of_did, of_key) = repo.unzip();
        let db = self.db.lock().unwrap();
        // Each way in by an index of its own: one repository's, or
        // everyone's, by name.
        let sql = match repo {
            Some(_) => {
                "SELECT DISTINCT s.name, s.did, s.repo_key FROM shared s
                 WHERE s.did = ?7 AND s.repo_key = ?8 AND s.kind = ?1 AND s.state = 1
                   AND (s.name, s.did, s.repo_key) > (?2, ?3, ?4) AND s.name < ?5
                   AND s.uri NOT IN down AND s.did NOT IN down
                   AND EXISTS (SELECT 1 FROM repo r WHERE r.did = s.did
                     AND r.key = s.repo_key AND r.uri NOT IN down)
                 ORDER BY s.name, s.did, s.repo_key LIMIT ?6"
            }
            None => {
                "SELECT DISTINCT s.name, s.did, s.repo_key FROM shared s
                 WHERE ?7 IS NULL AND ?8 IS NULL AND s.kind = ?1 AND s.state = 1
                   AND (s.name, s.did, s.repo_key) > (?2, ?3, ?4) AND s.name < ?5
                   AND s.uri NOT IN down AND s.did NOT IN down
                   AND EXISTS (SELECT 1 FROM repo r WHERE r.did = s.did
                     AND r.key = s.repo_key AND r.uri NOT IN down)
                 ORDER BY s.name, s.did, s.repo_key LIMIT ?6"
            }
        };
        let mut stmt = db.prepare_cached(sql)?;
        // The names from the prefix up to the last that could start so.
        let end = format!("{prefix}{}", char::MAX);
        let args = params![kind, name, did, key, end, limit + 1, of_did, of_key];
        let thing = |row: &rusqlite::Row| -> rusqlite::Result<Shared> {
            Ok(Shared {
                kind: kind.to_string(),
                name: row.get(0)?,
                did: row.get(1)?,
                key: row.get(2)?,
            })
        };
        let rows = stmt.query_map(args, thing)?;
        let mut found = rows.collect::<rusqlite::Result<Vec<_>>>()?;
        let more = found.len() > limit as usize;
        found.truncate(limit as usize);
        let next = found.last().filter(|_| more).map(cursor);
        Ok((found, next))
    }

    /// Each release of `thing` whose file is one, as read and marked if
    /// yanked, leaving out those taken down.
    pub fn shared_versions(&self, thing: &Shared) -> rusqlite::Result<Vec<Version>> {
        let db = self.db.lock().unwrap();
        let mut stmt = db.prepare_cached(
            "SELECT s.uri, r.value, s.value, y.uri IS NOT NULL
             FROM shared s JOIN release r ON r.uri = s.uri
             LEFT JOIN release_yank y ON y.release = s.uri
             WHERE s.did = ?1 AND s.repo_key = ?2 AND s.kind = ?3 AND s.state = 1
               AND s.name = ?4 AND s.uri NOT IN down",
        )?;
        let args = params![thing.did, thing.key, thing.kind, thing.name];
        let texts = |row: &rusqlite::Row| -> rusqlite::Result<(String, String, String, bool)> {
            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
        };
        let rows = stmt.query_map(args, texts)?;
        let rows = rows.collect::<rusqlite::Result<Vec<_>>>()?;
        let read = rows.into_iter().filter_map(|(uri, record, value, yanked)| {
            let record = serde_json::from_str::<ReleaseRecord>(&record).ok()?;
            let record = ReleaseRecord { yanked, ..record };
            Some((uri, record, serde_json::from_str(&value).ok()?))
        });
        Ok(read.collect())
    }

    /// How many public repositories use `thing`, and at most `limit` of
    /// them, by owner and name, leaving out those taken down.
    pub fn users(
        &self,
        thing: &Shared,
        limit: u32,
    ) -> rusqlite::Result<(u64, Vec<(String, String)>)> {
        let db = self.db.lock().unwrap();
        let args = params![thing.kind, thing.did, thing.key, thing.name, limit];
        let mut count = db.prepare_cached(
            "SELECT COUNT(*) FROM shared_use u
             WHERE u.kind = ?1 AND u.of_did = ?2 AND u.of_key = ?3 AND u.name = ?4
               AND u.did NOT IN down AND EXISTS (SELECT 1 FROM repo r
                 WHERE r.did = u.did AND r.key = u.key AND r.uri NOT IN down)",
        )?;
        let n = count.query_row(&args[..4], |row| row.get(0))?;
        let mut some = db.prepare_cached(
            "SELECT u.did, (SELECT r.name FROM repo r WHERE r.did = u.did AND r.key = u.key
                 AND r.uri NOT IN down LIMIT 1)
             FROM shared_use u
             WHERE u.kind = ?1 AND u.of_did = ?2 AND u.of_key = ?3 AND u.name = ?4
               AND u.did NOT IN down AND EXISTS (SELECT 1 FROM repo r
                 WHERE r.did = u.did AND r.key = u.key AND r.uri NOT IN down)
             LIMIT ?5",
        )?;
        let rows = some.query_map(args, |row| Ok((row.get(0)?, row.get(1)?)))?;
        Ok((n, rows.collect::<rusqlite::Result<Vec<_>>>()?))
    }
}