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::{config::Config, db};

pub struct Reader {
    cfg: Arc<Config>,
    pool: SqlitePool,
    connected: Arc<AtomicI32>,
}

impl Reader {
    pub fn new(cfg: Arc<Config>, pool: SqlitePool, connected: Arc<AtomicI32>) -> Self {
        Reader { cfg, pool, connected }
    }

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

async fn run_reader(cfg: Arc<Config>, pool: SqlitePool, connected: Arc<AtomicI32>) {
    if cfg.relays.is_empty() {
        warn!("no relays configured; reader idle");
        return;
    }

    let client = Client::default();

    for relay_url in &cfg.relays {
        match client.add_relay(relay_url.as_str()).await {
            Ok(_) => {
                // Seed relay row so sync state is tracked.
                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());

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

    let since = Timestamp::from(backfill_since);

    let filter = Filter::new()
        .kind(Kind::Custom(2003))
        .since(since);

    if let Err(e) = client.subscribe(filter, None).await {
        error!("subscribe failed: {e}");
        connected.store(0, Ordering::Relaxed);
        return;
    }

    info!("subscribed to kind 2003 events since unix={backfill_since}");

    // Handle incoming events.
    let pool_clone = pool.clone();
    if let Err(e) = client
        .handle_notifications(|notification| {
            let pool = pool_clone.clone();
            async move {
                match notification {
                    RelayPoolNotification::Event { event, .. } => {
                        handle_event(&pool, &event).await;
                    }
                    RelayPoolNotification::Message { relay_url, message } => {
                        debug!(url = %relay_url, "relay message: {:?}", message);
                    }
                    RelayPoolNotification::Shutdown => {
                        info!("relay pool shutdown");
                        return Ok(true); // exit handler
                    }
                }
                Ok(false)
            }
        })
        .await
    {
        error!("notification handler exited: {e}");
    }

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

async fn handle_event(pool: &SqlitePool, event: &nostr_sdk::Event) {
    match super::parser::parse(event) {
        Ok(rec) => {
            if let Err(e) = db::insert_torrent(pool, &rec).await {
                error!(event_id = %event.id, "db insert failed: {e}");
            } else {
                debug!(event_id = %event.id, title = %rec.title, "indexed event");
                let _ = db::update_relay_last_event(pool, "", event.created_at.as_secs() as i64).await;
            }
        }
        Err(e) => {
            debug!(event_id = %event.id, "skipping event: {e}");
        }
    }
}