Add member directory and aggregates, cut citation economics from spec (Draft 0.4)

src/node.rs

@@ -1,11 +1,12 @@
 use std::collections::{HashMap, HashSet};
+use std::fs;
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::{Arc, Mutex};
-use std::time::{Duration, Instant};
+use std::sync::{Arc, Mutex, RwLock};
+use std::time::{Duration, Instant, SystemTime};
 
 use anyhow::{Context, Result, anyhow};
-use axum::extract::State;
+use axum::extract::{Query, State};
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use axum::routing::{get, post};
@@ -15,25 +16,51 @@ use serde_json::{Value, json};
 use tokio::net::TcpListener;
 use tokio::task::JoinHandle;
 
-use crate::config::Config;
+use crate::config::{CLASS_ENRICHMENT, Config, Member, load_members};
 use crate::crypto::Keypair;
 use crate::index::{LocalIndex, SearchHit, response_items};
 use crate::message::{
-    Envelope, QueryBody, ResponseBody, ResponseItem, TYPE_AGGREGATE, TYPE_QUERY, TYPE_RESPONSE,
-    build_response,
+    AggregateBody, Envelope, QueryBody, ResponseBody, ResponseItem, TYPE_AGGREGATE, TYPE_QUERY,
+    TYPE_RESPONSE, build_response,
 };
 
+#[derive(Default)]
+struct Aggregates {
+    sent: HashMap<String, u64>,
+    passed: HashMap<(String, String), u64>,
+}
+
 pub struct Node {
     pub config: Config,
     pub key: Keypair,
     index: LocalIndex,
+    members: RwLock<Vec<Member>>,
+    members_mtime: Mutex<Option<SystemTime>>,
+    aggregates: Mutex<Aggregates>,
     seen: Mutex<HashSet<String>>,
     pending: Mutex<HashMap<String, Vec<(String, ResponseBody)>>>,
+    pending_aggregates: Mutex<HashMap<(String, String), AggregateBody>>,
     client: reqwest::Client,
     sent: AtomicU64,
     received: AtomicU64,
 }
 
+pub fn current_period() -> String {
+    chrono::Utc::now().format("%Y-%m").to_string()
+}
+
+pub fn is_valid_period(period: &str) -> bool {
+    if period.len() == 4 {
+        return period.chars().all(|c| c.is_ascii_digit());
+    }
+    if period.len() != 7 || period.as_bytes()[4] != b'-' {
+        return false;
+    }
+    let digits = period[0..4].chars().all(|c| c.is_ascii_digit());
+    let month: Result<u32, _> = period[5..7].parse();
+    digits && month.map(|m| (1..=12).contains(&m)).unwrap_or(false)
+}
+
 pub struct NodeHandle {
     pub addr: SocketAddr,
     pub pubkey: String,
@@ -83,6 +110,11 @@ impl Node {
     pub fn open(config: Config) -> Result<Arc<Self>> {
         let key = config.load_key()?;
         let index = LocalIndex::open(&config.index_dir())?;
+        let members_path = config.members_path();
+        let members = load_members(&members_path)?;
+        let members_mtime = fs::metadata(&members_path)
+            .and_then(|metadata| metadata.modified())
+            .ok();
         let client = reqwest::Client::builder()
             .timeout(Duration::from_secs(15))
             .build()
@@ -91,14 +123,73 @@ impl Node {
             config,
             key,
             index,
+            members: RwLock::new(members),
+            members_mtime: Mutex::new(members_mtime),
+            aggregates: Mutex::new(Aggregates::default()),
             seen: Mutex::new(HashSet::new()),
             pending: Mutex::new(HashMap::new()),
+            pending_aggregates: Mutex::new(HashMap::new()),
             client,
             sent: AtomicU64::new(0),
             received: AtomicU64::new(0),
         }))
     }
 
+    fn refresh_members(&self) {
+        let path = self.config.members_path();
+        let mtime = fs::metadata(&path)
+            .and_then(|metadata| metadata.modified())
+            .ok();
+        {
+            let last = self.members_mtime.lock().expect("members mtime lock");
+            if *last == mtime {
+                return;
+            }
+        }
+        let Ok(members) = load_members(&path) else {
+            return;
+        };
+        *self.members.write().expect("members lock") = members;
+        *self.members_mtime.lock().expect("members mtime lock") = mtime;
+    }
+
+    fn member_class(&self, members: &[Member], pubkey: &str) -> Option<String> {
+        members
+            .iter()
+            .find(|member| member.pubkey == pubkey)
+            .map(|member| member.class.clone())
+    }
+
+    pub fn aggregate_for(&self, period: &str, member: Option<&str>) -> AggregateBody {
+        let aggregates = self.aggregates.lock().expect("aggregates lock");
+        let in_period = |candidate: &str| {
+            if period.len() == 4 {
+                candidate.starts_with(period)
+            } else {
+                candidate == period
+            }
+        };
+        let sent = aggregates
+            .sent
+            .iter()
+            .filter(|(key, _)| in_period(key))
+            .map(|(_, count)| *count)
+            .sum();
+        let passed = aggregates
+            .passed
+            .iter()
+            .filter(|((key, key_period), _)| {
+                in_period(key_period) && member.map(|m| m == key).unwrap_or(true)
+            })
+            .map(|(_, count)| *count)
+            .sum();
+        AggregateBody {
+            period: period.to_string(),
+            sent,
+            passed,
+        }
+    }
+
     pub fn doc_count(&self) -> u64 {
         self.index.doc_count()
     }
@@ -158,6 +249,10 @@ impl Node {
                 .expect("pending lock")
                 .insert(qid.clone(), Vec::new());
             self.sent.fetch_add(1, Ordering::SeqCst);
+            {
+                let mut aggregates = self.aggregates.lock().expect("aggregates lock");
+                *aggregates.sent.entry(current_period()).or_default() += 1;
+            }
             let envelope = Envelope::new(&self.key, TYPE_QUERY, serde_json::to_value(&query)?);
             self.publish(&envelope).await;
             let timeout = Duration::from_millis(timeout_ms.unwrap_or(self.config.query.timeout_ms));
@@ -174,6 +269,13 @@ impl Node {
             self.received
                 .fetch_add(collected.len() as u64, Ordering::SeqCst);
             for (member, body) in collected {
+                {
+                    let mut aggregates = self.aggregates.lock().expect("aggregates lock");
+                    *aggregates
+                        .passed
+                        .entry((member.clone(), current_period()))
+                        .or_default() += 1;
+                }
                 responses.push(RemoteResponse {
                     member,
                     results: body.results,
@@ -231,9 +333,14 @@ impl Node {
         if envelope.verify().is_err() {
             return;
         }
-        if !self.config.node.trusted_keys.is_empty()
-            && !self.config.node.trusted_keys.contains(&envelope.from)
-        {
+        self.refresh_members();
+        let (class, listed) = {
+            let members = self.members.read().expect("members lock");
+            let class = self.member_class(&members, &envelope.from);
+            let listed = members.is_empty() || class.is_some();
+            (class, listed)
+        };
+        if !listed {
             return;
         }
         match envelope.msg_type.as_str() {
@@ -269,12 +376,45 @@ impl Node {
                 let Ok(body) = envelope.parse_body::<ResponseBody>() else {
                     return;
                 };
+                if class.as_deref() == Some(CLASS_ENRICHMENT)
+                    && body.results.iter().any(|result| result.content.is_some())
+                {
+                    return;
+                }
                 let mut pending = self.pending.lock().expect("pending lock");
                 if let Some(list) = pending.get_mut(&body.qid) {
                     list.push((envelope.from.clone(), body));
                 }
             }
-            TYPE_AGGREGATE => {}
+            TYPE_AGGREGATE => {
+                let Ok(value) = envelope.parse_body::<Value>() else {
+                    return;
+                };
+                let Some(period) = value.get("period").and_then(Value::as_str) else {
+                    return;
+                };
+                if !is_valid_period(period) {
+                    return;
+                }
+                if value.get("sent").is_some() {
+                    let Ok(body) = envelope.parse_body::<AggregateBody>() else {
+                        return;
+                    };
+                    let mut pending = self
+                        .pending_aggregates
+                        .lock()
+                        .expect("aggregate reply lock");
+                    pending.insert((envelope.from.clone(), body.period.clone()), body);
+                    return;
+                }
+                let node = self.clone();
+                let period = period.to_string();
+                let requester = envelope.from.clone();
+                let relay = relay.to_string();
+                tokio::spawn(async move {
+                    node.serve_aggregate(&period, &requester, &relay).await;
+                });
+            }
             _ => {}
         }
     }
@@ -287,6 +427,59 @@ impl Node {
         }
         let body = build_response(&query.qid, response_items(&hits), total, max);
         let envelope = Envelope::new(&self.key, TYPE_RESPONSE, serde_json::to_value(&body)?);
+        self.send_unicast(querier, envelope, relay).await
+    }
+
+    pub async fn request_aggregate(
+        &self,
+        to: &str,
+        period: &str,
+        timeout_ms: Option<u64>,
+    ) -> Result<AggregateBody> {
+        if !is_valid_period(period) {
+            return Err(anyhow!("period must be YYYY or YYYY-MM (monthly floor)"));
+        }
+        let relay = self
+            .config
+            .node
+            .relays
+            .first()
+            .ok_or_else(|| anyhow!("no relays configured"))?
+            .clone();
+        let envelope = Envelope::new(&self.key, TYPE_AGGREGATE, json!({ "period": period }));
+        self.send_unicast(to, envelope, &relay).await?;
+        let key = (to.to_string(), period.to_string());
+        let deadline = Instant::now()
+            + Duration::from_millis(timeout_ms.unwrap_or(self.config.query.timeout_ms));
+        loop {
+            {
+                let pending = self
+                    .pending_aggregates
+                    .lock()
+                    .expect("aggregate reply lock");
+                if let Some(body) = pending.get(&key) {
+                    return Ok(body.clone());
+                }
+            }
+            if Instant::now() >= deadline {
+                return Err(anyhow!("aggregate request timed out"));
+            }
+            tokio::time::sleep(Duration::from_millis(25)).await;
+        }
+    }
+
+    async fn serve_aggregate(&self, period: &str, requester: &str, relay: &str) {
+        let body = self.aggregate_for(period, Some(requester));
+        let Ok(value) = serde_json::to_value(&body) else {
+            return;
+        };
+        let envelope = Envelope::new(&self.key, TYPE_AGGREGATE, value);
+        if let Err(error) = self.send_unicast(requester, envelope, relay).await {
+            eprintln!("aggregate reply failed: {error}");
+        }
+    }
+
+    async fn send_unicast(&self, to: &str, envelope: Envelope, relay: &str) -> Result<()> {
         let mut relays = vec![relay.to_string()];
         for configured in &self.config.node.relays {
             if !relays.contains(configured) {
@@ -295,16 +488,12 @@ impl Node {
         }
         let mut last_error: Option<anyhow::Error> = None;
         for candidate in relays {
-            let url = format!(
-                "{}/v1/unicast?to={}",
-                candidate.trim_end_matches('/'),
-                querier
-            );
+            let url = format!("{}/v1/unicast?to={}", candidate.trim_end_matches('/'), to);
             match self.client.post(&url).json(&envelope).send().await {
                 Ok(response) if response.status().is_success() => return Ok(()),
                 Ok(response) => {
                     last_error = Some(anyhow!(
-                        "relay {candidate} rejected response: {}",
+                        "relay {candidate} rejected message: {}",
                         response.status()
                     ))
                 }
@@ -351,6 +540,8 @@ pub fn router(node: Arc<Node>) -> Router {
     Router::new()
         .route("/v1/local/query", post(local_query))
         .route("/v1/local/status", get(local_status))
+        .route("/v1/local/aggregates", get(local_aggregates))
+        .route("/v1/local/aggregate/request", post(local_aggregate_request))
         .with_state(node)
 }
 
@@ -390,19 +581,94 @@ async fn local_query(
 }
 
 async fn local_status(State(node): State<Arc<Node>>) -> Response {
+    let aggregates = node.aggregate_for(&current_period(), None);
     Json(json!({
         "name": node.config.node.name,
         "pubkey": node.key.public_hex(),
         "listen": node.config.node.listen,
         "relays": node.config.node.relays,
         "responder": node.config.node.responder,
+        "members": node.members.read().expect("members lock").len(),
         "doc_count": node.doc_count(),
         "sent": node.sent(),
         "received": node.received(),
+        "aggregates": aggregates,
     }))
     .into_response()
 }
 
+#[derive(Deserialize)]
+struct AggregateParams {
+    period: Option<String>,
+}
+
+async fn local_aggregates(
+    State(node): State<Arc<Node>>,
+    Query(params): Query<AggregateParams>,
+) -> Response {
+    let period = params.period.unwrap_or_else(current_period);
+    if !is_valid_period(&period) {
+        return (
+            StatusCode::BAD_REQUEST,
+            Json(json!({ "error": "period must be YYYY or YYYY-MM (monthly floor)" })),
+        )
+            .into_response();
+    }
+    Json(node.aggregate_for(&period, None)).into_response()
+}
+
+#[derive(Deserialize)]
+struct AggregateRequest {
+    to: String,
+    period: String,
+    timeout_ms: Option<u64>,
+}
+
+async fn local_aggregate_request(
+    State(node): State<Arc<Node>>,
+    Json(request): Json<AggregateRequest>,
+) -> Response {
+    match node
+        .request_aggregate(&request.to, &request.period, request.timeout_ms)
+        .await
+    {
+        Ok(body) => Json(body).into_response(),
+        Err(error) => (
+            StatusCode::BAD_GATEWAY,
+            Json(json!({ "error": error.to_string() })),
+        )
+            .into_response(),
+    }
+}
+
+pub async fn control_aggregate_request(
+    base: &str,
+    to: &str,
+    period: &str,
+    timeout_ms: Option<u64>,
+) -> Result<Value> {
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(30))
+        .build()?;
+    let url = format!("{}/v1/local/aggregate/request", base.trim_end_matches('/'));
+    let response = client
+        .post(&url)
+        .json(&json!({
+            "to": to,
+            "period": period,
+            "timeout_ms": timeout_ms,
+        }))
+        .send()
+        .await
+        .with_context(|| format!("calling local node at {url} (is `frxd serve` running?)"))?;
+    let status = response.status();
+    let value: Value = response.json().await.context("parsing node response")?;
+    if !status.is_success() {
+        return Err(anyhow!("local node error {status}: {value}"));
+    }
+    Ok(value)
+}
+
 pub async fn control_query(
     base: &str,
     text: &str,