vineyard

public
Star0
Watch0
2 commits 1 branch 0 tags 0 releases64.9 KiB
use std::cmp::Ordering;
use std::collections::{BTreeSet, HashSet};
use std::error::Error;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use anyhow::{Context as _, Result};
use nostr::event::{Event, EventId, Kind, Signature, Tags};
use nostr::filter::Filter;
use nostr::key::PublicKey;
use nostr::types::Timestamp;
use nostr_database::{DatabaseEventStatus, RejectedReason, SaveEventStatus};
use rusqlite::{
    Connection, OptionalExtension as _, Params, Row, Transaction, params, params_from_iter,
};
use tokio::sync::{Mutex, MutexGuard};
use tokio::task::spawn_blocking;

use crate::config::relay_url;
use crate::query::EventQuery;

const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS events(
    id BLOB PRIMARY KEY NOT NULL CHECK(length(id) = 32),
    pubkey BLOB NOT NULL CHECK(length(pubkey) = 32),
    created_at INTEGER NOT NULL CHECK(created_at > 0),
    kind INTEGER NOT NULL CHECK(kind BETWEEN 0 AND 65535),
    content TEXT NOT NULL,
    tags JSONB NOT NULL,
    sig BLOB NOT NULL CHECK(length(sig) = 64)
) WITHOUT ROWID;

CREATE INDEX IF NOT EXISTS idx_events_created_at ON events(created_at DESC);
CREATE INDEX IF NOT EXISTS idx_events_pubkey_created_at ON events(pubkey, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_events_pubkey_kind_created_at
ON events(pubkey, kind, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_events_kind_created_at ON events(kind, created_at DESC);

DROP INDEX IF EXISTS idx_events_pubkey;
DROP INDEX IF EXISTS idx_events_kind;

CREATE TABLE IF NOT EXISTS event_tags(
    event_id BLOB NOT NULL CHECK(length(event_id) = 32),
    tag_name TEXT NOT NULL CHECK(length(tag_name) = 1),
    tag_value TEXT NOT NULL,
    FOREIGN KEY(event_id) REFERENCES events(id) ON DELETE CASCADE,
    PRIMARY KEY(tag_name, tag_value, event_id)
);

CREATE INDEX IF NOT EXISTS idx_tags_event_id ON event_tags(event_id);

CREATE TABLE IF NOT EXISTS deleted_ids(
    event_id BLOB PRIMARY KEY NOT NULL CHECK(length(event_id) = 32)
) WITHOUT ROWID;

CREATE TABLE IF NOT EXISTS deleted_coordinates(
    pubkey BLOB NOT NULL CHECK(length(pubkey) = 32),
    kind INTEGER NOT NULL CHECK(kind BETWEEN 0 AND 65535),
    identifier TEXT NOT NULL,
    deleted_at INTEGER NOT NULL CHECK(deleted_at > 0),
    PRIMARY KEY(pubkey, kind, identifier)
) WITHOUT ROWID;

DROP TRIGGER IF EXISTS sanctum_deleted_after_insert;
DROP TABLE IF EXISTS sanctum_deleted_events;

CREATE TABLE IF NOT EXISTS vineyard_delivery(
    sequence INTEGER PRIMARY KEY AUTOINCREMENT,
    event_id BLOB NOT NULL UNIQUE REFERENCES events(id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS vineyard_peers(
    peer TEXT PRIMARY KEY,
    delivered INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE IF NOT EXISTS vineyard_checkpoints(
    peer TEXT,
    scope TEXT,
    until INTEGER NOT NULL,
    PRIMARY KEY(peer, scope)
);

CREATE TABLE IF NOT EXISTS vineyard_retries(
    peer TEXT,
    sequence INTEGER REFERENCES vineyard_delivery(sequence) ON DELETE CASCADE,
    retry_at INTEGER,
    error TEXT,
    PRIMARY KEY(peer, sequence)
);

CREATE TABLE IF NOT EXISTS vineyard_discovered_peers(
    signer BLOB,
    url TEXT,
    discovered_at INTEGER,
    PRIMARY KEY(signer, url)
) WITHOUT ROWID;

CREATE INDEX IF NOT EXISTS vineyard_retries_due
ON vineyard_retries(peer, retry_at, sequence);

CREATE INDEX IF NOT EXISTS vineyard_discovered_urls
ON vineyard_discovered_peers(url);

CREATE TRIGGER IF NOT EXISTS vineyard_delivery_after_insert
AFTER INSERT ON events
BEGIN
    INSERT OR IGNORE INTO vineyard_delivery(event_id) VALUES (NEW.id);
END;

";

const BACKFILL_DELIVERY: &str = "
INSERT INTO vineyard_delivery(event_id)
SELECT event.id
FROM events AS event
LEFT JOIN vineyard_delivery AS delivery ON delivery.event_id = event.id
WHERE delivery.event_id IS NULL
ORDER BY
    CASE WHEN event.kind = 30617 THEN 0 ELSE 1 END,
    event.created_at,
    event.id
";

const DISCOVERED_RELAY_TAGS: &str = "
SELECT event.pubkey, json_extract(tag.value, '$[1]'), event.created_at
FROM events AS event, json_each(event.tags) AS tag
WHERE event.kind = 10002
  AND json_extract(tag.value, '$[0]') = 'r'
  AND typeof(json_extract(tag.value, '$[1]')) = 'text'
ORDER BY event.created_at DESC, event.id
";

const DUE_DELIVERIES: &str = "
SELECT delivery.sequence, delivery.event_id
FROM vineyard_delivery AS delivery
JOIN vineyard_retries AS retry ON retry.sequence = delivery.sequence
WHERE retry.peer = ?1 AND retry.retry_at <= ?2
ORDER BY delivery.sequence
LIMIT ?3
";

const NEW_DELIVERIES: &str = "
SELECT sequence, event_id
FROM vineyard_delivery
WHERE sequence > ?1
ORDER BY sequence
LIMIT ?2
";

const REPOSITORIES: &str = "
SELECT lower(hex(event.pubkey)), tag.tag_value, event.tags
FROM events AS event
JOIN event_tags AS tag ON event.id = tag.event_id
WHERE event.kind = 30617 AND tag.tag_name = 'd'
ORDER BY 1, 2, event.created_at DESC, event.id
";

type Delivery = (u64, EventId);

#[derive(Debug, Clone)]
pub struct Journal {
    connection: Arc<Mutex<Connection>>,
    writes: Arc<Mutex<()>>,
}

struct StoredEvent {
    id: Vec<u8>,
    pubkey: Vec<u8>,
    created_at: i64,
    kind: i64,
    content: String,
    tags: String,
    signature: Vec<u8>,
}

impl Journal {
    pub async fn open(
        path: PathBuf,
        max_database_bytes: u64,
        max_discovered_peers: usize,
        excluded_peers: Vec<String>,
        writes: Arc<Mutex<()>>,
    ) -> Result<Self> {
        let connection = spawn_blocking(move || {
            open_connection(
                &path,
                max_database_bytes,
                max_discovered_peers,
                &excluded_peers,
            )
        })
        .await
        .context("journal task panicked")??;

        Ok(Self {
            connection: Arc::new(Mutex::new(connection)),
            writes,
        })
    }

    pub async fn pending(&self, peer: &str, limit: usize) -> Result<Vec<Delivery>> {
        let peer_url = peer.to_owned();
        self.run(move |connection| pending_deliveries(connection, &peer_url, limit))
            .await
    }

    pub async fn acknowledge(&self, peer: &str, sequence: u64) -> Result<()> {
        let peer_url = peer.to_owned();
        self.write(move |connection| acknowledge_delivery(connection, &peer_url, sequence))
            .await
    }

    pub async fn defer(&self, peer: &str, sequence: u64, error: &str, retry_at: u64) -> Result<()> {
        let peer_url = peer.to_owned();
        let message = error.chars().take(512).collect::<String>();
        self.write(move |connection| {
            defer_delivery(connection, &peer_url, sequence, &message, retry_at)
        })
        .await
    }

    pub async fn checkpoint(&self, peer: &str, scope: &str) -> Result<Option<u64>> {
        let peer_url = peer.to_owned();
        let scope_name = scope.to_owned();
        self.run(move |connection| read_checkpoint(connection, &peer_url, &scope_name))
            .await
    }

    pub async fn advance(&self, peer: &str, scope: &str, until: u64) -> Result<()> {
        let peer_url = peer.to_owned();
        let scope_name = scope.to_owned();
        self.write(move |connection| write_checkpoint(connection, &peer_url, &scope_name, until))
            .await
    }

    pub async fn repositories(&self, limit: usize) -> Result<Vec<(String, String)>> {
        self.run(move |connection| list_repositories(connection, limit))
            .await
    }

    pub async fn healthy(&self) -> bool {
        self.run(|connection| Ok(connection.query_row("SELECT 1", [], |_| Ok(()))?))
            .await
            .is_ok()
    }

    pub async fn save_locked(
        &self,
        event: Event,
        _write: &MutexGuard<'_, ()>,
    ) -> Result<SaveEventStatus> {
        self.run(move |connection| save_event(connection, &event))
            .await
    }

    pub async fn status(&self, event_id: EventId) -> Result<DatabaseEventStatus> {
        self.run(move |connection| event_status(connection, &event_id))
            .await
    }

    pub async fn event(&self, event_id: EventId) -> Result<Option<Event>> {
        self.run(move |connection| event_by_id(connection, &event_id))
            .await
    }

    pub async fn count(&self, filter: Filter) -> Result<usize> {
        self.run(move |connection| Ok(query_events(connection, &filter, usize::MAX)?.len()))
            .await
    }

    pub async fn query(&self, filter: Filter, max_bytes: usize) -> Result<BTreeSet<Event>> {
        self.run(move |connection| query_events(connection, &filter, max_bytes))
            .await
    }

    pub async fn replace_discovered_locked(
        &self,
        signer: PublicKey,
        urls: Vec<String>,
        max: usize,
        _write: &MutexGuard<'_, ()>,
    ) -> Result<()> {
        let signer_bytes = signer.as_bytes().to_vec();
        self.run(move |connection| replace_discovered(connection, &signer_bytes, urls, max))
            .await
    }

    pub async fn discovered(&self, limit: usize) -> Result<Vec<String>> {
        self.run(move |connection| list_discovered(connection, limit))
            .await
    }

    pub async fn forget_discovered(&self, urls: Vec<String>) -> Result<()> {
        self.write(move |connection| forget_discovered(connection, urls))
            .await
    }

    async fn write<T, F>(&self, operation: F) -> Result<T>
    where
        T: Send + 'static,
        F: FnOnce(&mut Connection) -> Result<T> + Send + 'static,
    {
        let _write = self.writes.lock().await;
        self.run(operation).await
    }

    async fn run<T, F>(&self, operation: F) -> Result<T>
    where
        T: Send + 'static,
        F: FnOnce(&mut Connection) -> Result<T> + Send + 'static,
    {
        let shared_connection = Arc::clone(&self.connection);
        let mut connection = shared_connection.lock_owned().await;
        spawn_blocking(move || operation(&mut connection))
            .await
            .context("journal task panicked")?
    }
}

fn open_connection(
    path: &Path,
    max_database_bytes: u64,
    max_discovered_peers: usize,
    excluded_peers: &[String],
) -> Result<Connection> {
    let mut connection =
        Connection::open(path).with_context(|| format!("opening journal {}", path.display()))?;
    configure(&connection, max_database_bytes)?;
    migrate(&mut connection, max_discovered_peers, excluded_peers)?;
    Ok(connection)
}

fn pending_deliveries(connection: &Connection, peer: &str, limit: usize) -> Result<Vec<Delivery>> {
    if limit == 0 {
        return Ok(Vec::new());
    }

    let retry_limit = limit / 2;
    let mut pending = if retry_limit == 0 {
        Vec::new()
    } else {
        read_deliveries(
            connection,
            DUE_DELIVERIES,
            params![peer, unix_seconds(), integer(retry_limit)?],
        )?
    };

    let remaining = limit - pending.len();
    if remaining == 0 {
        return Ok(pending);
    }

    let delivered = read_delivered(connection, peer)?;
    pending.extend(read_deliveries(
        connection,
        NEW_DELIVERIES,
        params![delivered, integer(remaining)?],
    )?);
    Ok(pending)
}

fn read_deliveries<P>(connection: &Connection, sql: &str, parameters: P) -> Result<Vec<Delivery>>
where
    P: Params,
{
    let mut statement = connection.prepare(sql)?;
    let mut rows = statement.query(parameters)?;
    let mut deliveries = Vec::new();

    while let Some(row) = rows.next()? {
        let sequence = row.get::<_, i64>(0)?;
        let event_id = row.get::<_, Vec<u8>>(1)?;
        deliveries.push((
            u64::try_from(sequence).context("negative delivery sequence")?,
            EventId::from_slice(&event_id)
                .map_err(|error| anyhow::anyhow!("invalid event id {error}"))?,
        ));
    }

    Ok(deliveries)
}

fn read_delivered(connection: &Connection, peer: &str) -> Result<i64> {
    Ok(connection
        .query_row(
            "SELECT delivered FROM vineyard_peers WHERE peer = ?1",
            [peer],
            |row| row.get(0),
        )
        .optional()?
        .unwrap_or(0))
}

fn acknowledge_delivery(connection: &mut Connection, peer: &str, sequence: u64) -> Result<()> {
    let stored_sequence = integer(sequence)?;
    let transaction = connection.transaction()?;
    transaction.execute(
        "DELETE FROM vineyard_retries WHERE peer = ?1 AND sequence = ?2",
        params![peer, stored_sequence],
    )?;
    advance_delivery(&transaction, peer, stored_sequence)?;
    transaction.commit()?;
    Ok(())
}

fn defer_delivery(
    connection: &mut Connection,
    peer: &str,
    sequence: u64,
    error: &str,
    retry_at: u64,
) -> Result<()> {
    let stored_sequence = integer(sequence)?;
    let stored_retry_at = integer(retry_at)?;
    let transaction = connection.transaction()?;
    advance_delivery(&transaction, peer, stored_sequence)?;
    transaction.execute(
        "
            INSERT INTO vineyard_retries(peer, sequence, retry_at, error)
            VALUES (?1, ?2, ?3, ?4)
            ON CONFLICT(peer, sequence) DO UPDATE SET
                retry_at = excluded.retry_at,
                error = excluded.error
            ",
        params![peer, stored_sequence, stored_retry_at, error],
    )?;
    transaction.commit()?;
    Ok(())
}

fn read_checkpoint(connection: &Connection, peer: &str, scope: &str) -> Result<Option<u64>> {
    let checkpoint = connection
        .query_row(
            "SELECT until FROM vineyard_checkpoints WHERE peer = ?1 AND scope = ?2",
            params![peer, scope],
            |row| row.get::<_, i64>(0),
        )
        .optional()?;

    checkpoint
        .map(|value| u64::try_from(value).context("negative checkpoint"))
        .transpose()
}

fn write_checkpoint(connection: &Connection, peer: &str, scope: &str, until: u64) -> Result<()> {
    connection.execute(
        "
            INSERT INTO vineyard_checkpoints(peer, scope, until)
            VALUES (?1, ?2, ?3)
            ON CONFLICT(peer, scope) DO UPDATE SET
                until = max(vineyard_checkpoints.until, excluded.until)
            ",
        params![peer, scope, integer(until)?],
    )?;
    Ok(())
}

fn list_repositories(connection: &Connection, limit: usize) -> Result<Vec<(String, String)>> {
    let mut statement = connection.prepare(REPOSITORIES)?;
    let mut rows = statement.query([])?;
    let mut repositories = Vec::new();
    let mut seen = BTreeSet::new();
    while let Some(row) = rows.next()? {
        let tags = serde_json::from_str::<Tags>(&row.get::<_, String>(2)?)?;
        if tags
            .expiration()
            .is_some_and(|expiration| expiration < Timestamp::now())
        {
            continue;
        }
        let repository = (row.get(0)?, row.get(1)?);
        if !seen.insert(repository.clone()) {
            continue;
        }
        repositories.push(repository);
        if repositories.len() == limit {
            break;
        }
    }
    Ok(repositories)
}

fn event_status(connection: &Connection, event_id: &EventId) -> Result<DatabaseEventStatus> {
    let deleted = connection.query_row(
        "SELECT EXISTS(SELECT 1 FROM deleted_ids WHERE event_id = ?1)",
        [event_id.as_bytes().as_slice()],
        |row| row.get::<_, bool>(0),
    )?;
    if deleted {
        return Ok(DatabaseEventStatus::Deleted);
    }
    let saved = connection.query_row(
        "SELECT EXISTS(SELECT 1 FROM events WHERE id = ?1)",
        [event_id.as_bytes().as_slice()],
        |row| row.get::<_, bool>(0),
    )?;
    Ok(if saved {
        DatabaseEventStatus::Saved
    } else {
        DatabaseEventStatus::NotExistent
    })
}

fn event_by_id(connection: &Connection, event_id: &EventId) -> Result<Option<Event>> {
    connection
        .query_row(
            "
            SELECT id, pubkey, created_at, kind, content, tags, sig
            FROM events
            WHERE id = ?1
            ",
            [event_id.as_bytes().as_slice()],
            StoredEvent::read,
        )
        .optional()?
        .map(StoredEvent::into_event)
        .transpose()
}

fn save_event(connection: &mut Connection, event: &Event) -> Result<SaveEventStatus> {
    if event.kind.is_ephemeral() {
        return Ok(SaveEventStatus::Rejected(RejectedReason::Ephemeral));
    }

    let transaction = connection.transaction()?;
    if event_exists(&transaction, &event.id)? {
        return Ok(SaveEventStatus::Rejected(RejectedReason::Duplicate));
    }
    if event_deleted(&transaction, &event.id)? || coordinate_deleted(&transaction, event)? {
        return Ok(SaveEventStatus::Rejected(RejectedReason::Deleted));
    }
    if replacement_rejected(&transaction, event)? {
        return Ok(SaveEventStatus::Rejected(RejectedReason::Replaced));
    }
    if event.kind == Kind::EventDeletion && !apply_deletion(&transaction, event)? {
        return Ok(SaveEventStatus::Rejected(RejectedReason::InvalidDelete));
    }

    transaction.execute(
        "
        INSERT INTO events(id, pubkey, created_at, kind, content, tags, sig)
        VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
        ",
        params![
            event.id.as_bytes().as_slice(),
            event.pubkey.as_bytes().as_slice(),
            integer(event.created_at.as_secs())?,
            i64::from(event.kind.as_u16()),
            &event.content,
            serde_json::to_string(&event.tags)?,
            event.sig.as_bytes().as_slice(),
        ],
    )?;
    for tag in event.tags.iter() {
        let (Some(name), Some(value)) = (tag.single_letter_tag(), tag.content()) else {
            continue;
        };
        transaction.execute(
            "
            INSERT OR IGNORE INTO event_tags(event_id, tag_name, tag_value)
            VALUES (?1, ?2, ?3)
            ",
            params![event.id.as_bytes().as_slice(), name.to_string(), value],
        )?;
    }
    transaction.commit()?;
    Ok(SaveEventStatus::Success)
}

fn event_exists(transaction: &Transaction<'_>, event_id: &EventId) -> Result<bool> {
    Ok(transaction.query_row(
        "SELECT EXISTS(SELECT 1 FROM events WHERE id = ?1)",
        [event_id.as_bytes().as_slice()],
        |row| row.get(0),
    )?)
}

fn event_deleted(transaction: &Transaction<'_>, event_id: &EventId) -> Result<bool> {
    Ok(transaction.query_row(
        "SELECT EXISTS(SELECT 1 FROM deleted_ids WHERE event_id = ?1)",
        [event_id.as_bytes().as_slice()],
        |row| row.get(0),
    )?)
}

fn coordinate_deleted(transaction: &Transaction<'_>, event: &Event) -> Result<bool> {
    let Some(coordinate) = event.coordinate() else {
        return Ok(false);
    };
    let deleted_at = transaction
        .query_row(
            "
            SELECT deleted_at
            FROM deleted_coordinates
            WHERE pubkey = ?1 AND kind = ?2 AND identifier = ?3
            ",
            params![
                coordinate.public_key.as_bytes().as_slice(),
                i64::from(coordinate.kind.as_u16()),
                coordinate.identifier,
            ],
            |row| row.get::<_, i64>(0),
        )
        .optional()?;
    Ok(deleted_at.is_some_and(|stored_deleted_at| {
        u64::try_from(stored_deleted_at).map_or(true, |value| event.created_at.as_secs() <= value)
    }))
}

fn replacement_rejected(transaction: &Transaction<'_>, event: &Event) -> Result<bool> {
    let existing = if event.kind.is_replaceable() {
        transaction
            .query_row(
                "
                SELECT id, created_at
                FROM events
                WHERE pubkey = ?1 AND kind = ?2
                LIMIT 1
                ",
                params![
                    event.pubkey.as_bytes().as_slice(),
                    i64::from(event.kind.as_u16()),
                ],
                |row| Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, i64>(1)?)),
            )
            .optional()?
    } else if event.kind.is_addressable() {
        let Some(identifier) = event.tags.identifier() else {
            return Ok(false);
        };
        transaction
            .query_row(
                "
                SELECT event.id, event.created_at
                FROM events AS event
                JOIN event_tags AS tag ON tag.event_id = event.id
                WHERE event.pubkey = ?1
                  AND event.kind = ?2
                  AND tag.tag_name = 'd'
                  AND tag.tag_value = ?3
                LIMIT 1
                ",
                params![
                    event.pubkey.as_bytes().as_slice(),
                    i64::from(event.kind.as_u16()),
                    identifier,
                ],
                |row| Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, i64>(1)?)),
            )
            .optional()?
    } else {
        return Ok(false);
    };

    let Some((stored_id, stored_at)) = existing else {
        return Ok(false);
    };
    let incoming_at = integer(event.created_at.as_secs())?;
    match stored_at.cmp(&incoming_at) {
        Ordering::Greater => Ok(true),
        Ordering::Equal if stored_id.as_slice() < event.id.as_bytes().as_slice() => Ok(true),
        Ordering::Equal | Ordering::Less => {
            transaction.execute("DELETE FROM events WHERE id = ?1", [stored_id])?;
            Ok(false)
        }
    }
}

fn apply_deletion(transaction: &Transaction<'_>, event: &Event) -> Result<bool> {
    for event_id in event.tags.event_ids() {
        let owner = transaction
            .query_row(
                "SELECT pubkey FROM events WHERE id = ?1",
                [event_id.as_bytes().as_slice()],
                |row| row.get::<_, Vec<u8>>(0),
            )
            .optional()?;
        if owner.is_some_and(|saved_owner| saved_owner.as_slice() != event.pubkey.as_bytes()) {
            return Ok(false);
        }
        transaction.execute(
            "INSERT OR IGNORE INTO deleted_ids(event_id) VALUES (?1)",
            [event_id.as_bytes().as_slice()],
        )?;
        transaction.execute(
            "DELETE FROM events WHERE id = ?1 AND pubkey = ?2",
            params![
                event_id.as_bytes().as_slice(),
                event.pubkey.as_bytes().as_slice(),
            ],
        )?;
    }

    for coordinate in event.tags.coordinates() {
        if coordinate.public_key != event.pubkey {
            return Ok(false);
        }
        transaction.execute(
            "
            INSERT INTO deleted_coordinates(pubkey, kind, identifier, deleted_at)
            VALUES (?1, ?2, ?3, ?4)
            ON CONFLICT(pubkey, kind, identifier) DO UPDATE SET
                deleted_at = max(deleted_coordinates.deleted_at, excluded.deleted_at)
            ",
            params![
                coordinate.public_key.as_bytes().as_slice(),
                i64::from(coordinate.kind.as_u16()),
                &coordinate.identifier,
                integer(event.created_at.as_secs())?,
            ],
        )?;
        if coordinate.kind.is_replaceable() {
            transaction.execute(
                "
                DELETE FROM events
                WHERE pubkey = ?1 AND kind = ?2 AND created_at <= ?3
                ",
                params![
                    coordinate.public_key.as_bytes().as_slice(),
                    i64::from(coordinate.kind.as_u16()),
                    integer(event.created_at.as_secs())?,
                ],
            )?;
        } else {
            transaction.execute(
                "
                DELETE FROM events
                WHERE id IN (
                    SELECT stored.id
                    FROM events AS stored
                    JOIN event_tags AS tag ON tag.event_id = stored.id
                    WHERE stored.pubkey = ?1
                      AND stored.kind = ?2
                      AND stored.created_at <= ?3
                      AND tag.tag_name = 'd'
                      AND tag.tag_value = ?4
                )
                ",
                params![
                    coordinate.public_key.as_bytes().as_slice(),
                    i64::from(coordinate.kind.as_u16()),
                    integer(event.created_at.as_secs())?,
                    &coordinate.identifier,
                ],
            )?;
        }
    }
    Ok(true)
}

fn query_events(
    connection: &Connection,
    filter: &Filter,
    max_bytes: usize,
) -> Result<BTreeSet<Event>> {
    let query = EventQuery::new(filter)?;
    if query.limit == 0 {
        return Ok(BTreeSet::new());
    }
    let mut statement = connection.prepare(&query.sql)?;
    let mut rows = statement.query(params_from_iter(query.parameters))?;
    let mut events = BTreeSet::new();
    let mut bytes = 0_usize;

    while let Some(row) = rows.next()? {
        let event = StoredEvent::read(row)?.into_event()?;
        if event.is_expired() {
            continue;
        }
        bytes = bytes
            .checked_add(event.as_json().len())
            .context("history byte count overflow")?;
        if bytes > max_bytes {
            anyhow::bail!("history byte limit exceeded");
        }
        events.insert(event);
        if events.len() == query.limit {
            break;
        }
    }

    Ok(events)
}

fn replace_discovered(
    connection: &mut Connection,
    signer: &[u8],
    urls: Vec<String>,
    max: usize,
) -> Result<()> {
    let transaction = connection.transaction()?;
    transaction.execute(
        "DELETE FROM vineyard_discovered_peers WHERE signer = ?1",
        [signer],
    )?;
    let stored_count = transaction.query_row(
        "SELECT count(DISTINCT url) FROM vineyard_discovered_peers",
        [],
        |row| row.get::<_, i64>(0),
    )?;
    let count = usize::try_from(stored_count).context("negative discovered peer count")?;

    for url in urls.into_iter().take(max.saturating_sub(count)) {
        transaction.execute(
            "
                INSERT OR IGNORE INTO vineyard_discovered_peers(signer, url, discovered_at)
                VALUES (?1, ?2, unixepoch())
                ",
            params![signer, url],
        )?;
    }

    transaction.commit()?;
    Ok(())
}

fn list_discovered(connection: &Connection, limit: usize) -> Result<Vec<String>> {
    let mut statement = connection
        .prepare("SELECT DISTINCT url FROM vineyard_discovered_peers ORDER BY url LIMIT ?1")?;
    let mut rows = statement.query([integer(limit)?])?;
    let mut urls = Vec::new();
    while let Some(row) = rows.next()? {
        urls.push(row.get(0)?);
    }
    Ok(urls)
}

fn forget_discovered(connection: &mut Connection, urls: Vec<String>) -> Result<()> {
    let transaction = connection.transaction()?;
    for url in urls {
        transaction.execute(
            "DELETE FROM vineyard_discovered_peers WHERE url = ?1",
            [url],
        )?;
    }
    transaction.commit()?;
    Ok(())
}

impl StoredEvent {
    fn read(row: &Row<'_>) -> rusqlite::Result<Self> {
        Ok(Self {
            id: row.get(0)?,
            pubkey: row.get(1)?,
            created_at: row.get(2)?,
            kind: row.get(3)?,
            content: row.get(4)?,
            tags: row.get(5)?,
            signature: row.get(6)?,
        })
    }

    fn into_event(self) -> Result<Event> {
        let id = EventId::from_slice(&self.id).context("invalid stored event id")?;
        let pubkey = PublicKey::from_slice(&self.pubkey).context("invalid stored public key")?;
        let created_at: Timestamp = self
            .created_at
            .try_into()
            .context("invalid event timestamp")?;
        let kind = Kind::from_u16(self.kind.try_into().context("invalid stored event kind")?);
        let tags: Tags = serde_json::from_str(&self.tags).context("invalid stored event tags")?;
        let signature =
            Signature::from_slice(&self.signature).context("invalid stored event signature")?;
        Ok(Event::new(
            id,
            pubkey,
            created_at,
            kind,
            tags,
            self.content,
            signature,
        ))
    }
}

fn configure(connection: &Connection, max_database_bytes: u64) -> Result<()> {
    connection.busy_timeout(Duration::from_secs(5))?;
    connection.pragma_update(None, "journal_mode", "WAL")?;
    let journal_mode: String =
        connection.pragma_query_value(None, "journal_mode", |row| row.get(0))?;
    if !journal_mode.eq_ignore_ascii_case("wal") {
        anyhow::bail!("journal mode is not wal");
    }

    connection.pragma_update(None, "synchronous", "FULL")?;
    connection.pragma_update(None, "foreign_keys", "ON")?;

    let stored_page_size =
        connection.query_row("PRAGMA page_size", [], |row| row.get::<_, i64>(0))?;
    let page_size = u64::try_from(stored_page_size).context("invalid database page size")?;
    let page_budget = max_database_bytes / page_size;
    if page_budget == 0 {
        anyhow::bail!("database budget is smaller than one page");
    }
    let max_pages = integer(page_budget)?;
    let page_count = connection.query_row("PRAGMA page_count", [], |row| row.get::<_, i64>(0))?;
    if page_count > max_pages {
        anyhow::bail!("database exceeds size budget");
    }
    connection.pragma_update(None, "max_page_count", max_pages)?;
    Ok(())
}

fn migrate(
    connection: &mut Connection,
    max_discovered_peers: usize,
    excluded_peers: &[String],
) -> Result<()> {
    let transaction = connection.transaction()?;
    migrate_legacy_tables(&transaction)?;
    transaction.execute_batch(SCHEMA)?;
    transaction.execute(BACKFILL_DELIVERY, [])?;
    transaction.commit()?;
    backfill_discovered(connection, max_discovered_peers, excluded_peers)?;
    Ok(())
}

fn migrate_legacy_tables(transaction: &Transaction<'_>) -> Result<()> {
    for (old, new) in [
        ("sanctum_delivery", "vineyard_delivery"),
        ("sanctum_peers", "vineyard_peers"),
        ("sanctum_checkpoints", "vineyard_checkpoints"),
        ("sanctum_retries", "vineyard_retries"),
        ("sanctum_discovered_peers", "vineyard_discovered_peers"),
    ] {
        let old_exists = table_exists(transaction, old)?;
        let new_exists = table_exists(transaction, new)?;
        if old_exists && new_exists {
            anyhow::bail!("database contains both {old} and {new}");
        }
        if old_exists {
            transaction.execute_batch(&format!("ALTER TABLE {old} RENAME TO {new}"))?;
        }
    }
    transaction.execute_batch(
        "
        DROP TRIGGER IF EXISTS sanctum_delivery_after_insert;
        DROP INDEX IF EXISTS sanctum_retries_due;
        DROP INDEX IF EXISTS sanctum_discovered_urls;
        ",
    )?;
    Ok(())
}

fn table_exists(transaction: &Transaction<'_>, name: &str) -> Result<bool> {
    Ok(transaction.query_row(
        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1)",
        [name],
        |row| row.get(0),
    )?)
}

fn backfill_discovered(
    connection: &mut Connection,
    max_discovered_peers: usize,
    excluded_peers: &[String],
) -> Result<()> {
    let excluded = excluded_peers.iter().cloned().collect::<HashSet<_>>();
    let relay_tags = {
        let mut statement = connection.prepare(DISCOVERED_RELAY_TAGS)?;
        let mut rows = statement.query([])?;
        let mut relay_tags = Vec::new();
        while let Some(row) = rows.next()? {
            relay_tags.push((
                row.get::<_, Vec<u8>>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, i64>(2)?,
            ));
        }
        relay_tags
    };

    let transaction = connection.transaction()?;
    transaction.execute("DELETE FROM vineyard_discovered_peers", [])?;
    let mut urls = HashSet::new();
    for (signer, raw_url, discovered_at) in relay_tags {
        let Some(url) = relay_url(&raw_url).map(|url| url.to_string()) else {
            continue;
        };
        if excluded.contains(&url) || (!urls.contains(&url) && urls.len() >= max_discovered_peers) {
            continue;
        }
        urls.insert(url.clone());
        transaction.execute(
            "
            INSERT OR IGNORE INTO vineyard_discovered_peers(signer, url, discovered_at)
            VALUES (?1, ?2, ?3)
            ",
            params![signer, url, discovered_at],
        )?;
    }
    transaction.commit()?;
    Ok(())
}

fn advance_delivery(transaction: &Transaction<'_>, peer: &str, sequence: i64) -> Result<()> {
    transaction.execute(
        "
        INSERT INTO vineyard_peers(peer, delivered)
        VALUES (?1, ?2)
        ON CONFLICT(peer) DO UPDATE SET
            delivered = max(vineyard_peers.delivered, excluded.delivered)
        ",
        params![peer, sequence],
    )?;
    Ok(())
}

fn integer<T>(value: T) -> Result<i64>
where
    T: TryInto<i64>,
    T::Error: Error + Send + Sync + 'static,
{
    value
        .try_into()
        .context("value exceeds sqlite integer range")
}

fn unix_seconds() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |duration| {
            i64::try_from(duration.as_secs()).unwrap_or(i64::MAX)
        })
}