phase 3: WoT graph, curation sets, reports, publisher CLI

- Migration 003: add wot_level/muted/report_count to publishers;
  create curation_items and reports tables
- WoT builder (wot/follows.rs): fetches operator kind-3 contact list
  to build depth-1 and depth-2 trust graph; ingests kind-10000 mute
  list; runs at startup and refreshes every 24h
- Trust scoring (wot/trust.rs): recomputes trust from wot_level and
  report_count after each WoT build
- Reader: subscribes to kind-30004 (curation sets) from configured
  pubkeys; subscribes to kind-1984 (reports) from anyone; WoT/block/
  mute check gates kind-2003 ingest when wot_only=true; auto-block
  publishers that hit report threshold
- Search: curated events (referenced by kind-30004) sort first in
  both FTS and full-scan paths
- Publisher CLI: list, info, block, unblock, trust, mute subcommands
- Fix .gitignore: /bin/ not bin/ (was shadowing src/bin/)

.gitignore

@@ -1,2 +1,3 @@
-bin/
+/bin/
+target/
 *.db

src/bin/kindexr-cli.rs

@@ -0,0 +1,150 @@
+use clap::{Parser, Subcommand};
+use kindexr::{config, db};
+use rand::Rng;
+
+/// kindexr-cli — admin CLI for kindexr
+#[derive(Parser)]
+#[command(version)]
+struct Cli {
+    /// Path to configuration file
+    #[arg(long, default_value = "/etc/kindexr/config.yaml")]
+    config: String,
+
+    #[command(subcommand)]
+    command: Command,
+}
+
+#[derive(Subcommand)]
+enum Command {
+    /// Manage API keys
+    Apikey {
+        #[command(subcommand)]
+        cmd: ApikeyCmd,
+    },
+    /// Manage publishers (trust, blocks, WoT)
+    Publisher {
+        #[command(subcommand)]
+        cmd: PublisherCmd,
+    },
+}
+
+#[derive(Subcommand)]
+enum ApikeyCmd {
+    /// Create a new API key
+    Create {
+        /// Key label (e.g. sonarr, radarr)
+        #[arg(long)]
+        label: String,
+    },
+}
+
+#[derive(Subcommand)]
+enum PublisherCmd {
+    /// List known publishers
+    List {
+        #[arg(long, default_value = "50")]
+        limit: i64,
+        #[arg(long, default_value = "0")]
+        offset: i64,
+    },
+    /// Show details for a specific publisher
+    Info {
+        pubkey: String,
+    },
+    /// Block a publisher (their events will be dropped at ingest)
+    Block {
+        pubkey: String,
+    },
+    /// Unblock a publisher
+    Unblock {
+        pubkey: String,
+    },
+    /// Manually set a publisher's trust score (0.0–1.0)
+    Trust {
+        pubkey: String,
+        #[arg(long)]
+        score: f64,
+    },
+    /// Mute a publisher
+    Mute {
+        pubkey: String,
+    },
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let cli = Cli::parse();
+    let cfg = config::load(&cli.config)?;
+    let pool = db::open(&cfg.database.path).await?;
+
+    match cli.command {
+        Command::Apikey { cmd } => match cmd {
+            ApikeyCmd::Create { label } => {
+                let key = generate_key();
+                db::create_api_key(&pool, &key, &label).await?;
+                println!("{key}");
+            }
+        },
+        Command::Publisher { cmd } => match cmd {
+            PublisherCmd::List { limit, offset } => {
+                let rows = db::list_publishers(&pool, limit, offset).await?;
+                if rows.is_empty() {
+                    println!("no publishers");
+                    return Ok(());
+                }
+                println!("{:<66}  {:>5}  {:>3}  {:>6}  {:>5}  {:>5}",
+                    "pubkey", "trust", "wot", "events", "block", "muted");
+                println!("{}", "-".repeat(100));
+                for r in rows {
+                    println!("{:<66}  {:>5.2}  {:>3}  {:>6}  {:>5}  {:>5}",
+                        r.pubkey,
+                        r.trust,
+                        r.wot_level.map_or("-".into(), |l| l.to_string()),
+                        r.torrents_n,
+                        if r.blocked { "yes" } else { "no" },
+                        if r.muted { "yes" } else { "no" },
+                    );
+                }
+            }
+            PublisherCmd::Info { pubkey } => {
+                match db::get_publisher(&pool, &pubkey).await? {
+                    None => println!("unknown publisher"),
+                    Some(r) => {
+                        println!("pubkey:       {}", r.pubkey);
+                        println!("name:         {}", r.name.as_deref().unwrap_or("—"));
+                        println!("trust:        {:.3}", r.trust);
+                        println!("wot_level:    {}", r.wot_level.map_or("—".into(), |l| l.to_string()));
+                        println!("blocked:      {}", r.blocked);
+                        println!("muted:        {}", r.muted);
+                        println!("report_count: {}", r.report_count);
+                        println!("torrents:     {}", r.torrents_n);
+                    }
+                }
+            }
+            PublisherCmd::Block { pubkey } => {
+                db::block_publisher(&pool, &pubkey).await?;
+                println!("blocked {pubkey}");
+            }
+            PublisherCmd::Unblock { pubkey } => {
+                db::unblock_publisher(&pool, &pubkey).await?;
+                println!("unblocked {pubkey}");
+            }
+            PublisherCmd::Trust { pubkey, score } => {
+                let score = score.clamp(0.0, 1.0);
+                db::set_publisher_trust(&pool, &pubkey, score).await?;
+                println!("set trust={score:.3} for {pubkey}");
+            }
+            PublisherCmd::Mute { pubkey } => {
+                db::mute_publisher(&pool, &pubkey).await?;
+                println!("muted {pubkey}");
+            }
+        },
+    }
+
+    Ok(())
+}
+
+fn generate_key() -> String {
+    let bytes: [u8; 32] = rand::thread_rng().gen();
+    hex::encode(bytes)
+}

src/config.rs

@@ -44,6 +44,7 @@ pub struct CurationConfig {
     pub blocklist: Vec<String>,
     pub curation_sets: Vec<String>,
     pub exclude_categories: Vec<i32>,
+    pub auto_block_threshold: i32,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -99,6 +100,7 @@ impl Default for Config {
                 exclude_categories: vec![
                     6000, 6010, 6020, 6030, 6040, 6050, 6060, 6070, 6080, 6090,
                 ],
+                auto_block_threshold: 5,
             },
             tmdb: TmdbConfig {
                 enabled: true,

src/db/migrations/003_wot.sql

@@ -0,0 +1,26 @@
+-- WoT trust level: 0=operator, 1=direct follow, 2=follow-of-follow, NULL=unknown
+ALTER TABLE publishers ADD COLUMN wot_level INTEGER;
+ALTER TABLE publishers ADD COLUMN muted INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE publishers ADD COLUMN report_count INTEGER NOT NULL DEFAULT 0;
+
+-- Events referenced by kind-30004 curation sets
+CREATE TABLE IF NOT EXISTS curation_items (
+    set_event_id  TEXT NOT NULL,
+    set_pubkey    TEXT NOT NULL,
+    item_event_id TEXT NOT NULL,
+    added_at      INTEGER NOT NULL,
+    PRIMARY KEY (set_event_id, item_event_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_curation_item ON curation_items(item_event_id);
+
+-- NIP-56 reports
+CREATE TABLE IF NOT EXISTS reports (
+    report_event_id TEXT PRIMARY KEY,
+    reported_pubkey TEXT NOT NULL,
+    reporter_pubkey TEXT NOT NULL,
+    reason          TEXT,
+    created_at      INTEGER NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_reports_reported ON reports(reported_pubkey);

src/db/queries.rs

@@ -173,7 +173,11 @@ pub async fn search(pool: &SqlitePool, p: &SearchParams) -> anyhow::Result<Vec<T
             sql.push_str(&cat_where);
             sql.push(')');
         }
-        sql.push_str(" ORDER BY torrents_fts.rank LIMIT ? OFFSET ?");
+        // Curated events surface first, then by FTS rank
+        sql.push_str(
+            " ORDER BY EXISTS(SELECT 1 FROM curation_items WHERE item_event_id = t.event_id) DESC,\
+             torrents_fts.rank LIMIT ? OFFSET ?",
+        );
 
         let mut q = sqlx::query_as(&sql).bind(&p.query);
         for cat in &p.cats {
@@ -222,7 +226,11 @@ pub async fn search(pool: &SqlitePool, p: &SearchParams) -> anyhow::Result<Vec<T
             sql.push_str(" WHERE ");
             sql.push_str(&conditions.join(" AND "));
         }
-        sql.push_str(" ORDER BY created_at DESC LIMIT ? OFFSET ?");
+        // Curated events surface first, then by recency
+        sql.push_str(
+            " ORDER BY EXISTS(SELECT 1 FROM curation_items WHERE item_event_id = event_id) DESC,\
+             created_at DESC LIMIT ? OFFSET ?",
+        );
 
         let mut q = sqlx::query_as::<_, TorrentRow>(&sql);
 
@@ -384,6 +392,216 @@ pub async fn update_relay_sync(pool: &SqlitePool, url: &str, ts: i64) -> anyhow:
     Ok(())
 }
 
+// ─── Publishers / WoT ─────────────────────────────────────────────────────────
+
+pub struct PublisherRow {
+    pub pubkey: String,
+    pub name: Option<String>,
+    pub trust: f64,
+    pub blocked: bool,
+    pub muted: bool,
+    pub wot_level: Option<i32>,
+    pub report_count: i32,
+    pub torrents_n: i32,
+    pub first_seen: Option<i64>,
+    pub last_seen: Option<i64>,
+}
+
+impl sqlx::FromRow<'_, sqlx::sqlite::SqliteRow> for PublisherRow {
+    fn from_row(row: &sqlx::sqlite::SqliteRow) -> Result<Self, sqlx::Error> {
+        use sqlx::Row;
+        Ok(PublisherRow {
+            pubkey: row.try_get("pubkey")?,
+            name: row.try_get("name")?,
+            trust: row.try_get("trust")?,
+            blocked: row.try_get::<i64, _>("blocked")? != 0,
+            muted: row.try_get::<i64, _>("muted")? != 0,
+            wot_level: row.try_get("wot_level")?,
+            report_count: row.try_get("report_count")?,
+            torrents_n: row.try_get("torrents_n")?,
+            first_seen: row.try_get("first_seen")?,
+            last_seen: row.try_get("last_seen")?,
+        })
+    }
+}
+
+/// Insert or update a publisher's WoT level. Only updates if the new level is lower (closer to operator).
+pub async fn upsert_publisher_wot(pool: &SqlitePool, pubkey: &str, level: i32) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT INTO publishers (pubkey, wot_level, trust)
+         VALUES (?, ?, 0.0)
+         ON CONFLICT(pubkey) DO UPDATE SET
+           wot_level = CASE
+             WHEN excluded.wot_level < COALESCE(wot_level, 999) THEN excluded.wot_level
+             ELSE wot_level
+           END",
+    )
+    .bind(pubkey)
+    .bind(level)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn mute_publisher(pool: &SqlitePool, pubkey: &str) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT INTO publishers (pubkey, muted) VALUES (?, 1)
+         ON CONFLICT(pubkey) DO UPDATE SET muted = 1",
+    )
+    .bind(pubkey)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn block_publisher(pool: &SqlitePool, pubkey: &str) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT INTO publishers (pubkey, blocked) VALUES (?, 1)
+         ON CONFLICT(pubkey) DO UPDATE SET blocked = 1",
+    )
+    .bind(pubkey)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn unblock_publisher(pool: &SqlitePool, pubkey: &str) -> anyhow::Result<()> {
+    sqlx::query("UPDATE publishers SET blocked = 0 WHERE pubkey = ?")
+        .bind(pubkey)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+pub async fn set_publisher_trust(pool: &SqlitePool, pubkey: &str, trust: f64) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT INTO publishers (pubkey, trust) VALUES (?, ?)
+         ON CONFLICT(pubkey) DO UPDATE SET trust = excluded.trust",
+    )
+    .bind(pubkey)
+    .bind(trust)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Returns true if the publisher is allowed to have their events indexed.
+/// Blocked and muted publishers are always rejected.
+/// If wot_only=true, only publishers with a known wot_level are accepted.
+pub async fn is_publisher_allowed(pool: &SqlitePool, pubkey: &str, wot_only: bool) -> anyhow::Result<bool> {
+    let row: Option<(i64, i64, Option<i64>)> = sqlx::query_as(
+        "SELECT blocked, muted, wot_level FROM publishers WHERE pubkey = ?",
+    )
+    .bind(pubkey)
+    .fetch_optional(pool)
+    .await?;
+
+    match row {
+        None => Ok(!wot_only), // unknown publisher: allowed unless wot_only
+        Some((blocked, muted, wot_level)) => {
+            if blocked != 0 || muted != 0 {
+                return Ok(false);
+            }
+            if wot_only && wot_level.is_none() {
+                return Ok(false);
+            }
+            Ok(true)
+        }
+    }
+}
+
+pub async fn list_publishers(pool: &SqlitePool, limit: i64, offset: i64) -> anyhow::Result<Vec<PublisherRow>> {
+    sqlx::query_as(
+        "SELECT pubkey, name, trust, blocked, muted, wot_level, report_count, torrents_n, first_seen, last_seen
+         FROM publishers ORDER BY last_seen DESC LIMIT ? OFFSET ?",
+    )
+    .bind(limit)
+    .bind(offset)
+    .fetch_all(pool)
+    .await
+    .context("list publishers")
+}
+
+pub async fn get_publisher(pool: &SqlitePool, pubkey: &str) -> anyhow::Result<Option<PublisherRow>> {
+    sqlx::query_as(
+        "SELECT pubkey, name, trust, blocked, muted, wot_level, report_count, torrents_n, first_seen, last_seen
+         FROM publishers WHERE pubkey = ?",
+    )
+    .bind(pubkey)
+    .fetch_optional(pool)
+    .await
+    .context("get publisher")
+}
+
+// ─── Curation ─────────────────────────────────────────────────────────────────
+
+pub async fn insert_curation_item(
+    pool: &SqlitePool,
+    set_event_id: &str,
+    set_pubkey: &str,
+    item_event_id: &str,
+    now: i64,
+) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT OR IGNORE INTO curation_items (set_event_id, set_pubkey, item_event_id, added_at)
+         VALUES (?,?,?,?)",
+    )
+    .bind(set_event_id)
+    .bind(set_pubkey)
+    .bind(item_event_id)
+    .bind(now)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+// ─── Reports ──────────────────────────────────────────────────────────────────
+
+pub async fn insert_report(
+    pool: &SqlitePool,
+    report_event_id: &str,
+    reported_pubkey: &str,
+    reporter_pubkey: &str,
+    reason: Option<&str>,
+    created_at: i64,
+    auto_block_threshold: i32,
+) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT OR IGNORE INTO reports (report_event_id, reported_pubkey, reporter_pubkey, reason, created_at)
+         VALUES (?,?,?,?,?)",
+    )
+    .bind(report_event_id)
+    .bind(reported_pubkey)
+    .bind(reporter_pubkey)
+    .bind(reason)
+    .bind(created_at)
+    .execute(pool)
+    .await?;
+
+    // Update report_count
+    let count: (i64,) = sqlx::query_as(
+        "SELECT COUNT(*) FROM reports WHERE reported_pubkey = ?",
+    )
+    .bind(reported_pubkey)
+    .fetch_one(pool)
+    .await?;
+
+    sqlx::query(
+        "INSERT INTO publishers (pubkey, report_count, blocked) VALUES (?, ?, ?)
+         ON CONFLICT(pubkey) DO UPDATE SET
+           report_count = excluded.report_count,
+           blocked = CASE WHEN excluded.report_count >= ? THEN 1 ELSE blocked END",
+    )
+    .bind(reported_pubkey)
+    .bind(count.0)
+    .bind(if count.0 >= auto_block_threshold as i64 { 1i64 } else { 0i64 })
+    .bind(auto_block_threshold)
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}
+
 // ─── Helpers ──────────────────────────────────────────────────────────────────
 
 fn null_str(s: &str) -> Option<&str> {

src/lib.rs

@@ -3,6 +3,7 @@ pub mod db;
 pub mod enrich;
 pub mod nostr;
 pub mod torznab;
+pub mod wot;
 
 use std::{
     sync::{atomic::AtomicI32, Arc},

src/main.rs

@@ -1,7 +1,7 @@
 use anyhow::Context;
 use axum::{extract::State, routing::get, Json, Router};
 use clap::Parser;
-use kindexr::{config, db, enrich::tmdb::Enricher, nostr, torznab, AppState};
+use kindexr::{config, db, enrich::tmdb::Enricher, nostr, torznab, wot, AppState};
 use serde_json::{json, Value};
 use std::{
     sync::{atomic::AtomicI32, Arc},
@@ -62,6 +62,8 @@ async fn main() -> anyhow::Result<()> {
     let reader = nostr::reader::Reader::new(cfg.clone(), pool.clone(), relay_count, enricher);
     reader.start();
 
+    wot::follows::WotBuilder::new(cfg.clone(), pool.clone()).start();
+
     let app = Router::new()
         .route("/health", get(health_handler))
         .merge(torznab::server::router(state.clone()))

src/nostr/reader.rs

@@ -3,10 +3,7 @@ use std::{
     time::{SystemTime, UNIX_EPOCH},
 };
 
-use nostr_sdk::{
-    prelude::*,
-    RelayPoolNotification,
-};
+use nostr_sdk::{prelude::*, RelayPoolNotification};
 use sqlx::SqlitePool;
 use tracing::{debug, error, info, warn};
 
@@ -42,16 +39,12 @@ async fn run_reader(cfg: Arc<Config>, pool: SqlitePool, connected: Arc<AtomicI32
     }
 
     let client = Client::default();
-
     for relay_url in &cfg.relays {
         match client.add_relay(relay_url.as_str()).await {
-            Ok(_) => {
-                let _ = db::upsert_relay(&pool, relay_url).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());
@@ -61,17 +54,35 @@ async fn run_reader(cfg: Arc<Config>, pool: SqlitePool, connected: Arc<AtomicI32
         .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}");
+    // Subscribe to kind 2003 (torrents)
+    if let Err(e) = client.subscribe(Filter::new().kind(Kind::Custom(2003)).since(since), None).await {
+        error!("subscribe(2003) failed: {e}");
         connected.store(0, Ordering::Relaxed);
         return;
     }
+    info!("subscribed to kind 2003 since unix={backfill_since}");
 
-    info!("subscribed to kind 2003 events since unix={backfill_since}");
+    // Subscribe to kind 30004 (curation sets) from configured pubkeys
+    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(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(since), None).await {
+        warn!("subscribe(1984) failed: {e}");
+    }
 
     let pool_clone = pool.clone();
     if let Err(e) = client
@@ -82,7 +93,12 @@ async fn run_reader(cfg: Arc<Config>, pool: SqlitePool, connected: Arc<AtomicI32
             async move {
                 match notification {
                     RelayPoolNotification::Event { event, .. } => {
-                        handle_event(&pool, &cfg, &enricher, &event).await;
+                        match event.kind.as_u16() {
+                            2003 => handle_torrent(&pool, &cfg, &enricher, &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);
@@ -103,17 +119,33 @@ async fn run_reader(cfg: Arc<Config>, pool: SqlitePool, connected: Arc<AtomicI32
     connected.store(0, Ordering::Relaxed);
 }
 
-async fn handle_event(pool: &SqlitePool, cfg: &Config, enricher: &Enricher, event: &nostr_sdk::Event) {
+async fn handle_torrent(pool: &SqlitePool, cfg: &Config, enricher: &Enricher, event: &nostr_sdk::Event) {
     match super::parser::parse(event) {
         Ok(rec) => {
-            // Drop events in excluded categories (adult content etc.)
+            // 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 excl % 1000 == 0 { cat >= excl && cat < excl + 1000 } else { cat == excl }
                 });
                 if excluded {
                     debug!(event_id = %event.id, cat, "dropped: excluded category");
@@ -124,10 +156,9 @@ async fn handle_event(pool: &SqlitePool, cfg: &Config, enricher: &Enricher, even
             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 event");
+                    debug!(event_id = %event.id, title = %rec.title, "indexed torrent");
                     let _ = db::update_relay_last_event(pool, "", event.created_at.as_secs() as i64).await;
 
-                    // Spawn TMDB enrichment when no ID was in the event.
                     if enricher.enabled() && rec.tmdb_id.is_empty() {
                         let enricher = enricher.clone();
                         let pool = pool.clone();
@@ -135,7 +166,6 @@ async fn handle_event(pool: &SqlitePool, cfg: &Config, enricher: &Enricher, even
                         let title = rec.title.clone();
                         let newznab_cat = rec.newznab_cat;
                         tokio::spawn(async move {
-                            // Re-parse title to get year for TMDB lookup.
                             let parsed = crate::enrich::parser::parse(&title);
                             enricher.enrich(&pool, &event_id, &parsed.clean, parsed.year, newznab_cat).await;
                         });
@@ -143,8 +173,56 @@ async fn handle_event(pool: &SqlitePool, cfg: &Config, enricher: &Enricher, even
                 }
             }
         }
-        Err(e) => {
-            debug!(event_id = %event.id, "skipping event: {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 && s[0] == "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");
+            }
         }
     }
 }

src/wot/follows.rs

@@ -0,0 +1,157 @@
+use std::{sync::Arc, time::Duration};
+
+use nostr_sdk::{prelude::*, RelayPoolNotification};
+use sqlx::SqlitePool;
+use tracing::{debug, error, info, warn};
+
+use crate::{config::Config, db};
+
+pub struct WotBuilder {
+    cfg: Arc<Config>,
+    pool: SqlitePool,
+}
+
+impl WotBuilder {
+    pub fn new(cfg: Arc<Config>, pool: SqlitePool) -> Self {
+        Self { cfg, pool }
+    }
+
+    pub fn start(self) {
+        tokio::spawn(async move {
+            loop {
+                if !self.cfg.curation.operator_pubkey.is_empty() {
+                    if let Err(e) = self.build().await {
+                        error!("wot build failed: {e}");
+                    }
+                }
+                tokio::time::sleep(Duration::from_secs(24 * 3600)).await;
+            }
+        });
+    }
+
+    async fn build(&self) -> anyhow::Result<()> {
+        let operator_hex = &self.cfg.curation.operator_pubkey;
+        let operator_pk: PublicKey = operator_hex.parse().map_err(|e| anyhow::anyhow!("invalid operator_pubkey: {e}"))?;
+
+        info!("wot: building graph for operator {operator_hex:.12}…");
+
+        // Fetch kind 3 (contact list) and kind 10000 (mute list) from operator
+        let events = self.fetch(
+            Filter::new()
+                .author(operator_pk)
+                .kinds(vec![Kind::ContactList, Kind::Custom(10000)]),
+            30,
+        ).await;
+
+        // --- Kind 3: contact list ---
+        let kind3 = events.iter()
+            .filter(|e| e.kind == Kind::ContactList)
+            .max_by_key(|e| e.created_at.as_secs());
+
+        let mut level1: Vec<String> = vec![];
+        if let Some(ev) = kind3 {
+            level1 = p_tags(ev);
+            info!("wot: {} direct follows", level1.len());
+
+            db::upsert_publisher_wot(&self.pool, operator_hex, 0).await?;
+            for pk in &level1 {
+                db::upsert_publisher_wot(&self.pool, pk, 1).await?;
+            }
+
+            // Depth 2: fetch kind 3 for each level-1 follow
+            if self.cfg.curation.follow_depth >= 2 && !level1.is_empty() {
+                let pks: Vec<PublicKey> = level1.iter()
+                    .filter_map(|s| s.parse().ok())
+                    .collect();
+
+                let fof_events = self.fetch(
+                    Filter::new().authors(pks).kind(Kind::ContactList),
+                    90,
+                ).await;
+
+                let mut added = 0usize;
+                for ev in &fof_events {
+                    for pk in p_tags(ev) {
+                        db::upsert_publisher_wot(&self.pool, &pk, 2).await?;
+                        added += 1;
+                    }
+                }
+                info!("wot: {added} level-2 entries processed");
+            }
+        }
+
+        // --- Kind 10000: mute list ---
+        let kind10000 = events.iter()
+            .filter(|e| e.kind.as_u16() == 10000)
+            .max_by_key(|e| e.created_at.as_secs());
+
+        if let Some(ev) = kind10000 {
+            let muted = p_tags(ev);
+            info!("wot: {} pubkeys in mute list", muted.len());
+            for pk in &muted {
+                db::mute_publisher(&self.pool, pk).await?;
+            }
+        }
+
+        // Recompute trust scores now that WoT is fresh.
+        crate::wot::trust::recompute(&self.pool).await?;
+
+        info!("wot: graph build complete");
+        Ok(())
+    }
+
+    /// Subscribes to a filter, collects events for `timeout_secs`, returns them.
+    async fn fetch(&self, filter: Filter, timeout_secs: u64) -> Vec<nostr_sdk::Event> {
+        let client = Client::default();
+        for relay in &self.cfg.relays {
+            if let Err(e) = client.add_relay(relay.as_str()).await {
+                warn!(url = relay, "wot: relay add failed: {e}");
+            }
+        }
+        client.connect().await;
+
+        if let Err(e) = client.subscribe(filter, None).await {
+            warn!("wot: subscribe failed: {e}");
+            return vec![];
+        }
+
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<nostr_sdk::Event>();
+        let tx2 = tx.clone();
+        let client_h = client.clone();
+
+        let handler = tokio::spawn(async move {
+            let _ = client_h.handle_notifications(|n| {
+                let tx = tx2.clone();
+                async move {
+                    if let RelayPoolNotification::Event { event, .. } = n {
+                        let _ = tx.send(*event);
+                    }
+                    Ok(false)
+                }
+            }).await;
+        });
+
+        let mut events = Vec::new();
+        let sleep = tokio::time::sleep(Duration::from_secs(timeout_secs));
+        tokio::pin!(sleep);
+        loop {
+            tokio::select! {
+                Some(e) = rx.recv() => events.push(e),
+                _ = &mut sleep => break,
+            }
+        }
+
+        handler.abort();
+        debug!("wot: fetched {} events in {timeout_secs}s window", events.len());
+        events
+    }
+}
+
+fn p_tags(event: &nostr_sdk::Event) -> Vec<String> {
+    event.tags.iter()
+        .filter_map(|t| {
+            let s = t.as_slice();
+            if s.len() >= 2 && s[0] == "p" { Some(s[1].clone()) } else { None }
+        })
+        .collect()
+}

src/wot/mod.rs

@@ -0,0 +1,2 @@
+pub mod follows;
+pub mod trust;

src/wot/trust.rs

@@ -0,0 +1,28 @@
+use sqlx::SqlitePool;
+use tracing::debug;
+
+/// Recompute trust scores for all publishers with a known WoT level.
+/// Score = WoT contribution − report penalty. Clamped to [−1, 1].
+pub async fn recompute(pool: &SqlitePool) -> anyhow::Result<()> {
+    sqlx::query(
+        "UPDATE publishers SET trust = MAX(-1.0, MIN(1.0,
+           CASE wot_level
+             WHEN 0 THEN 1.0
+             WHEN 1 THEN 0.8
+             WHEN 2 THEN 0.5
+             ELSE 0.0
+           END
+           - (report_count * 0.15)
+         ))
+         WHERE wot_level IS NOT NULL",
+    )
+    .execute(pool)
+    .await?;
+
+    let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM publishers WHERE wot_level IS NOT NULL")
+        .fetch_one(pool)
+        .await?;
+    debug!("trust: recomputed {} publishers", count.0);
+
+    Ok(())
+}