kindexr/src/nostr/reader.rs

use std::{
    sync::{atomic::{AtomicI32, Ordering}, Arc},
    time::{SystemTime, UNIX_EPOCH},
};

use nostr_sdk::{prelude::*, RelayPoolNotification};
use sqlx::SqlitePool;
use tracing::{debug, error, info, warn};

use crate::{blossom::BlossomClient, config::Config, db, enrich::tmdb::Enricher};

pub struct Reader {
    cfg: Arc<Config>,
    pool: SqlitePool,
    connected: Arc<AtomicI32>,
    enricher: Arc<Enricher>,
    blossom: Option<Arc<BlossomClient>>,
    nostr_opts: nostr_sdk::ClientOptions,
}

impl Reader {
    pub fn new(
        cfg: Arc<Config>,
        pool: SqlitePool,
        connected: Arc<AtomicI32>,
        enricher: Arc<Enricher>,
        blossom: Option<Arc<BlossomClient>>,
        nostr_opts: nostr_sdk::ClientOptions,
    ) -> Self {
        Reader { cfg, pool, connected, enricher, blossom, nostr_opts }
    }

    pub fn start(self) {
        let cfg = self.cfg.clone();
        let pool = self.pool.clone();
        let connected = self.connected.clone();
        let enricher = self.enricher.clone();
        let blossom = self.blossom.clone();
        let nostr_opts = self.nostr_opts.clone();
        tokio::spawn(async move {
            run_reader(cfg, pool, connected, enricher, blossom, nostr_opts).await;
        });
    }
}

async fn run_reader(cfg: Arc<Config>, pool: SqlitePool, connected: Arc<AtomicI32>, enricher: Arc<Enricher>, blossom: Option<Arc<BlossomClient>>, nostr_opts: nostr_sdk::ClientOptions) {
    if cfg.relays.is_empty() {
        warn!("no relays configured; reader idle");
        return;
    }

    let client = Client::builder().opts(nostr_opts).build();
    for relay_url in &cfg.relays {
        match client.add_relay(relay_url.as_str()).await {
            Ok(_) => { let _ = db::upsert_relay(&pool, relay_url).await; }
            Err(e) => warn!(url = relay_url, "failed to add relay: {e}"),
        }
    }
    client.connect().await;
    connected.store(cfg.relays.len() as i32, Ordering::Relaxed);
    info!("relay pool connected ({} relays)", cfg.relays.len());

    let backfill_since = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
        .saturating_sub(cfg.backfill_days as u64 * 86400);

    // Load per-relay last-seen timestamps so we only request events we haven't
    // indexed yet. Fresh relays fall back to the full backfill window.
    let relay_timestamps = db::get_all_relay_timestamps(&pool).await;

    // Subscribe to kind 2003 (torrents) per relay with its own since value.
    let mut any_subscribed = false;
    for relay_url in &cfg.relays {
        let relay_since = relay_timestamps
            .get(relay_url.as_str())
            .copied()
            .map(|ts| (ts as u64).max(backfill_since))
            .unwrap_or(backfill_since);
        let since_ts = Timestamp::from(relay_since);

        match client.subscribe_to(
            [relay_url.as_str()],
            Filter::new().kind(Kind::Custom(2003)).since(since_ts),
            None,
        ).await {
            Ok(_) => {
                debug!(url = relay_url, since = relay_since, "subscribed kind 2003");
                any_subscribed = true;
            }
            Err(e) => warn!(url = relay_url, "subscribe(2003) failed: {e}"),
        }
    }
    if !any_subscribed {
        error!("all relay subscriptions failed");
        connected.store(0, Ordering::Relaxed);
        return;
    }
    info!("subscribed to kind 2003 on {} relays (per-relay since)", cfg.relays.len());

    // Subscribe to kind 30004 (curation sets) from configured pubkeys
    let global_since = Timestamp::from(backfill_since);
    let curation_pks: Vec<PublicKey> = cfg.curation.curation_sets.iter()
        .filter_map(|s| s.parse().ok())
        .collect();
    if !curation_pks.is_empty() {
        if let Err(e) = client.subscribe(
            Filter::new().kind(Kind::Custom(30004)).authors(curation_pks).since(global_since),
            None,
        ).await {
            warn!("subscribe(30004) failed: {e}");
        } else {
            info!("subscribed to kind 30004 ({} curation pubkeys)", cfg.curation.curation_sets.len());
        }
    }

    // Subscribe to kind 1984 (reports)
    if let Err(e) = client.subscribe(Filter::new().kind(Kind::Custom(1984)).since(global_since), None).await {
        warn!("subscribe(1984) failed: {e}");
    }

    let pool_clone = pool.clone();
    if let Err(e) = client
        .handle_notifications(|notification| {
            let pool = pool_clone.clone();
            let cfg = cfg.clone();
            let enricher = enricher.clone();
            let blossom = blossom.clone();
            async move {
                match notification {
                    RelayPoolNotification::Event { relay_url, event, .. } => {
                        match event.kind.as_u16() {
                            2003 => handle_torrent(&pool, &cfg, &enricher, blossom.as_deref(), relay_url.as_str(), &event).await,
                            30004 => handle_curation_set(&pool, &event).await,
                            1984 => handle_report(&pool, &cfg, &event).await,
                            _ => {}
                        }
                    }
                    RelayPoolNotification::Message { relay_url, message } => {
                        debug!(url = %relay_url, "relay message: {:?}", message);
                    }
                    RelayPoolNotification::Shutdown => {
                        info!("relay pool shutdown");
                        return Ok(true);
                    }
                }
                Ok(false)
            }
        })
        .await
    {
        error!("notification handler exited: {e}");
    }

    connected.store(0, Ordering::Relaxed);
}

async fn handle_torrent(pool: &SqlitePool, cfg: &Config, enricher: &Enricher, blossom: Option<&BlossomClient>, relay_url: &str, event: &nostr_sdk::Event) {
    match super::parser::parse(event) {
        Ok(rec) => {
            // WoT / block / mute check
            if cfg.curation.wot_only
                || !cfg.curation.blocklist.is_empty()
                || !cfg.curation.allowlist.is_empty()
            {
                let allowed = db::is_publisher_allowed(pool, &rec.pubkey, cfg.curation.wot_only)
                    .await
                    .unwrap_or(!cfg.curation.wot_only);
                if !allowed {
                    debug!(event_id = %event.id, pubkey = %rec.pubkey, "dropped: publisher not allowed");
                    return;
                }
            } else {
                // Even without wot_only, respect explicit blocks/mutes
                if let Ok(false) = db::is_publisher_allowed(pool, &rec.pubkey, false).await {
                    debug!(event_id = %event.id, pubkey = %rec.pubkey, "dropped: blocked/muted publisher");
                    return;
                }
            }

            // Category exclusion
            if let Some(cat) = rec.newznab_cat {
                let excluded = cfg.curation.exclude_categories.iter().any(|&excl| {
                    if excl % 1000 == 0 { cat >= excl && cat < excl + 1000 } else { cat == excl }
                });
                if excluded {
                    debug!(event_id = %event.id, cat, "dropped: excluded category");
                    return;
                }
            }

            match db::insert_torrent(pool, &rec).await {
                Err(e) => error!(event_id = %event.id, "db insert failed: {e}"),
                Ok(()) => {
                    debug!(event_id = %event.id, title = %rec.title, "indexed torrent");
                    let _ = db::update_relay_last_event(pool, relay_url, event.created_at.as_secs() as i64).await;

                    if enricher.enabled() && rec.tmdb_id.is_empty() {
                        let enricher = enricher.clone();
                        let pool = pool.clone();
                        let event_id = rec.event_id.clone();
                        let title = rec.title.clone();
                        let newznab_cat = rec.newznab_cat;
                        tokio::spawn(async move {
                            let parsed = crate::enrich::parser::parse(&title);
                            enricher.enrich(&pool, &event_id, &parsed.clean, parsed.year, newznab_cat).await;
                        });
                    }

                    // Blossom: if the event included a `url` tag pointing to the .torrent file
                    // and Blossom is enabled, fetch and cache it for /torrent/<info_hash>.torrent
                    if blossom.is_some() && !rec.blossom_url.is_empty() {
                        let pool = pool.clone();
                        let info_hash = rec.info_hash.clone();
                        let url = rec.blossom_url.clone();
                        tokio::spawn(async move {
                            let client = reqwest::Client::new();
                            match client.get(&url).send().await {
                                Ok(resp) if resp.status().is_success() => {
                                    match resp.bytes().await {
                                        Ok(bytes) => {
                                            let _ = db::store_torrent_blob(&pool, &info_hash, &bytes, &url).await;
                                            debug!(info_hash = %info_hash, "blossom blob cached");
                                        }
                                        Err(e) => debug!(url = %url, "blossom fetch body error: {e}"),
                                    }
                                }
                                Ok(resp) => debug!(url = %url, "blossom fetch HTTP {}", resp.status()),
                                Err(e) => debug!(url = %url, "blossom fetch failed: {e}"),
                            }
                        });
                    }
                }
            }
        }
        Err(e) => debug!(event_id = %event.id, "skipping event: {e}"),
    }
}

async fn handle_curation_set(pool: &SqlitePool, event: &nostr_sdk::Event) {
    let set_event_id = event.id.to_hex();
    let set_pubkey = event.pubkey.to_hex();
    let now = event.created_at.as_secs() as i64;

    // Extract 'e' tags referencing kind-2003 events
    let mut count = 0usize;
    for tag in event.tags.iter() {
        let s = tag.as_slice();
        if s.len() >= 2 && s[0] == "e" {
            let item_event_id = &s[1];
            if let Err(e) = db::insert_curation_item(pool, &set_event_id, &set_pubkey, item_event_id, now).await {
                debug!("curation_item insert failed: {e}");
            } else {
                count += 1;
            }
        }
    }
    if count > 0 {
        debug!(set_event_id = %set_event_id, "curation set: {count} items indexed");
    }
}

async fn handle_report(pool: &SqlitePool, cfg: &Config, event: &nostr_sdk::Event) {
    let report_event_id = event.id.to_hex();
    let reporter = event.pubkey.to_hex();
    let created_at = event.created_at.as_secs() as i64;
    let reason = if event.content.is_empty() { None } else { Some(event.content.as_str()) };

    for tag in event.tags.iter() {
        let s = tag.as_slice();
        if s.len() < 2 {
            continue;
        }
        match s[0].as_str() {
            // NIP-56: `p` tag — report against a pubkey directly.
            "p" => {
                let reported = &s[1];
                if let Err(e) = db::insert_report(
                    pool,
                    &report_event_id,
                    reported,
                    &reporter,
                    reason,
                    created_at,
                    cfg.curation.auto_block_threshold,
                ).await {
                    debug!("report insert failed: {e}");
                } else {
                    debug!(reported = %reported, "report indexed (p-tag)");
                }
            }
            // NIP-56: `e` tag — report against a specific event.
            // Look up the event's publisher and penalise them.
            "e" => {
                let event_id_ref = &s[1];
                match db::get_event_pubkey(pool, event_id_ref).await {
                    Ok(Some(pubkey)) => {
                        if let Err(e) = db::insert_report(
                            pool,
                            &report_event_id,
                            &pubkey,
                            &reporter,
                            reason,
                            created_at,
                            cfg.curation.auto_block_threshold,
                        ).await {
                            debug!("report insert failed: {e}");
                        } else {
                            debug!(event_id = %event_id_ref, pubkey = %pubkey, "report indexed (e-tag)");
                        }
                    }
                    Ok(None) => {
                        // Event not in our DB — ignore.
                    }
                    Err(e) => debug!("get_event_pubkey failed: {e}"),
                }
            }
            _ => {}
        }
    }
}