chadtmiller.com/knotless
Git hosted in your atproto account, on any stock PDS, fast enough for agents.
git clone knotless://chadtmiller.com/knotlessneeds the knotless git helper. set it up
//! 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());
}
}