kindexr/src/main.rs

use anyhow::Context;
use axum::{extract::State, routing::get, Json, Router};
use clap::Parser;
use kindexr::{blossom::BlossomClient, config, db, enrich::tmdb::Enricher, nostr, publisher, torznab, ui, wot, AppState};
use serde_json::{json, Value};
use std::{
    net::SocketAddr,
    sync::{atomic::AtomicI32, Arc},
    time::Instant,
};
use tokio::signal;
use tracing::{info, warn};

/// kindexr — Nostr-native Torznab indexer
#[derive(Parser)]
#[command(version)]
struct Cli {
    /// Path to configuration file
    #[arg(long, default_value = "/etc/kindexr/config.yaml")]
    config: String,
}

const VERSION: &str = env!("CARGO_PKG_VERSION");

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let cli = Cli::parse();
    let cfg = config::load(&cli.config).context("load config")?;

    let log_level: tracing::Level = match cfg.logging.level.as_str() {
        "debug" => tracing::Level::DEBUG,
        "warn" | "warning" => tracing::Level::WARN,
        "error" => tracing::Level::ERROR,
        _ => tracing::Level::INFO,
    };

    if cfg.logging.format == "json" {
        tracing_subscriber::fmt()
            .json()
            .with_max_level(log_level)
            .init();
    } else {
        tracing_subscriber::fmt().with_max_level(log_level).init();
    }

    info!("starting kindexr {VERSION}");

    // ── Shared HTTP client with optional proxy ──────────────────────────────
    let http_client = build_http_client(&cfg.network.proxy)?;

    // ── nostr-sdk connection options with optional SOCKS5 proxy ─────────────
    let nostr_opts = build_nostr_opts(&cfg.network.proxy);

    let pool = db::open(&cfg.database.path).await?;
    info!("database ready");

    db::sync_relays(&pool, &cfg.relays).await?;

    let relay_count = Arc::new(AtomicI32::new(0));
    let cfg = Arc::new(cfg);

    let state = AppState {
        pool: pool.clone(),
        cfg: cfg.clone(),
        version: VERSION,
        started_at: Instant::now(),
        relay_count: relay_count.clone(),
    };

    let enricher = Arc::new(Enricher::new(cfg.tmdb.api_key.clone(), http_client.clone()));

    let blossom: Option<Arc<BlossomClient>> = if cfg.blossom.enabled && !cfg.blossom.server.is_empty() {
        info!(server = %cfg.blossom.server, "blossom client enabled");
        Some(Arc::new(BlossomClient::new(&cfg.blossom.server, http_client.clone())))
    } else {
        None
    };

    let reader = nostr::reader::Reader::new(cfg.clone(), pool.clone(), relay_count, enricher.clone(), blossom.clone(), nostr_opts.clone());
    reader.start();

    wot::follows::WotBuilder::new(cfg.clone(), pool.clone(), nostr_opts.clone()).start();
    nostr::profiles::ProfileFetcher::new(cfg.clone(), pool.clone(), nostr_opts.clone()).start();

    if cfg.publisher.enabled {
        match nostr::signer::Signer::from_nsec(&cfg.publisher.identity.nsec) {
            Ok(signer) => {
                let watcher = publisher::watcher::Watcher::new(
                    cfg.clone(),
                    pool.clone(),
                    Arc::new(signer),
                    enricher,
                    blossom.clone(),
                );
                watcher.start();
                info!("publisher watcher started");
            }
            Err(e) => {
                warn!("publisher.enabled=true but nsec is invalid — watcher not started: {e}");
            }
        }
    }

    let app = Router::new()
        .route("/health", get(health_handler))
        .merge(torznab::server::router(state.clone()))
        .merge(ui::router())
        .with_state(state);

    let listener = tokio::net::TcpListener::bind(&cfg.server.listen)
        .await
        .with_context(|| format!("bind to {}", cfg.server.listen))?;
    info!("listening on {}", cfg.server.listen);

    axum::serve(listener, app)
        .with_graceful_shutdown(shutdown_signal())
        .await
        .context("http server")?;

    info!("kindexr stopped");
    Ok(())
}

fn build_http_client(proxy: &config::ProxyConfig) -> anyhow::Result<reqwest::Client> {
    let mut b = reqwest::Client::builder();
    if let Some(addr) = proxy.socks5_addr() {
        let proxy_url = if addr.starts_with("socks5://") {
            addr
        } else {
            format!("socks5://{addr}")
        };
        info!(proxy = %proxy_url, "HTTP client using SOCKS5 proxy ({})", proxy.mode);
        b = b.proxy(reqwest::Proxy::all(&proxy_url).context("invalid proxy URL")?);
    }
    Ok(b.build()?)
}

fn build_nostr_opts(proxy: &config::ProxyConfig) -> nostr_sdk::ClientOptions {
    use nostr_sdk::{client::Connection, ClientOptions};
    if let Some(addr_str) = proxy.socks5_addr() {
        match addr_str.parse::<SocketAddr>() {
            Ok(addr) => {
                info!(proxy = %addr_str, "nostr-sdk using SOCKS5 proxy ({})", proxy.mode);
                ClientOptions::new().connection(Connection::new().proxy(addr))
            }
            Err(e) => {
                warn!("invalid proxy address '{addr_str}': {e} — using direct connection");
                ClientOptions::new()
            }
        }
    } else {
        ClientOptions::new()
    }
}

async fn health_handler(State(state): State<AppState>) -> Json<Value> {
    let db_ok = db::get_stats(&state.pool).await.is_ok();
    let uptime = state.started_at.elapsed().as_secs();
    let relay_count = state.relay_count.load(std::sync::atomic::Ordering::Relaxed);

    Json(json!({
        "status": "ok",
        "version": state.version,
        "db_ok": db_ok,
        "relays_configured": state.cfg.relays.len(),
        "relays_connected": relay_count,
        "uptime_seconds": uptime,
    }))
}

async fn shutdown_signal() {
    let ctrl_c = async {
        signal::ctrl_c().await.expect("ctrl-c handler");
    };
    let terminate = async {
        signal::unix::signal(signal::unix::SignalKind::terminate())
            .expect("SIGTERM handler")
            .recv()
            .await;
    };
    tokio::select! {
        _ = ctrl_c => {},
        _ = terminate => {},
    }
    info!("received shutdown signal");
}