Add mergeable NIP-45 HLL counts

CHANGELOG.md

@@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   maximum-timestamp markers, immediate query and rebroadcast suppression,
   recipient gift-wrap cleanup, undeletable request records, and bounded
   restart-safe physical deletion.
+- Mergeable NIP-45 HyperLogLog sketches for canonical single-target COUNT
+  filters, including hex, address, and arbitrary-string offset derivation.
 
 ### Fixed
 

README.md

@@ -24,9 +24,9 @@ quirk. It also provides an additional Unix-domain socket transport.
 - **Nostr-first protocol behavior.** EVENT/REQ/CLOSE/COUNT/EOSE/OK/NOTICE/
   CLOSED/AUTH and NEG-* are tested against pinned NIPs. Differential tests are
   migration and regression evidence, not a promise to retain upstream bugs.
-- **NIP-42 AUTH, NIP-45 COUNT, NIP-50 ranked content search, NIP-62
-  restart-safe Request to Vanish, NIP-70 protected events, NIP-59 gift-wrap
-  deletion semantics, NIP-77 negentropy set
+- **NIP-42 AUTH, NIP-45 COUNT with mergeable HyperLogLog sketches, NIP-50
+  ranked content search, NIP-62 restart-safe Request to Vanish, NIP-70
+  protected events, NIP-59 gift-wrap deletion semantics, NIP-77 negentropy set
   reconciliation** (persistent LMDB B-tree, tree-backed multi-round sync
   sessions).
 - **Standards-first ephemeral delivery**: ephemeral kinds are live-only by

crates/wok-compat/tests/e2e_transports.rs

@@ -8,6 +8,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio_tungstenite::tungstenite::Message;
 use wok_compat::{sign_event, sign_event_with_key};
 use wok_db::{Env, EnvOptions};
+use wok_query::{HyperLogLog, NostrFilter};
 use wok_relay::Config;
 
 fn now_secs() -> u64 {
@@ -238,6 +239,132 @@ async fn nip62_vanish_is_immediate_and_blocks_rebroadcast() {
     handle.request_shutdown();
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn nip45_count_returns_mergeable_hll_for_canonical_tag_query() {
+    let dir = tempfile::tempdir().unwrap();
+    let env = Env::open(dir.path(), EnvOptions::default()).unwrap();
+    env.ensure_initialized().unwrap();
+    let cfg = test_cfg(dir.path());
+    let handle = wok_relay::start(env, cfg).unwrap();
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let relay = handle.clone();
+    tokio::spawn(async move {
+        let _ = wok_ws::serve_listener(relay, listener).await;
+    });
+    tokio::time::sleep(Duration::from_millis(50)).await;
+
+    let (mut ws, _) = tokio_tungstenite::connect_async(format!("ws://{addr}/"))
+        .await
+        .unwrap();
+    let keys: Vec<Keypair> = {
+        let mut rng = rand::thread_rng();
+        (0..4).map(|_| Keypair::new(SECP256K1, &mut rng)).collect()
+    };
+    let target = sign_event(json!({
+        "created_at": now_secs() - 20,
+        "kind": 1,
+        "tags": [],
+        "content": "hll-target",
+    }));
+    let target_id = target["id"].as_str().unwrap().to_string();
+    ws.send(Message::Text(json!(["EVENT", target]).to_string().into()))
+        .await
+        .unwrap();
+    let _ = recv_until(&mut ws, |text| text.contains("\"OK\"")).await;
+
+    for (index, key) in keys.iter().enumerate() {
+        let reaction = sign_event_with_key(
+            json!({
+                "created_at": now_secs() - 10 + index as u64,
+                "kind": 7,
+                "tags": [["e", target_id]],
+                "content": "+",
+            }),
+            key,
+        );
+        ws.send(Message::Text(json!(["EVENT", reaction]).to_string().into()))
+            .await
+            .unwrap();
+        let accepted = recv_until(&mut ws, |text| text.contains("\"OK\"")).await;
+        assert!(accepted.iter().any(|text| text.contains("true")));
+    }
+    let repeated_author = sign_event_with_key(
+        json!({
+            "created_at": now_secs(),
+            "kind": 7,
+            "tags": [["e", target_id]],
+            "content": "second reaction from one author",
+        }),
+        &keys[0],
+    );
+    ws.send(Message::Text(
+        json!(["EVENT", repeated_author]).to_string().into(),
+    ))
+    .await
+    .unwrap();
+    let accepted = recv_until(&mut ws, |text| text.contains("\"OK\"")).await;
+    assert!(accepted.iter().any(|text| text.contains("true")));
+
+    let count_filter = json!({"#e":[target_id], "kinds":[7]});
+    ws.send(Message::Text(
+        json!(["COUNT", "hll", count_filter]).to_string().into(),
+    ))
+    .await
+    .unwrap();
+    let response = recv_until(&mut ws, |text| text.contains("\"COUNT\"")).await;
+    let body = response
+        .iter()
+        .find_map(|text| {
+            serde_json::from_str::<serde_json::Value>(text)
+                .ok()
+                .filter(|value| value[0] == "COUNT")
+                .map(|value| value[2].clone())
+        })
+        .expect("COUNT body");
+    assert_eq!(body["count"], 5);
+    let actual = body["hll"].as_str().expect("HLL response");
+    assert_eq!(actual.len(), 512);
+
+    let parsed_filter = NostrFilter::parse(&count_filter, 500, 3).unwrap();
+    let mut expected = HyperLogLog::for_filter(&parsed_filter).unwrap();
+    for key in &keys {
+        let (pubkey, _) = key.x_only_public_key();
+        expected.add_pubkey(&pubkey.serialize());
+    }
+    assert_eq!(actual, expected.encode_hex());
+
+    let empty_target = "00".repeat(32);
+    ws.send(Message::Text(
+        json!(["COUNT", "empty-hll", {"#e":[empty_target], "kinds":[7]}])
+            .to_string()
+            .into(),
+    ))
+    .await
+    .unwrap();
+    let empty = recv_until(&mut ws, |text| text.contains("empty-hll")).await;
+    assert!(empty
+        .iter()
+        .any(|text| text.contains(&format!("\"hll\":\"{}\"", "00".repeat(256)))));
+
+    ws.send(Message::Text(
+        json!(["COUNT", "ambiguous-hll", {
+            "#e":[target_id], "#p":["11".repeat(32)], "kinds":[7]
+        }])
+        .to_string()
+        .into(),
+    ))
+    .await
+    .unwrap();
+    let ambiguous = recv_until(&mut ws, |text| text.contains("ambiguous-hll")).await;
+    assert!(ambiguous
+        .iter()
+        .filter(|text| text.contains("\"COUNT\""))
+        .all(|text| !text.contains("\"hll\"")));
+
+    handle.request_shutdown();
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn nip50_ranked_historical_and_live_search() {
     let dir = tempfile::tempdir().unwrap();

crates/wok-compat/tests/nip_conformance.rs

@@ -134,10 +134,21 @@ fn nip45_count_encoding() {
         sub_id: "c".into(),
         count: 3,
         limited: true,
+        hll: Some("00".repeat(256)),
     }
     .to_json();
     assert!(s.contains("\"count\":3"));
     assert!(s.contains("\"limited\":true"));
+    assert!(!s.contains("\"hll\""));
+
+    let s = RelayMessage::Count {
+        sub_id: "c".into(),
+        count: 0,
+        limited: false,
+        hll: Some("00".repeat(256)),
+    }
+    .to_json();
+    assert!(s.contains("\"hll\""));
 }
 
 #[test]

crates/wok-query/src/hll.rs

@@ -0,0 +1,145 @@
+//! NIP-45 HyperLogLog registers for mergeable COUNT responses.
+
+use crate::NostrFilter;
+
+/// NIP-45 fixes the precision at 8 bits: 256 one-byte registers.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HyperLogLog {
+    offset: usize,
+    registers: [u8; 256],
+}
+
+impl HyperLogLog {
+    pub fn for_filter(filter: &NostrFilter) -> Option<Self> {
+        Some(Self {
+            offset: offset_for_filter(filter)?,
+            registers: [0; 256],
+        })
+    }
+
+    pub fn add_pubkey(&mut self, pubkey: &[u8]) {
+        let Some(window) = pubkey
+            .get(self.offset..self.offset.saturating_add(8))
+            .filter(|window| window.len() == 8)
+        else {
+            return;
+        };
+        let register = window[0] as usize;
+        let mut value = 0u64;
+        for byte in &window[1..] {
+            value = (value << 8) | u64::from(*byte);
+        }
+        // `value` occupies the low 56 bits. Remove the eight padding zeroes
+        // counted by u64::leading_zeros, then add one as specified by NIP-45.
+        let rank = value.leading_zeros().saturating_sub(8).saturating_add(1) as u8;
+        self.registers[register] = self.registers[register].max(rank);
+    }
+
+    pub fn encode_hex(&self) -> String {
+        const HEX: &[u8; 16] = b"0123456789abcdef";
+        let mut encoded = String::with_capacity(512);
+        for register in self.registers {
+            encoded.push(HEX[(register >> 4) as usize] as char);
+            encoded.push(HEX[(register & 0x0f) as usize] as char);
+        }
+        encoded
+    }
+
+    #[cfg(test)]
+    fn registers(&self) -> &[u8; 256] {
+        &self.registers
+    }
+}
+
+/// Canonical HLL requests count one target. Multiple tag names or target
+/// values have ambiguous merge semantics, so callers deliberately omit HLL
+/// for those shapes.
+pub fn offset_for_filter(filter: &NostrFilter) -> Option<usize> {
+    if filter.tags.len() != 1 {
+        return None;
+    }
+    let (tag, values) = filter.tags.first_key_value()?;
+    if values.size() != 1 {
+        return None;
+    }
+    let value = values.at(0);
+    let seed: [u8; 32] = if matches!(tag, 'e' | 'p') {
+        value.try_into().ok()?
+    } else if value.len() == 64 {
+        match std::str::from_utf8(value)
+            .ok()
+            .and_then(|hex| wok_event::from_lower_hex_exact(hex).ok())
+        {
+            Some(bytes) => bytes.try_into().ok()?,
+            None => wok_event::sha256(value),
+        }
+    } else if let Some(pubkey) = std::str::from_utf8(value)
+        .ok()
+        .and_then(|address| wok_event::parse_a_tag(address).ok())
+        .map(|(_, pubkey, _)| pubkey)
+    {
+        pubkey
+    } else {
+        wok_event::sha256(value)
+    };
+    Some(usize::from(seed[16] >> 4) + 8)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    fn filter(value: serde_json::Value) -> NostrFilter {
+        NostrFilter::parse(&value, 500, 3).unwrap()
+    }
+
+    #[test]
+    fn offset_covers_hex_address_and_hashed_values() {
+        let hex = format!("{}f{}", "0".repeat(32), "0".repeat(31));
+        assert_eq!(offset_for_filter(&filter(json!({"#e":[hex]}))), Some(23));
+
+        let pubkey = format!("{}a{}", "0".repeat(32), "0".repeat(31));
+        let address = format!("30023:{pubkey}:profile");
+        assert_eq!(
+            offset_for_filter(&filter(json!({"#a":[address]}))),
+            Some(18)
+        );
+
+        let seed = wok_event::sha256(b"arbitrary-target");
+        assert_eq!(
+            offset_for_filter(&filter(json!({"#t":["arbitrary-target"]}))),
+            Some(usize::from(seed[16] >> 4) + 8)
+        );
+    }
+
+    #[test]
+    fn sketch_matches_the_fixed_precision_bit_rules() {
+        let mut hll = HyperLogLog {
+            offset: 8,
+            registers: [0; 256],
+        };
+        let mut pubkey = [0u8; 32];
+        hll.add_pubkey(&pubkey);
+        assert_eq!(hll.registers()[0], 57);
+
+        pubkey[8] = 7;
+        pubkey[9] = 0x10;
+        hll.add_pubkey(&pubkey);
+        assert_eq!(hll.registers()[7], 4);
+        assert_eq!(hll.encode_hex().len(), 512);
+    }
+
+    #[test]
+    fn ambiguous_filter_shapes_do_not_get_a_sketch() {
+        assert!(offset_for_filter(&filter(json!({"kinds":[7]}))).is_none());
+        assert!(
+            offset_for_filter(&filter(json!({"#e":["00".repeat(32), "11".repeat(32)]}))).is_none()
+        );
+        assert!(offset_for_filter(&filter(json!({
+            "#e":["00".repeat(32)],
+            "#p":["11".repeat(32)]
+        })))
+        .is_none());
+    }
+}

crates/wok-query/src/lib.rs

@@ -1,10 +1,12 @@
 pub mod filter;
+pub mod hll;
 pub mod monitor;
 pub mod scan;
 pub mod scheduler;
 pub mod subid;
 
 pub use filter::{dumb_match, FilterValidator, NostrFilter, NostrFilterGroup};
+pub use hll::{offset_for_filter as nip45_hll_offset, HyperLogLog};
 pub use monitor::{ActiveMonitors, Recipient};
 pub use scan::{foreach_by_filter, DbQuery, DbScan};
 pub use scheduler::QueryScheduler;