vineyard

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()