From b6705d5b8556b062aee3ec527047d0e5ad7d3977 Mon Sep 17 00:00:00 2001 From: enki Date: Sun, 17 May 2026 11:31:59 -0700 Subject: [PATCH] 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 | 3 +- src/bin/kindexr-cli.rs | 150 +++++++++++++++++++++++ src/config.rs | 2 + src/db/migrations/003_wot.sql | 26 ++++ src/db/queries.rs | 222 +++++++++++++++++++++++++++++++++- src/lib.rs | 1 + src/main.rs | 4 +- src/nostr/reader.rs | 132 +++++++++++++++----- src/wot/follows.rs | 157 ++++++++++++++++++++++++ src/wot/mod.rs | 2 + src/wot/trust.rs | 28 +++++ 11 files changed, 696 insertions(+), 31 deletions(-) create mode 100644 src/bin/kindexr-cli.rs create mode 100644 src/db/migrations/003_wot.sql create mode 100644 src/wot/follows.rs create mode 100644 src/wot/mod.rs create mode 100644 src/wot/trust.rs diff --git a/.gitignore b/.gitignore index 05e67d9..fe67cc5 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ -bin/ +/bin/ +target/ *.db diff --git a/src/bin/kindexr-cli.rs b/src/bin/kindexr-cli.rs new file mode 100644 index 0000000..23a820e --- /dev/null +++ b/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) +} diff --git a/src/config.rs b/src/config.rs index eb7e1b8..94f497e 100644 --- a/src/config.rs +++ b/src/config.rs @@ -44,6 +44,7 @@ pub struct CurationConfig { pub blocklist: Vec, pub curation_sets: Vec, pub exclude_categories: Vec, + 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, diff --git a/src/db/migrations/003_wot.sql b/src/db/migrations/003_wot.sql new file mode 100644 index 0000000..b4789bf --- /dev/null +++ b/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); diff --git a/src/db/queries.rs b/src/db/queries.rs index ef7a3ce..aa436af 100644 --- a/src/db/queries.rs +++ b/src/db/queries.rs @@ -173,7 +173,11 @@ pub async fn search(pool: &SqlitePool, p: &SearchParams) -> anyhow::Result anyhow::Result(&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, + pub trust: f64, + pub blocked: bool, + pub muted: bool, + pub wot_level: Option, + pub report_count: i32, + pub torrents_n: i32, + pub first_seen: Option, + pub last_seen: Option, +} + +impl sqlx::FromRow<'_, sqlx::sqlite::SqliteRow> for PublisherRow { + fn from_row(row: &sqlx::sqlite::SqliteRow) -> Result { + use sqlx::Row; + Ok(PublisherRow { + pubkey: row.try_get("pubkey")?, + name: row.try_get("name")?, + trust: row.try_get("trust")?, + blocked: row.try_get::("blocked")? != 0, + muted: row.try_get::("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 { + let row: Option<(i64, i64, Option)> = 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> { + 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> { + 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> { diff --git a/src/lib.rs b/src/lib.rs index 643cf71..1602f40 100644 --- a/src/lib.rs +++ b/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}, diff --git a/src/main.rs b/src/main.rs index b9ce463..18ff5d6 100644 --- a/src/main.rs +++ b/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())) diff --git a/src/nostr/reader.rs b/src/nostr/reader.rs index 52707ed..193ed0a 100644 --- a/src/nostr/reader.rs +++ b/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, pool: SqlitePool, connected: Arc { - 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, pool: SqlitePool, connected: Arc = 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, pool: SqlitePool, connected: Arc { - 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, pool: SqlitePool, connected: Arc { - // 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"); + } } } } diff --git a/src/wot/follows.rs b/src/wot/follows.rs new file mode 100644 index 0000000..02faa88 --- /dev/null +++ b/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, + pool: SqlitePool, +} + +impl WotBuilder { + pub fn new(cfg: Arc, 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 = 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 = 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 { + 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::(); + 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 { + 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() +} diff --git a/src/wot/mod.rs b/src/wot/mod.rs new file mode 100644 index 0000000..1ffd405 --- /dev/null +++ b/src/wot/mod.rs @@ -0,0 +1,2 @@ +pub mod follows; +pub mod trust; diff --git a/src/wot/trust.rs b/src/wot/trust.rs new file mode 100644 index 0000000..7d09fa3 --- /dev/null +++ b/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(()) +}