//! The appview's index: every public repository and every pull, issue,
//! comment and state change anyone has written about one, from the firehose
//! and a backfill. Private repositories never reach it; their records are
//! not on the firehose.
//!
//! It keeps what moderation needs too: whom owners block from all their
//! repositories, labels from the labelers the server follows, and, kept
//! when the rest is rebuilt, its own takedowns, the labels it gives out for
//! them, and the reports it was sent.

use knotless::forge::Entry;
use knotless::records::{RepoRef, nsid};
use rusqlite::{Connection, OptionalExtension, params};
use serde::Serialize;
use serde_json::Value;
use std::path::Path;
use std::sync::Mutex;

pub struct Index {
    db: Mutex<Connection>,
}

/// A repository as the index knows it.
#[derive(Serialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct Listed {
    pub did: String,
    pub name: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub fork_of: Option<String>,
    pub created_at: String,
    /// Its record, which pages are cut after.
    #[serde(skip)]
    pub uri: String,
}

/// A ref moving, as a claim recorded it.
#[derive(Debug, Clone, PartialEq)]
pub struct Push {
    /// The claim, which pages are cut after.
    pub uri: String,
    /// The repository's record.
    pub repo: String,
    pub did: String,
    /// The repository's name now.
    pub name: String,
    /// The repository it was forked from, if it is a fork.
    pub fork_of: Option<String>,
    /// The full ref name, such as `refs/heads/main`.
    pub refname: String,
    /// Where it was, when it was there before.
    pub from: Option<String>,
    /// Where it is now, unless it was deleted.
    pub to: Option<String>,
    pub created_at: String,
}

/// The shape of the tables. An index of another shape is dropped and
/// built again: it is a cache, and `--backfill` refills it.
const VERSION: i64 = 6;

/// Drops the tables that are a cache. Takedowns, the labels given out for
/// them and reports are not, and stay.
const CACHE: &str =
    "DROP TABLE IF EXISTS repo; DROP TABLE IF EXISTS forge; DROP TABLE IF EXISTS push;
    DROP TABLE IF EXISTS moved; DROP TABLE IF EXISTS block; DROP TABLE IF EXISTS label;
    DROP TABLE IF EXISTS labeler;";

impl Index {
    pub fn open(path: &Path) -> rusqlite::Result<Index> {
        let db = Connection::open(path)?;
        let version: i64 = db.query_row("PRAGMA user_version", [], |row| row.get(0))?;
        if version != VERSION {
            db.execute_batch(CACHE)?;
        }
        // A repository is known by its owner and key; its name, and the
        // URIs others wrote naming it, can change.
        db.execute_batch(&format!(
            "PRAGMA journal_mode = WAL;
             PRAGMA synchronous = NORMAL;
             PRAGMA user_version = {VERSION};
             CREATE TABLE IF NOT EXISTS repo (
               uri TEXT PRIMARY KEY, did TEXT NOT NULL, name TEXT NOT NULL, key TEXT NOT NULL,
               description TEXT, fork_of TEXT, fork_did TEXT, fork_key TEXT,
               from_did TEXT, from_key TEXT, created_at TEXT NOT NULL, value TEXT NOT NULL);
             CREATE INDEX IF NOT EXISTS repo_key ON repo (did, key);
             CREATE INDEX IF NOT EXISTS repo_fork ON repo (fork_did, fork_key);
             CREATE INDEX IF NOT EXISTS repo_created ON repo (created_at);
             CREATE TABLE IF NOT EXISTS forge (
               uri TEXT PRIMARY KEY, author TEXT NOT NULL, repo_did TEXT NOT NULL,
               repo_key TEXT NOT NULL, value TEXT NOT NULL);
             CREATE INDEX IF NOT EXISTS forge_repo ON forge (repo_did, repo_key);
             CREATE TABLE IF NOT EXISTS push (
               uri TEXT PRIMARY KEY, did TEXT NOT NULL, key TEXT NOT NULL, ref TEXT NOT NULL,
               before TEXT, after TEXT, created_at TEXT NOT NULL);
             CREATE INDEX IF NOT EXISTS push_created ON push (created_at);
             CREATE TABLE IF NOT EXISTS moved (
               uri TEXT PRIMARY KEY, from_did TEXT NOT NULL, from_key TEXT NOT NULL,
               to_did TEXT NOT NULL, to_key TEXT NOT NULL);
             CREATE TABLE IF NOT EXISTS block (
               uri TEXT PRIMARY KEY, did TEXT NOT NULL, subject TEXT NOT NULL);
             CREATE INDEX IF NOT EXISTS block_did ON block (did);
             CREATE TABLE IF NOT EXISTS label (
               src TEXT NOT NULL, uri TEXT NOT NULL, val TEXT NOT NULL, exp TEXT,
               PRIMARY KEY (src, uri, val));
             CREATE TABLE IF NOT EXISTS labeler (src TEXT PRIMARY KEY, seq INTEGER NOT NULL);
             CREATE TABLE IF NOT EXISTS takedown (
               subject TEXT PRIMARY KEY, value TEXT NOT NULL, at TEXT NOT NULL);
             CREATE TABLE IF NOT EXISTS emitted (seq INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT NOT NULL);
             CREATE TABLE IF NOT EXISTS report (
               id INTEGER PRIMARY KEY AUTOINCREMENT, value TEXT NOT NULL);
             CREATE TEMP VIEW IF NOT EXISTS down AS
               SELECT subject FROM takedown
               UNION SELECT uri FROM label
                 WHERE exp IS NULL OR exp > strftime('%Y-%m-%dT%H:%M:%fZ', 'now');"
        ))?;
        Ok(Index { db: Mutex::new(db) })
    }

    /// Records the record at `uri`, by `did`, replacing what was there.
    pub fn put(
        &self,
        did: &str,
        collection: &str,
        rkey: &str,
        value: &Value,
    ) -> rusqlite::Result<()> {
        let uri = format!("at://{did}/{collection}/{rkey}");
        let db = self.db.lock().unwrap();
        let text = value.to_string();
        let named = |field: &str| serde_json::from_value::<RepoRef>(value[field].clone()).ok();
        let key = value["key"].as_str().unwrap_or_default();
        // An owner's word that the repository moved to another account:
        // its offer, or the redirect left when the old copy went.
        let away = match collection {
            nsid::REPO => named("transfer").map(|to| (key.to_string(), to)),
            nsid::REDIRECT => named("to").map(|to| (to.key.clone(), to)),
            _ => None,
        };
        db.execute("DELETE FROM moved WHERE uri = ?1", [&uri])?;
        if let Some((from_key, to)) = away.filter(|(_, to)| to.did() != did) {
            let moved = db.execute(
                "INSERT INTO moved VALUES (?1, ?2, ?3, ?4, ?5)",
                params![uri, did, from_key, to.did(), to.key],
            );
            moved?;
        }
        // Bound, then checked: a `?` alone on a line counts as a line of its
        // own for coverage.
        let wrote = if collection == nsid::REPO {
            let fork = named("forkOf");
            let from = named("transferredFrom");
            db.execute(
                "INSERT OR REPLACE INTO repo VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
                params![
                    uri,
                    did,
                    rkey,
                    key,
                    value["description"].as_str(),
                    fork.as_ref().map(|f| &f.uri),
                    fork.as_ref().map(RepoRef::did),
                    fork.as_ref().map(|f| &f.key),
                    from.as_ref().map(RepoRef::did),
                    from.as_ref().map(|f| &f.key),
                    value["createdAt"].as_str().unwrap_or_default(),
                    text
                ],
            )
        } else if collection == nsid::CLAIM {
            // A ref moving, kept after compaction drops its claim: the
            // claims are the reflog, and this is the history of it.
            db.execute(
                "INSERT OR REPLACE INTO push VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
                params![
                    uri,
                    did,
                    value["repo"].as_str().unwrap_or_default(),
                    value["name"].as_str().unwrap_or_default(),
                    value["from"].as_str(),
                    value["to"].as_str(),
                    value["createdAt"].as_str().unwrap_or_default()
                ],
            )
        } else if let Some(repo) = named("repo") {
            db.execute(
                "INSERT OR REPLACE INTO forge VALUES (?1, ?2, ?3, ?4, ?5)",
                params![uri, did, repo.did(), repo.key, text],
            )
        } else if collection == nsid::BLOCK {
            // A block naming no repository keeps its subject out of all the
            // blocker's.
            let subject = value["subject"].as_str().unwrap_or_default();
            db.execute(
                "INSERT OR REPLACE INTO block VALUES (?1, ?2, ?3)",
                params![uri, did, subject],
            )
        } else {
            Ok(0)
        };
        wrote?;
        Ok(())
    }

    pub fn delete(&self, did: &str, collection: &str, rkey: &str) -> rusqlite::Result<()> {
        let uri = format!("at://{did}/{collection}/{rkey}");
        let db = self.db.lock().unwrap();
        db.execute("DELETE FROM repo WHERE uri = ?1", [&uri])?;
        db.execute("DELETE FROM forge WHERE uri = ?1", [&uri])?;
        db.execute("DELETE FROM moved WHERE uri = ?1", [&uri])?;
        db.execute("DELETE FROM block WHERE uri = ?1", [&uri])?;
        Ok(())
    }

    /// Every forge record about `did`'s repository `key` that stands: not
    /// taken down, not by an account taken down, and not kept out by the
    /// repository's owner.
    pub fn forge(&self, did: &str, key: &str) -> rusqlite::Result<Vec<Entry>> {
        let db = self.db.lock().unwrap();
        // With those about the repository it moved from, when both owners
        // say so: it names that one, and that one's owner said it moved here.
        let mut stmt = db.prepare_cached(
            "SELECT * FROM (
               SELECT uri, author, value FROM forge WHERE repo_did = ?1 AND repo_key = ?2
               UNION ALL
               SELECT f.uri, f.author, f.value FROM repo r
                 JOIN moved m ON m.from_did = r.from_did AND m.from_key = r.from_key
                   AND m.to_did = r.did AND m.to_key = r.key
                 JOIN forge f ON f.repo_did = r.from_did AND f.repo_key = r.from_key
               WHERE r.did = ?1 AND r.key = ?2
               GROUP BY f.uri)
             WHERE uri NOT IN down AND author NOT IN down
               AND coalesce(json_extract(value, '$.subject'), '') NOT IN down",
        )?;
        let entries = stmt
            .query_map([did, key], entry)?
            .collect::<rusqlite::Result<_>>()?;
        let mut stmt = db.prepare_cached("SELECT subject FROM block WHERE did = ?1")?;
        let blocked = stmt
            .query_map([did], |row| row.get(0))?
            .collect::<rusqlite::Result<Vec<String>>>()?;
        Ok(knotless::forge::moderate(did, key, &blocked, entries))
    }

    /// Whether any of `subjects`, accounts or records, is taken down here
    /// or by a labeler this server follows.
    pub fn down(&self, subjects: &[&str]) -> rusqlite::Result<bool> {
        let db = self.db.lock().unwrap();
        let mut stmt = db.prepare_cached("SELECT 1 FROM down WHERE subject = ?1")?;
        for s in subjects {
            if stmt.exists([s])? {
                return Ok(true);
            }
        }
        Ok(false)
    }

    /// Takes `subject` down, or puts it back: `value` is how the takedown
    /// names it. Returns the label that says so, and its place among those
    /// this server has given out.
    pub fn take_down(
        &self,
        src: &str,
        subject: &str,
        value: &Value,
        applied: bool,
    ) -> rusqlite::Result<(i64, Value)> {
        let db = self.db.lock().unwrap();
        let at = knotless::records::now();
        if applied {
            db.execute(
                "INSERT OR REPLACE INTO takedown VALUES (?1, ?2, ?3)",
                params![subject, value.to_string(), at],
            )?;
        } else {
            db.execute("DELETE FROM takedown WHERE subject = ?1", [subject])?;
        }
        let mut label = serde_json::json!({
            "ver": 1, "src": src, "uri": subject, "val": knotless::moderation::TAKEDOWN, "cts": at,
        });
        if !applied {
            label["neg"] = true.into();
        }
        db.execute(
            "INSERT INTO emitted (label) VALUES (?1)",
            [label.to_string()],
        )?;
        Ok((db.last_insert_rowid(), label))
    }

    /// How a takedown of `subject` named it, if it is taken down.
    pub fn taken_down(&self, subject: &str) -> rusqlite::Result<Option<Value>> {
        let db = self.db.lock().unwrap();
        db.query_row(
            "SELECT value FROM takedown WHERE subject = ?1",
            [subject],
            json,
        )
        .optional()
    }

    /// The labels this server gave out after `seq`, oldest first.
    pub fn emitted(&self, seq: i64) -> rusqlite::Result<Vec<(i64, Value)>> {
        let db = self.db.lock().unwrap();
        let mut stmt =
            db.prepare_cached("SELECT seq, label FROM emitted WHERE seq > ?1 ORDER BY seq")?;
        stmt.query_map([seq], numbered)?.collect()
    }

    /// Takedowns standing whose subject starts with `prefix`, as labels
    /// from `src`.
    pub fn labels(&self, src: &str, prefix: &str) -> rusqlite::Result<Vec<Value>> {
        let db = self.db.lock().unwrap();
        let mut stmt = db.prepare_cached(
            "SELECT subject, at FROM takedown WHERE substr(subject, 1, length(?1)) = ?1
             ORDER BY subject LIMIT 250",
        )?;
        let rows = stmt
            .query_map([prefix], pair)?
            .collect::<rusqlite::Result<Vec<(String, String)>>>()?;
        let label = |(uri, at)| {
            serde_json::json!({
                "ver": 1, "src": src, "uri": uri, "val": knotless::moderation::TAKEDOWN, "cts": at,
            })
        };
        Ok(rows.into_iter().map(label).collect())
    }

    /// Takes in a label from the labeler `src`, seen at `seq` of its stream.
    /// Only those that take things out of view are kept.
    pub fn label(&self, src: &str, seq: i64, label: &Value) -> rusqlite::Result<()> {
        let db = self.db.lock().unwrap();
        let val = label["val"].as_str().unwrap_or_default();
        let uri = label["uri"].as_str().unwrap_or_default();
        let known = [knotless::moderation::TAKEDOWN, knotless::moderation::HIDE];
        if label["src"] == src && known.contains(&val) {
            if label["neg"] == true {
                db.execute(
                    "DELETE FROM label WHERE src = ?1 AND uri = ?2 AND val = ?3",
                    [src, uri, val],
                )?;
            } else {
                db.execute(
                    "INSERT OR REPLACE INTO label VALUES (?1, ?2, ?3, ?4)",
                    params![src, uri, val, label["exp"].as_str()],
                )?;
            }
        }
        db.execute(
            "INSERT OR REPLACE INTO labeler VALUES (?1, ?2)",
            params![src, seq],
        )?;
        Ok(())
    }

    /// Where the labeler `src`'s stream was last read to.
    pub fn labeler_seq(&self, src: &str) -> rusqlite::Result<Option<i64>> {
        let db = self.db.lock().unwrap();
        db.query_row("SELECT seq FROM labeler WHERE src = ?1", [src], |row| {
            row.get(0)
        })
        .optional()
    }

    /// Keeps a report, as `com.atproto.moderation.createReport` answers it
    /// without its `id`, and returns it with the `id` it was given.
    pub fn report(&self, mut report: Value) -> rusqlite::Result<Value> {
        let db = self.db.lock().unwrap();
        db.execute(
            "INSERT INTO report (value) VALUES (?1)",
            [report.to_string()],
        )?;
        report["id"] = db.last_insert_rowid().into();
        Ok(report)
    }

    /// Reports, newest first, from before `before` when it is given.
    pub fn reports(&self, limit: usize, before: Option<i64>) -> rusqlite::Result<Vec<Value>> {
        let db = self.db.lock().unwrap();
        let mut stmt = db.prepare_cached(
            "SELECT id, value FROM report WHERE id < ?1 ORDER BY id DESC LIMIT ?2",
        )?;
        let before = before.unwrap_or(i64::MAX);
        stmt.query_map(params![before, limit as i64], numbered)?
            .map(|r| {
                r.map(|(id, mut report): (i64, Value)| {
                    report["id"] = id.into();
                    report
                })
            })
            .collect()
    }

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

    /// The repository record at `uri`.
    pub fn repo(&self, uri: &str) -> rusqlite::Result<Option<Value>> {
        let db = self.db.lock().unwrap();
        db.query_row("SELECT value FROM repo WHERE uri = ?1", [uri], |row| {
            row.get::<_, String>(0)
        })
        .optional()
        .map(|v| v.and_then(|v| serde_json::from_str(&v).ok()))
    }

    /// Repositories, newest first, whose name or description holds `query`.
    pub fn repos(&self, query: &str, limit: usize) -> rusqlite::Result<Vec<Listed>> {
        self.repos_page(query, limit, None).map(|(repos, _)| repos)
    }

    /// A page of [`Index::repos`] after `cursor`, and the cursor of the
    /// next page when there may be one.
    pub fn repos_page(
        &self,
        query: &str,
        limit: usize,
        cursor: Option<&str>,
    ) -> rusqlite::Result<(Vec<Listed>, Option<String>)> {
        let db = self.db.lock().unwrap();
        let pattern = format!("%{}%", query.replace(['%', '_'], ""));
        let (at, uri) = after(cursor);
        let mut stmt = db.prepare_cached(
            "SELECT did, name, description, fork_of, created_at, uri FROM repo
             WHERE (name LIKE ?1 OR description LIKE ?1)
               AND (created_at < ?2 OR (created_at = ?2 AND uri < ?3))
               AND did NOT IN down AND uri NOT IN down
             ORDER BY created_at DESC, uri DESC LIMIT ?4",
        )?;
        let rows: Vec<Listed> = stmt
            .query_map(params![pattern, at, uri, limit as i64], listed)?
            .collect::<rusqlite::Result<_>>()?;
        let next = (rows.len() == limit)
            .then(|| rows.last().map(|r| cursor_at(&r.created_at, &r.uri)))
            .flatten();
        Ok((rows, next))
    }

    /// Pulls, issues, comments and changes of state on any repository,
    /// newest first, with the owner and key of the repository each is on:
    /// without what its owner hides, with what was said about that, or
    /// what those its owner blocks wrote, as `forge` leaves them out. After
    /// `cursor`, as [`cursor_at`] makes one, if given.
    pub fn recent(
        &self,
        limit: usize,
        cursor: Option<&str>,
    ) -> rusqlite::Result<Vec<(Entry, String, String)>> {
        let (at, uri) = after(cursor);
        let db = self.db.lock().unwrap();
        let mut stmt = db.prepare_cached(
            "SELECT f.uri, f.author, f.value, f.repo_did, f.repo_key FROM forge f
             WHERE f.uri NOT LIKE '%/dev.knotless.hide/%' AND f.uri NOT LIKE '%/dev.knotless.block/%'
               AND (json_extract(f.value, '$.createdAt') < ?2
                 OR (json_extract(f.value, '$.createdAt') = ?2 AND f.uri < ?3))
               AND f.uri NOT IN down AND f.author NOT IN down
               AND coalesce(json_extract(f.value, '$.subject'), '') NOT IN down
               AND f.author NOT IN (SELECT subject FROM block WHERE did = f.repo_did)
               AND NOT EXISTS (SELECT 1 FROM forge o
                 WHERE o.repo_did = f.repo_did AND o.repo_key = f.repo_key AND o.author = f.repo_did
                   AND ((o.uri LIKE '%/dev.knotless.block/%'
                         AND json_extract(o.value, '$.subject') = f.author)
                     OR (o.uri LIKE '%/dev.knotless.hide/%'
                         AND json_extract(o.value, '$.subject')
                           IN (f.uri, coalesce(json_extract(f.value, '$.subject'), '')))))
             ORDER BY json_extract(f.value, '$.createdAt') DESC, f.uri DESC LIMIT ?1",
        )?;
        stmt.query_map(params![limit as i64, at, uri], recent)?
            .collect()
    }

    /// Refs that moved on repositories that stand, newest first; after
    /// `cursor`, as [`cursor_at`] makes one, if given.
    pub fn pushes(&self, limit: usize, cursor: Option<&str>) -> rusqlite::Result<Vec<Push>> {
        let db = self.db.lock().unwrap();
        let (at, uri) = after(cursor);
        let mut stmt = db.prepare_cached(
            "SELECT r.uri, p.did, r.name, r.fork_of, p.ref, p.before, p.after, p.created_at, p.uri
               FROM push p JOIN repo r ON r.did = p.did AND r.key = p.key
             WHERE p.did NOT IN down AND r.uri NOT IN down
               AND (p.created_at < ?2 OR (p.created_at = ?2 AND p.uri < ?3))
             ORDER BY p.created_at DESC, p.uri DESC LIMIT ?1",
        )?;
        stmt.query_map(params![limit as i64, at, uri], push)?
            .collect()
    }

    /// Where the repository at `repo` had `refname` just before `at`, as
    /// far as the index saw.
    pub fn tip_before(
        &self,
        repo: &str,
        refname: &str,
        at: &str,
    ) -> rusqlite::Result<Option<String>> {
        let db = self.db.lock().unwrap();
        let mut stmt = db.prepare_cached(
            "SELECT p.after FROM push p JOIN repo r ON r.did = p.did AND r.key = p.key
             WHERE r.uri = ?1 AND p.ref = ?2 AND p.created_at < ?3
             ORDER BY p.created_at DESC LIMIT 1",
        )?;
        let tip = stmt
            .query_row([repo, refname, at], |row| row.get(0))
            .optional()?;
        Ok(tip.flatten())
    }

    /// Forks of `did`'s repository `key`.
    pub fn forks(&self, did: &str, key: &str) -> rusqlite::Result<Vec<Listed>> {
        let db = self.db.lock().unwrap();
        let mut stmt = db.prepare_cached(
            "SELECT did, name, description, fork_of, created_at, uri FROM repo
             WHERE fork_did = ?1 AND fork_key = ?2 AND did NOT IN down AND uri NOT IN down
             ORDER BY created_at",
        )?;
        stmt.query_map([did, key], listed)?.collect()
    }
}

/// A number and the JSON after it.
fn numbered(row: &rusqlite::Row) -> rusqlite::Result<(i64, Value)> {
    Ok((row.get(0)?, json(row)?))
}

fn pair(row: &rusqlite::Row) -> rusqlite::Result<(String, String)> {
    Ok((row.get(0)?, row.get(1)?))
}

/// The JSON in a row's last column.
fn json(row: &rusqlite::Row) -> rusqlite::Result<Value> {
    let text: String = row.get(row.as_ref().column_count() - 1)?;
    Ok(serde_json::from_str(&text).unwrap_or_default())
}

fn entry(row: &rusqlite::Row) -> rusqlite::Result<Entry> {
    let value: String = row.get(2)?;
    Ok(Entry {
        uri: row.get(0)?,
        author: row.get(1)?,
        value: serde_json::from_str(&value).unwrap_or_default(),
    })
}

fn push(row: &rusqlite::Row) -> rusqlite::Result<Push> {
    Ok(Push {
        repo: row.get(0)?,
        did: row.get(1)?,
        name: row.get(2)?,
        fork_of: row.get(3)?,
        refname: row.get(4)?,
        from: row.get(5)?,
        to: row.get(6)?,
        created_at: row.get(7)?,
        uri: row.get(8)?,
    })
}

fn recent(row: &rusqlite::Row) -> rusqlite::Result<(Entry, String, String)> {
    Ok((entry(row)?, row.get(3)?, row.get(4)?))
}

/// Where a page cut after the record `uri`, made at `at`, starts the next:
/// newest first, ties broken by URI, so the next starts strictly after it.
pub fn cursor_at(at: &str, uri: &str) -> String {
    format!("{at} {uri}")
}

/// The time and URI a cursor is after, or past everything without one.
fn after(cursor: Option<&str>) -> (&str, &str) {
    cursor
        .and_then(|c| c.split_once(' '))
        .unwrap_or(("\u{10ffff}", ""))
}

fn listed(row: &rusqlite::Row) -> rusqlite::Result<Listed> {
    Ok(Listed {
        did: row.get(0)?,
        name: row.get(1)?,
        description: row.get(2)?,
        fork_of: row.get(3)?,
        created_at: row.get(4)?,
        uri: row.get(5)?,
    })
}

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

    #[test]
    fn not_a_database() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("i.db");
        std::fs::write(&path, vec![7u8; 4096]).unwrap();
        assert!(Index::open(&path).is_err());
    }

    #[test]
    fn a_database_of_another_shape() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("i.db");
        // Claiming to be this version, so it is kept: every use fails.
        Connection::open(&path)
            .unwrap()
            .execute_batch(&format!(
                "PRAGMA user_version = {VERSION};
                 CREATE TABLE repo (did, key, fork_did, fork_key, created_at);
                 CREATE TABLE forge (repo_did, repo_key);"
            ))
            .unwrap();
        let index = Index::open(&path).unwrap();
        assert!(index.put("did:a", nsid::REPO, "r", &json!({})).is_err());
        let issue = json!({"repo": {"uri": "at://did:a/dev.knotless.repo/r", "key": "k"}});
        assert!(index.put("did:a", nsid::ISSUE, "1", &issue).is_err());
        assert!(index.repos("", 1).is_err());
        assert!(index.forks("did:a", "k").is_err());
        assert!(index.forge("did:a", "k").is_err());
        assert!(index.name("did:a", "k").is_err());
        assert!(index.recent(1, None).is_err());
        assert!(index.pushes(1, None).is_err());
    }

    #[test]
    fn moderation() {
        let dir = tempfile::tempdir().unwrap();
        let index = Index::open(&dir.path().join("i.db")).unwrap();
        let named = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k"});
        let issue = |did: &str, rkey: &str| {
            index
                .put(
                    did,
                    nsid::ISSUE,
                    rkey,
                    &json!({"repo": named, "title": "t"}),
                )
                .unwrap();
            format!("at://{did}/dev.knotless.issue/{rkey}")
        };
        let forge = || -> Vec<String> {
            index
                .forge("did:a", "k")
                .unwrap()
                .into_iter()
                .map(|e| e.uri)
                .collect()
        };
        // Blocks from all the owner's repositories, and from this one.
        let (first, second) = (issue("did:b", "1"), issue("did:c", "2"));
        index
            .put("did:a", nsid::BLOCK, "b", &json!({"subject": "did:c"}))
            .unwrap();
        assert_eq!(forge(), std::slice::from_ref(&first));
        index.delete("did:a", nsid::BLOCK, "b").unwrap();
        assert_eq!(forge().len(), 2);
        index
            .put(
                "did:a",
                nsid::BLOCK,
                "c",
                &json!({"subject": "did:c", "repo": named, "createdAt": "t"}),
            )
            .unwrap();
        assert_eq!(
            forge(),
            [first.clone(), "at://did:a/dev.knotless.block/c".into()]
        );
        index.delete("did:a", nsid::BLOCK, "c").unwrap();
        assert_eq!(forge().len(), 2);

        // Takedowns, and the labels that say so.
        let subject = json!({"$type": "com.atproto.repo.strongRef", "uri": first});
        let (seq, label) = index
            .take_down("did:web:me", &first, &subject, true)
            .unwrap();
        assert_eq!((seq, label["val"].as_str()), (1, Some("!takedown")));
        assert_eq!(forge(), std::slice::from_ref(&second));
        assert!(index.down(&["did:x", &first]).unwrap());
        assert_eq!(index.taken_down(&first).unwrap(), Some(subject.clone()));
        assert_eq!(index.labels("did:web:me", "at://did:b/").unwrap().len(), 1);
        assert!(index.labels("did:web:me", "did:").unwrap().is_empty());
        let (_, undone) = index
            .take_down("did:web:me", &first, &subject, false)
            .unwrap();
        assert_eq!(undone["neg"], true);
        assert_eq!(index.taken_down(&first).unwrap(), None);
        assert_eq!(index.emitted(0).unwrap().len(), 2);
        assert_eq!(index.emitted(1).unwrap()[0].1, undone);

        // Labels from a labeler: only its own, only those that hide, and
        // not past their expiry.
        let label = |src: &str, uri: &str, val: &str, extra: Value| {
            let mut l = json!({"src": src, "uri": uri, "val": val});
            l.as_object_mut()
                .unwrap()
                .extend(extra.as_object().unwrap().clone());
            l
        };
        assert_eq!(index.labeler_seq("did:web:l").unwrap(), None);
        let none = json!({});
        index
            .label(
                "did:web:l",
                3,
                &label("did:web:other", "did:c", "!takedown", none.clone()),
            )
            .unwrap();
        index
            .label(
                "did:web:l",
                4,
                &label("did:web:l", "did:c", "porn", none.clone()),
            )
            .unwrap();
        index
            .label(
                "did:web:l",
                5,
                &label(
                    "did:web:l",
                    "did:b",
                    "!hide",
                    json!({"exp": "2000-01-01T00:00:00Z"}),
                ),
            )
            .unwrap();
        assert_eq!(forge().len(), 2);
        index
            .label(
                "did:web:l",
                6,
                &label("did:web:l", "did:c", "!hide", none.clone()),
            )
            .unwrap();
        assert_eq!(forge(), std::slice::from_ref(&first));
        index
            .label(
                "did:web:l",
                7,
                &label("did:web:l", "did:c", "!hide", json!({"neg": true})),
            )
            .unwrap();
        assert_eq!(forge().len(), 2);
        assert_eq!(index.labeler_seq("did:web:l").unwrap(), Some(7));

        // Reports, newest first, a page at a time.
        for n in 0..3 {
            let report = index.report(json!({"reason": n})).unwrap();
            assert_eq!(report["id"], n + 1);
        }
        let page = index.reports(2, None).unwrap();
        assert_eq!(
            (page[0]["id"].as_i64(), page[1]["reason"].as_i64()),
            (Some(3), Some(1))
        );
        assert_eq!(index.reports(2, Some(2)).unwrap()[0]["id"], 1);

        // Takedowns and reports outlive a rebuild; the cache does not.
        drop(index);
        let db = Connection::open(dir.path().join("i.db")).unwrap();
        db.execute_batch("PRAGMA user_version = 1").unwrap();
        drop(db);
        let index = Index::open(&dir.path().join("i.db")).unwrap();
        assert_eq!(index.reports(10, None).unwrap().len(), 3);
        assert_eq!(index.emitted(0).unwrap().len(), 2);
        assert_eq!(index.labeler_seq("did:web:l").unwrap(), None);
    }

    #[test]
    fn what_happened() {
        let dir = tempfile::tempdir().unwrap();
        let index = Index::open(&dir.path().join("i.db")).unwrap();
        let repo = json!({"name": "r", "key": "k", "createdAt": "2026-01-01T00:00:00Z"});
        index.put("did:a", nsid::REPO, "r", &repo).unwrap();
        let claim = |rkey: &str, from: Option<&str>, to: Option<&str>, at: &str| {
            let mut c =
                json!({"repo": "k", "name": "refs/heads/main", "generation": 0, "createdAt": at});
            if let Some(f) = from {
                c["from"] = json!(f);
            }
            if let Some(t) = to {
                c["to"] = json!(t);
            }
            index.put("did:a", nsid::CLAIM, rkey, &c).unwrap();
        };
        claim("k.m.0", None, Some("a1"), "2026-01-02T00:00:00Z");
        claim("k.m.1", Some("a1"), Some("b2"), "2026-01-03T00:00:00Z");
        // Of a repository the index does not know.
        index
            .put(
                "did:a",
                nsid::CLAIM,
                "x.m.0",
                &json!({"repo": "x", "name": "refs/heads/main", "createdAt": "2026-01-04T00:00:00Z"}),
            )
            .unwrap();
        let pushes = index.pushes(10, None).unwrap();
        assert_eq!(
            pushes,
            [
                Push {
                    uri: "at://did:a/dev.knotless.claim/k.m.1".into(),
                    repo: "at://did:a/dev.knotless.repo/r".into(),
                    did: "did:a".into(),
                    name: "r".into(),
                    fork_of: None,
                    refname: "refs/heads/main".into(),
                    from: Some("a1".into()),
                    to: Some("b2".into()),
                    created_at: "2026-01-03T00:00:00Z".into(),
                },
                Push {
                    uri: "at://did:a/dev.knotless.claim/k.m.0".into(),
                    repo: "at://did:a/dev.knotless.repo/r".into(),
                    did: "did:a".into(),
                    name: "r".into(),
                    fork_of: None,
                    refname: "refs/heads/main".into(),
                    from: None,
                    to: Some("a1".into()),
                    created_at: "2026-01-02T00:00:00Z".into(),
                },
            ]
        );
        let uri = "at://did:a/dev.knotless.repo/r";
        let main = "refs/heads/main";
        let tip = |at| index.tip_before(uri, main, at).unwrap();
        assert_eq!(tip("2026-01-03T00:00:00Z"), Some("a1".into()));
        assert_eq!(tip("2026-01-02T00:00:00Z"), None);
        claim("k.m.2", Some("b2"), None, "2026-01-03T12:00:00Z");
        assert_eq!(tip("2026-01-04T00:00:00Z"), None);
        index.delete("did:a", nsid::CLAIM, "k.m.2").unwrap();
        // Compaction drops claims; the history stays.
        index.delete("did:a", nsid::CLAIM, "k.m.0").unwrap();
        assert_eq!(index.pushes(10, None).unwrap().len(), 3);
        assert_eq!(index.pushes(1, None).unwrap().len(), 1);
        let all = index.pushes(10, None).unwrap();
        let at = cursor_at(&all[0].created_at, &all[0].uri);
        assert_eq!(index.pushes(10, Some(&at)).unwrap(), all[1..]);

        // Pulls, issues and what was said, newest first, without what the
        // owner hides or blocks, or those records themselves.
        let named = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k"});
        let at = |collection: &str, rkey: &str| format!("at://did:b/{collection}/{rkey}");
        for (did, collection, rkey, at, subject) in [
            ("did:b", nsid::ISSUE, "1", "2026-01-05", None),
            (
                "did:b",
                nsid::COMMENT,
                "2",
                "2026-01-07",
                Some(at(nsid::ISSUE, "1")),
            ),
            ("did:b", nsid::PULL, "4", "2026-01-06", None),
            ("did:b", nsid::ISSUE, "5", "2026-01-09", None),
            (
                "did:b",
                nsid::COMMENT,
                "6",
                "2026-01-10",
                Some(at(nsid::ISSUE, "5")),
            ),
            (
                "did:a",
                nsid::HIDE,
                "7",
                "2026-01-11",
                Some(at(nsid::ISSUE, "5")),
            ),
            ("did:d", nsid::ISSUE, "8", "2026-01-12", None),
            (
                "did:a",
                nsid::BLOCK,
                "9",
                "2026-01-13",
                Some("did:d".into()),
            ),
            ("did:e", nsid::ISSUE, "10", "2026-01-14", None),
            // Only the owner's are heeded.
            (
                "did:b",
                nsid::HIDE,
                "11",
                "2026-01-15",
                Some(at(nsid::PULL, "4")),
            ),
        ] {
            let value = json!({"repo": named, "createdAt": at, "subject": subject});
            index.put(did, collection, rkey, &value).unwrap();
        }
        index
            .put("did:a", nsid::BLOCK, "12", &json!({"subject": "did:e"}))
            .unwrap();
        let recent = index.recent(10, None).unwrap();
        let kinds: Vec<&str> = recent
            .iter()
            .map(|(e, _, _)| e.uri.rsplit('/').nth(1).unwrap())
            .collect();
        assert_eq!(kinds, [nsid::COMMENT, nsid::PULL, nsid::ISSUE]);
        assert_eq!((recent[0].1.as_str(), recent[0].2.as_str()), ("did:a", "k"));
        assert_eq!(index.recent(1, None).unwrap().len(), 1);
        // Paged by time and URI, the next page starting strictly after.
        let first = &index.recent(1, None).unwrap()[0].0;
        let at = cursor_at(first.value["createdAt"].as_str().unwrap(), &first.uri);
        let rest = index.recent(10, Some(&at)).unwrap();
        assert_eq!(rest.len(), 2);
        assert_eq!(rest[0].0.uri, recent[1].0.uri);

        // Gone with the repository.
        index.delete("did:a", nsid::REPO, "r").unwrap();
        assert!(index.pushes(10, None).unwrap().is_empty());
    }

    #[test]
    fn an_index_of_another_version_is_rebuilt() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("i.db");
        Connection::open(&path)
            .unwrap()
            .execute_batch("CREATE TABLE repo (did, created_at); CREATE TABLE forge (repo);")
            .unwrap();
        let index = Index::open(&path).unwrap();
        index
            .put("did:a", nsid::REPO, "r", &json!({"key": "k"}))
            .unwrap();
        assert_eq!(index.name("did:a", "k").unwrap().as_deref(), Some("r"));
    }

    #[test]
    fn records_in_and_out() {
        let dir = tempfile::tempdir().unwrap();
        let index = Index::open(&dir.path().join("i.db")).unwrap();
        let repo =
            json!({"name": "r", "key": "k", "description": "a thing", "createdAt": "2026-01-01"});
        index.put("did:a", nsid::REPO, "r", &repo).unwrap();
        let uri = "at://did:a/dev.knotless.repo/r";
        let named = json!({"uri": uri, "key": "k"});
        let fork = json!({"name": "r", "key": "k2", "forkOf": named, "createdAt": "2026-01-02"});
        index.put("did:b", nsid::REPO, "r", &fork).unwrap();
        index
            .put(
                "did:b",
                nsid::ISSUE,
                "1",
                &json!({"repo": named, "title": "t"}),
            )
            .unwrap();
        // About a repository since deleted, whose name was taken again.
        let gone = json!({"uri": uri, "key": "k0"});
        index
            .put(
                "did:b",
                nsid::ISSUE,
                "3",
                &json!({"repo": gone, "title": "old"}),
            )
            .unwrap();
        index
            .put("did:b", nsid::ISSUE, "2", &json!({"title": "no repo"}))
            .unwrap();
        assert_eq!(index.repo(uri).unwrap(), Some(repo));
        assert_eq!(index.repo("at://did:x/dev.knotless.repo/r").unwrap(), None);
        assert_eq!(index.forge("did:a", "k").unwrap().len(), 1);
        assert_eq!(index.name("did:a", "k").unwrap().as_deref(), Some("r"));
        assert_eq!(index.name("did:a", "k0").unwrap(), None);
        let all = index.repos("", 10).unwrap();
        assert_eq!(
            all.iter().map(|r| r.did.as_str()).collect::<Vec<_>>(),
            ["did:b", "did:a"]
        );
        assert_eq!(index.repos("THING", 10).unwrap().len(), 1);
        let (page, cursor) = index.repos_page("", 1, None).unwrap();
        assert_eq!(
            (page[0].did.as_str(), cursor.as_deref()),
            ("did:b", Some("2026-01-02 at://did:b/dev.knotless.repo/r"))
        );
        let (page, cursor) = index.repos_page("", 1, cursor.as_deref()).unwrap();
        assert_eq!(page[0].did, "did:a");
        assert_eq!(
            index.repos_page("", 1, cursor.as_deref()).unwrap(),
            (vec![], None)
        );
        let listed = serde_json::to_value(&page[0]).unwrap();
        assert!(listed.get("forkOf").is_none() && listed["description"] == "a thing");
        assert_eq!(index.forks("did:a", "k").unwrap()[0].did, "did:b");
        let all = index.repos("", 10).unwrap();
        assert_eq!(all[0].fork_of.as_deref(), Some(uri));
        index.delete("did:b", nsid::ISSUE, "1").unwrap();
        index.delete("did:a", nsid::REPO, "r").unwrap();
        assert!(index.forge("did:a", "k").unwrap().is_empty());
        assert_eq!(index.repos("", 10).unwrap().len(), 1);
        // A second open finds the tables already there.
        drop(index);
        assert!(Index::open(&dir.path().join("i.db")).is_ok());
        assert!(Index::open(dir.path()).is_err());
    }
}