Merge pull request #34 from at-microcosm/spacedust

spacedust link firehose

Cargo.lock

···
 checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26"
 
 [[package]]
+name = "aws-lc-rs"
+version = "1.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "93fcc8f365936c834db5514fc45aee5b1202d677e6b40e48468aaaa8183ca8c7"
+dependencies = [
+ "aws-lc-sys",
+ "zeroize",
+]
+
+[[package]]
+name = "aws-lc-sys"
+version = "0.29.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "61b1d86e7705efe1be1b569bab41d4fa1e14e220b60a160f78de2db687add079"
+dependencies = [
+ "bindgen 0.69.5",
+ "cc",
+ "cmake",
+ "dunce",
+ "fs_extra",
+]
+
+[[package]]
 name = "axum"
 version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "itertools 0.12.1",
  "lazy_static",
  "lazycell",
+ "log",
+ "prettyplease",
  "proc-macro2",
  "quote",
  "regex",
  "rustc-hash 1.1.0",
  "shlex",
  "syn",
+ "which",
 ]
 
 [[package]]
···
 
 [[package]]
 name = "clap"
-version = "4.5.35"
+version = "4.5.40"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d8aa86934b44c19c50f87cc2790e19f54f7a67aedb64101c2e1a2e5ecfb73944"
+checksum = "40b6887a1d8685cebccf115538db5c0efe625ccac9696ad45c409d96566e910f"
 dependencies = [
  "clap_builder",
  "clap_derive",
···
 
 [[package]]
 name = "clap_builder"
-version = "4.5.35"
+version = "4.5.40"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2414dbb2dd0695280da6ea9261e327479e9d37b0630f6b53ba2a11c60c679fd9"
+checksum = "e0c66c08ce9f0c698cbce5c0279d0bb6ac936d8674174fe48f736533b964f59e"
 dependencies = [
  "anstream",
  "anstyle",
···
 
 [[package]]
 name = "clap_derive"
-version = "4.5.32"
+version = "4.5.40"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09176aae279615badda0765c0c0b3f6ed53f4709118af73cf4655d85d1530cd7"
+checksum = "d2c7947ae4cc3d851207c1adb5b5e260ff0cca11446b1d6d1423788e442257ce"
 dependencies = [
  "heck",
  "proc-macro2",
···
 ]
 
 [[package]]
+name = "cmake"
+version = "0.1.54"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e7caa3f9de89ddbe2c607f4101924c5abec803763ae9534e4f4d7d8f84aa81f0"
+dependencies = [
+ "cc",
+]
+
+[[package]]
 name = "colorchoice"
 version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "tokio",
  "tokio-util",
  "tower-http",
- "tungstenite",
+ "tungstenite 0.26.2",
  "zstd",
 ]
 
···
 version = "0.9.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "91e195e091a93c46f7102ec7818a2aa394e1e1771c3ab4825963fa03e45afb8f"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
+[[package]]
+name = "core-foundation"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6"
 dependencies = [
  "core-foundation-sys",
  "libc",
···
 
 [[package]]
 name = "ctrlc"
-version = "3.4.6"
+version = "3.4.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "697b5419f348fd5ae2478e8018cb016c00a5881c7f46c717de98ffd135a5651c"
+checksum = "46f93780a459b7d656ef7f071fe699c4d3d2cb201c4b24d085b6ddc505276e73"
 dependencies = [
  "nix",
  "windows-sys 0.59.0",
···
 
 [[package]]
 name = "dropshot"
-version = "0.16.0"
+version = "0.16.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a37c505dad56e0c1fa5ed47e29fab1a1ab2d1a9d93e952024bb47168969705f6"
+checksum = "50e8fed669e35e757646ad10f97c4d26dd22cce3da689b307954f7000d2719d0"
 dependencies = [
  "async-stream",
  "async-trait",
···
  "openapiv3",
  "paste",
  "percent-encoding",
- "rustls",
+ "rustls 0.22.4",
  "rustls-pemfile",
  "schemars",
  "scopeguard",
···
  "slog-term",
  "thiserror 2.0.12",
  "tokio",
- "tokio-rustls",
+ "tokio-rustls 0.25.0",
  "toml",
  "uuid",
  "version_check",
···
 
 [[package]]
 name = "dropshot_endpoint"
-version = "0.16.0"
+version = "0.16.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b1a6db3728f0195e3ad62807649913aaba06d45421e883416e555e51464ef67"
+checksum = "acebb687581abdeaa2c89fa448818a5f803b0e68e5d7e7a1cf585a8f3c5c57ac"
 dependencies = [
  "heck",
  "proc-macro2",
···
 ]
 
 [[package]]
+name = "dunce"
+version = "1.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
+
+[[package]]
 name = "dyn-clone"
 version = "1.0.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "rustix 0.38.44",
  "windows-sys 0.52.0",
 ]
+
+[[package]]
+name = "fs_extra"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
 [[package]]
 name = "futures"
···
 checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
 
 [[package]]
+name = "home"
+version = "0.5.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "589533453244b0995c858700322199b2becb13b627df2851f64a2775d024abcf"
+dependencies = [
+ "windows-sys 0.59.0",
+]
+
+[[package]]
 name = "hostname"
 version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[package]]
+name = "hyper-rustls"
+version = "0.27.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e3c93eb611681b207e1fe55d5a71ecf91572ec8a6705cdb6857f7d8d5242cf58"
+dependencies = [
+ "http",
+ "hyper",
+ "hyper-util",
+ "rustls 0.23.28",
+ "rustls-native-certs",
+ "rustls-pki-types",
+ "tokio",
+ "tokio-rustls 0.26.2",
+ "tower-service",
+]
+
+[[package]]
 name = "hyper-util"
 version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "serde_json",
  "thiserror 2.0.12",
  "tokio",
- "tokio-tungstenite",
+ "tokio-tungstenite 0.26.2",
  "url",
  "zstd",
 ]
···
 
 [[package]]
 name = "metrics-exporter-prometheus"
-version = "0.17.0"
+version = "0.17.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "df88858cd28baaaf2cfc894e37789ed4184be0e1351157aec7bf3c2266c793fd"
+checksum = "989903b4c7abfa6827a8d1128ef42faf83f8969d429797c5431f236f2cae8b8b"
 dependencies = [
  "base64 0.22.1",
  "http-body-util",
  "hyper",
+ "hyper-rustls",
  "hyper-util",
  "indexmap 2.9.0",
  "ipnet",
···
  "openssl-probe",
  "openssl-sys",
  "schannel",
- "security-framework",
+ "security-framework 2.11.1",
  "security-framework-sys",
  "tempfile",
 ]
 
 [[package]]
 name = "nix"
-version = "0.29.0"
+version = "0.30.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46"
+checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6"
 dependencies = [
  "bitflags",
  "cfg-if",
···
 ]
 
 [[package]]
+name = "prettyplease"
+version = "0.2.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6837b9e10d61f45f987d50808f83d1ee3d206c66acf650c3e4ae2e1f6ddedf55"
+dependencies = [
+ "proc-macro2",
+ "syn",
+]
+
+[[package]]
 name = "proc-macro2"
 version = "1.0.94"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "log",
  "ring",
  "rustls-pki-types",
- "rustls-webpki",
+ "rustls-webpki 0.102.8",
+ "subtle",
+ "zeroize",
+]
+
+[[package]]
+name = "rustls"
+version = "0.23.28"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7160e3e10bf4535308537f3c4e1641468cd0e485175d6163087c0393c7d46643"
+dependencies = [
+ "aws-lc-rs",
+ "once_cell",
+ "rustls-pki-types",
+ "rustls-webpki 0.103.3",
  "subtle",
  "zeroize",
 ]
 
 [[package]]
+name = "rustls-native-certs"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7fcff2dd52b58a8d98a70243663a0d234c4e2b79235637849d15913394a247d3"
+dependencies = [
+ "openssl-probe",
+ "rustls-pki-types",
+ "schannel",
+ "security-framework 3.2.0",
+]
+
+[[package]]
 name = "rustls-pemfile"
 version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "64ca1bc8749bd4cf37b5ce386cc146580777b4e8572c7b97baf22c83f444bee9"
 dependencies = [
+ "ring",
+ "rustls-pki-types",
+ "untrusted",
+]
+
+[[package]]
+name = "rustls-webpki"
+version = "0.103.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e4a72fe2bcf7a6ac6fd7d0b9e5cb68aeb7d4c0a0271730218b3e92d43b4eb435"
+dependencies = [
+ "aws-lc-rs",
  "ring",
  "rustls-pki-types",
  "untrusted",
···
 checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02"
 dependencies = [
  "bitflags",
- "core-foundation",
+ "core-foundation 0.9.4",
+ "core-foundation-sys",
+ "libc",
+ "security-framework-sys",
+]
+
+[[package]]
+name = "security-framework"
+version = "3.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "271720403f46ca04f7ba6f55d438f8bd878d6b8ca0a1046e8228c4145bcbb316"
+dependencies = [
+ "bitflags",
+ "core-foundation 0.10.1",
  "core-foundation-sys",
  "libc",
  "security-framework-sys",
···
 
 [[package]]
 name = "serde_spanned"
-version = "0.6.8"
+version = "0.6.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "87607cb1398ed59d48732e575a4c28a7a8ebf2454b964fe3f224f2afc07909e1"
+checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3"
 dependencies = [
  "serde",
 ]
···
 ]
 
 [[package]]
+name = "spacedust"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "clap",
+ "ctrlc",
+ "dropshot",
+ "env_logger",
+ "futures",
+ "http",
+ "jetstream",
+ "links",
+ "log",
+ "metrics",
+ "metrics-exporter-prometheus 0.17.1",
+ "rand 0.9.1",
+ "schemars",
+ "semver",
+ "serde",
+ "serde_json",
+ "serde_qs",
+ "thiserror 2.0.12",
+ "tinyjson",
+ "tokio",
+ "tokio-tungstenite 0.27.0",
+ "tokio-util",
+]
+
+[[package]]
 name = "spin"
 version = "0.9.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 
 [[package]]
 name = "syn"
-version = "2.0.100"
+version = "2.0.103"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b09a44accad81e1ba1cd74a32461ba89dee89095ba17b32f5d03683b1b1fc2a0"
+checksum = "e4307e30089d6fd6aff212f2da3a1f9e32f3223b1f010fb09b7c95f90f3ca1e8"
 dependencies = [
  "proc-macro2",
  "quote",
···
 
 [[package]]
 name = "tokio"
-version = "1.44.2"
+version = "1.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e6b88822cbe49de4185e3a4cbf8321dd487cf5fe0c5c65695fef6346371e9c48"
+checksum = "75ef51a33ef1da925cea3e4eb122833cb377c61439ca401b770f54902b806779"
 dependencies = [
  "backtrace",
  "bytes",
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "775e0c0f0adb3a2f22a00c4745d728b479985fc15ee7ca6a2608388c5569860f"
 dependencies = [
- "rustls",
+ "rustls 0.22.4",
  "rustls-pki-types",
  "tokio",
 ]
 
 [[package]]
+name = "tokio-rustls"
+version = "0.26.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8e727b36a1a0e8b74c376ac2211e40c2c8af09fb4013c60d910495810f008e9b"
+dependencies = [
+ "rustls 0.23.28",
+ "tokio",
+]
+
+[[package]]
 name = "tokio-tungstenite"
 version = "0.26.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "native-tls",
  "tokio",
  "tokio-native-tls",
- "tungstenite",
+ "tungstenite 0.26.2",
+]
+
+[[package]]
+name = "tokio-tungstenite"
+version = "0.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "489a59b6730eda1b0171fcfda8b121f4bee2b35cba8645ca35c5f7ba3eb736c1"
+dependencies = [
+ "futures-util",
+ "log",
+ "tokio",
+ "tungstenite 0.27.0",
 ]
 
 [[package]]
···
 
 [[package]]
 name = "toml"
-version = "0.8.20"
+version = "0.8.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cd87a5cdd6ffab733b2f74bc4fd7ee5fff6634124999ac278c35fc78c6120148"
+checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362"
 dependencies = [
  "serde",
  "serde_spanned",
···
 
 [[package]]
 name = "toml_datetime"
-version = "0.6.8"
+version = "0.6.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0dd7358ecb8fc2f8d014bf86f6f638ce72ba252a2c3a2572f2a795f1d23efb41"
+checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c"
 dependencies = [
  "serde",
 ]
 
 [[package]]
 name = "toml_edit"
-version = "0.22.24"
+version = "0.22.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "17b4795ff5edd201c7cd6dca065ae59972ce77d1b80fa0a84d94950ece7d1474"
+checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a"
 dependencies = [
  "indexmap 2.9.0",
  "serde",
  "serde_spanned",
  "toml_datetime",
+ "toml_write",
  "winnow",
 ]
 
 [[package]]
+name = "toml_write"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801"
+
+[[package]]
 name = "tower"
 version = "0.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[package]]
+name = "tungstenite"
+version = "0.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eadc29d668c91fcc564941132e17b28a7ceb2f3ebf0b9dae3e03fd7a6748eb0d"
+dependencies = [
+ "bytes",
+ "data-encoding",
+ "http",
+ "httparse",
+ "log",
+ "rand 0.9.1",
+ "sha1",
+ "thiserror 2.0.12",
+ "utf-8",
+]
+
+[[package]]
 name = "typenum"
 version = "1.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "log",
  "lsm-tree",
  "metrics",
- "metrics-exporter-prometheus 0.17.0",
+ "metrics-exporter-prometheus 0.17.1",
  "schemars",
  "semver",
  "serde",
···
 ]
 
 [[package]]
+name = "which"
+version = "4.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "87ba24419a2078cd2b0f2ede2691b6c66d8e47836da3b6db8265ebad47afbfc7"
+dependencies = [
+ "either",
+ "home",
+ "once_cell",
+ "rustix 0.38.44",
+]
+
+[[package]]
 name = "winapi"
 version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 
 [[package]]
 name = "winnow"
-version = "0.7.6"
+version = "0.7.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "63d3fcd9bba44b03821e7d699eeee959f3126dcc4aa8e4ae18ec617c2a5cea10"
+checksum = "74c7b26e3480b707944fc872477815d29a8e429d2f93a1ce000f5fa84a15cbcd"
 dependencies = [
  "memchr",
 ]

Cargo.toml

···
     "jetstream",
     "ufos",
     "ufos/fuzz",
+    "spacedust",
 ]

jetstream/src/error.rs

···
 pub enum JetstreamEventError {
     #[error("failed to load built-in zstd dictionary for decoding: {0}")]
     CompressionDictionaryError(io::Error),
-
     #[error("failed to send ping or pong: {0}")]
     PingPongError(#[from] tokio_tungstenite::tungstenite::Error),
+    #[error("no messages received within ttl")]
+    NoMessagesReceived,
     #[error("jetstream event receiver closed")]
     ReceiverClosedError,
 }

jetstream/src/lib.rs

···
         Receiver,
         Sender,
     },
+    time::timeout,
 };
 use tokio_tungstenite::{
     connect_async,
···
     /// can help prevent that if your consumer sometimes pauses, at a cost of higher memory
     /// usage while events are buffered.
     pub channel_size: usize,
+    /// How long since the last jetstream message before we consider the connection dead
+    ///
+    /// Default: 15s
+    pub liveliness_ttl: Duration,
 }
 
 impl Default for JetstreamConfig {
···
             omit_user_agent_jetstream_info: false,
             replay_on_reconnect: false,
             channel_size: 4096, // a few seconds of firehose buffer
+            liveliness_ttl: Duration::from_secs(15),
         }
     }
 }
···
 
         let (send_channel, receive_channel) = channel(self.config.channel_size);
         let replay_on_reconnect = self.config.replay_on_reconnect;
+        let liveliness_ttl = self.config.liveliness_ttl;
         let build_request = self.config.get_request_builder();
 
         tokio::task::spawn(async move {
···
                 if let Ok((ws_stream, _)) = connect_async(req).await {
                     let t_connected = Instant::now();
                     log::info!("jetstream connected. starting websocket task...");
-                    if let Err(e) =
-                        websocket_task(dict, ws_stream, send_channel.clone(), &mut last_cursor)
-                            .await
+                    if let Err(e) = websocket_task(
+                        dict,
+                        ws_stream,
+                        send_channel.clone(),
+                        &mut last_cursor,
+                        liveliness_ttl,
+                    )
+                    .await
                     {
                         match e {
                             JetstreamEventError::ReceiverClosedError => {
···
                             JetstreamEventError::CompressionDictionaryError(_) => {
                                 #[cfg(feature="metrics")]
                                 counter!("jetstream_disconnects", "reason" => "zstd", "fatal" => "no").increment(1);
+                            }
+                            JetstreamEventError::NoMessagesReceived => {
+                                #[cfg(feature="metrics")]
+                                counter!("jetstream_disconnects", "reason" => "ttl", "fatal" => "no").increment(1);
                             }
                             JetstreamEventError::PingPongError(_) => {
                                 #[cfg(feature="metrics")]
···
     ws: WebSocketStream<MaybeTlsStream<TcpStream>>,
     send_channel: JetstreamSender,
     last_cursor: &mut Option<Cursor>,
+    liveliness_ttl: Duration,
 ) -> Result<(), JetstreamEventError> {
     // TODO: Use the write half to allow the user to change configuration settings on the fly.
     let (mut socket_write, mut socket_read) = ws.split();
 
     let mut closing_connection = false;
     loop {
-        match socket_read.next().await {
+        let next = match timeout(liveliness_ttl, socket_read.next()).await {
+            Ok(n) => n,
+            Err(_) => {
+                log::warn!("jetstream no events for {liveliness_ttl:?}, closing");
+                _ = socket_write.close().await;
+                return Err(JetstreamEventError::NoMessagesReceived);
+            }
+        };
+        match next {
             Some(Ok(message)) => match message {
                 Message::Text(json) => {
                     #[cfg(feature = "metrics")]

spacedust/Cargo.toml

···
+[package]
+name = "spacedust"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+async-trait = "0.1.88"
+clap = { version = "4.5.40", features = ["derive"] }
+ctrlc = "3.4.7"
+dropshot = "0.16.2"
+env_logger = "0.11.8"
+futures = "0.3.31"
+http = "1.3.1"
+jetstream = { path = "../jetstream", features = ["metrics"] }
+links = { path = "../links" }
+log = "0.4.27"
+metrics = "0.24.2"
+metrics-exporter-prometheus = { version = "0.17.1", features = ["http-listener"] }
+rand = "0.9.1"
+schemars = "0.8.22"
+semver = "1.0.26"
+serde = { version = "1.0.219", features = ["derive"] }
+serde_json = "1.0.140"
+serde_qs = "1.0.0-rc.3"
+thiserror = "2.0.12"
+tinyjson = "2.5.1"
+tokio = { version = "1.45.1", features = ["full"] }
+tokio-tungstenite = "0.27.0"
+tokio-util = "0.7.15"

spacedust/src/consumer.rs

···
+use tokio_util::sync::CancellationToken;
+use crate::LinkEvent;
+use crate::error::ConsumerError;
+use crate::removable_delay_queue;
+use jetstream::{
+    DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector,
+    events::{CommitOp, Cursor, EventKind},
+};
+use links::collect_links;
+use tokio::sync::broadcast;
+
+const MAX_LINKS_PER_EVENT: usize = 100;
+
+pub async fn consume(
+    b: broadcast::Sender<LinkEvent>,
+    d: removable_delay_queue::Input<(String, usize), LinkEvent>,
+    jetstream_endpoint: String,
+    cursor: Option<Cursor>,
+    no_zstd: bool,
+    shutdown: CancellationToken,
+) -> Result<(), ConsumerError> {
+    let endpoint = DefaultJetstreamEndpoints::endpoint_or_shortcut(&jetstream_endpoint);
+    if endpoint == jetstream_endpoint {
+        log::info!("connecting to jetstream at {endpoint}");
+    } else {
+        log::info!("connecting to jetstream at {jetstream_endpoint} => {endpoint}");
+    }
+    let config: JetstreamConfig = JetstreamConfig {
+        endpoint,
+        compression: if no_zstd {
+            JetstreamCompression::None
+        } else {
+            JetstreamCompression::Zstd
+        },
+        replay_on_reconnect: true,
+        channel_size: 1024, // buffer up to ~1s of jetstream events
+        ..Default::default()
+    };
+    let mut receiver = JetstreamConnector::new(config)?
+        .connect_cursor(cursor)
+        .await?;
+
+    log::info!("receiving jetstream messages..");
+    loop {
+        if shutdown.is_cancelled() {
+            log::info!("exiting consumer for shutdown");
+            return Ok(());
+        }
+        let Some(event) = receiver.recv().await else {
+            log::error!("could not receive jetstream event, bailing");
+            break;
+        };
+
+        if event.kind != EventKind::Commit {
+            continue;
+        }
+        let Some(commit) = event.commit else {
+            log::warn!("jetstream commit event missing commit data, ignoring");
+            continue;
+        };
+
+        let at_uri = format!("at://{}/{}/{}", &*event.did, &*commit.collection, &*commit.rkey);
+
+        // TODO: keep a buffer and remove quick deletes to debounce notifs
+        // for now we just drop all deletes eek
+        if commit.operation == CommitOp::Delete {
+            d.remove_range((at_uri.clone(), 0)..=(at_uri.clone(), MAX_LINKS_PER_EVENT)).await;
+            continue;
+        }
+        let Some(record) = commit.record else {
+            log::warn!("jetstream commit update/delete missing record, ignoring");
+            continue;
+        };
+
+        let jv = match record.get().parse() {
+            Ok(v) => v,
+            Err(e) => {
+                log::warn!("jetstream record failed to parse, ignoring: {e}");
+                continue;
+            }
+        };
+
+        // todo: indicate if the link limit was reached (-> links omitted)
+        for (i, link) in collect_links(&jv).into_iter().enumerate() {
+            if i >= MAX_LINKS_PER_EVENT {
+                log::warn!("jetstream event has too many links, ignoring the rest");
+                break;
+            }
+            let link_ev = LinkEvent {
+                collection: commit.collection.to_string(),
+                path: link.path,
+                origin: at_uri.clone(),
+                rev: commit.rev.to_string(),
+                target: link.target.into_string(),
+            };
+            let _ = b.send(link_ev.clone()); // only errors if no subscribers are connected, which is just fine.
+            d.enqueue((at_uri.clone(), i), link_ev)
+                .await
+                .map_err(|_| ConsumerError::DelayQueueOutputDropped)?;
+        }
+    }
+
+    Err(ConsumerError::JetstreamEnded)
+}

spacedust/src/delay.rs

···
+use crate::removable_delay_queue;
+use crate::LinkEvent;
+use tokio_util::sync::CancellationToken;
+use tokio::sync::broadcast;
+use crate::error::DelayError;
+
+pub async fn to_broadcast(
+    source: removable_delay_queue::Output<(String, usize), LinkEvent>,
+    dest: broadcast::Sender<LinkEvent>,
+    shutdown: CancellationToken,
+) -> Result<(), DelayError> {
+    loop {
+        tokio::select! {
+            ev = source.next() => match ev {
+                Some(event) => {
+                    let _ = dest.send(event); // only errors of there are no listeners, but that's normal
+                },
+                None => return Err(DelayError::DelayEnded),
+            },
+            _ = shutdown.cancelled() => return Ok(()),
+        }
+    }
+}

spacedust/src/error.rs

···
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum MainTaskError {
+    #[error(transparent)]
+    ConsumerTaskError(#[from] ConsumerError),
+    #[error(transparent)]
+    ServerTaskError(#[from] ServerError),
+    #[error(transparent)]
+    DelayTaskError(#[from] DelayError),
+}
+
+#[derive(Debug, Error)]
+pub enum ConsumerError {
+    #[error(transparent)]
+    JetstreamConnectionError(#[from] jetstream::error::ConnectionError),
+    #[error(transparent)]
+    JetstreamConfigValidationError(#[from] jetstream::error::ConfigValidationError),
+    #[error("jetstream ended")]
+    JetstreamEnded,
+    #[error("delay queue output dropped")]
+    DelayQueueOutputDropped,
+}
+
+#[derive(Debug, Error)]
+pub enum DelayError {
+    #[error("delay ended")]
+    DelayEnded,
+}
+
+#[derive(Debug, Error)]
+pub enum ServerError {
+    #[error("failed to configure server logger: {0}")]
+    ConfigLogError(std::io::Error),
+    #[error("failed to render json for openapi: {0}")]
+    OpenApiJsonFail(serde_json::Error),
+    #[error(transparent)]
+    FailedToBuildServer(#[from] dropshot::BuildError),
+    #[error("server exited: {0}")]
+    ServerExited(String),
+    #[error("server closed badly: {0}")]
+    BadClose(String),
+}

spacedust/src/lib.rs

···
+pub mod consumer;
+pub mod delay;
+pub mod error;
+pub mod server;
+pub mod subscriber;
+pub mod removable_delay_queue;
+
+use serde::Serialize;
+
+#[derive(Debug, Clone)]
+pub struct LinkEvent {
+    collection: String,
+    path: String,
+    origin: String,
+    target: String,
+    rev: String,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all="snake_case")]
+pub struct ClientEvent {
+    kind: String, // "link"
+    origin: String, // "live", "replay", "backfill"
+    link: ClientLinkEvent,
+}
+
+#[derive(Debug, Serialize)]
+struct ClientLinkEvent {
+    operation: String, // "create", "delete" (prob no update, though maybe for rev?)
+    source: String,
+    source_record: String,
+    source_rev: String,
+    subject: String,
+    // TODO: include the record too? would save clients a level of hydration
+}
+
+impl From<LinkEvent> for ClientLinkEvent {
+    fn from(link: LinkEvent) -> Self {
+        let undotted = link.path.strip_prefix('.').unwrap_or_else(|| {
+            eprintln!("link path did not have expected '.' prefix: {}", link.path);
+            ""
+        });
+        Self {
+            operation: "create".to_string(),
+            source: format!("{}:{undotted}", link.collection),
+            source_record: link.origin,
+            source_rev: link.rev,
+            subject: link.target,
+        }
+    }
+}

spacedust/src/main.rs

···
+use spacedust::error::MainTaskError;
+use spacedust::consumer;
+use spacedust::server;
+use spacedust::delay;
+use spacedust::removable_delay_queue::removable_delay_queue;
+
+use clap::Parser;
+use metrics_exporter_prometheus::PrometheusBuilder;
+use tokio::sync::broadcast;
+use tokio_util::sync::CancellationToken;
+use std::time::Duration;
+
+/// Aggregate links in the at-mosphere
+#[derive(Parser, Debug, Clone)]
+#[command(version, about, long_about = None)]
+struct Args {
+    /// Jetstream server to connect to (exclusive with --fixture). Provide either a wss:// URL, or a shorhand value:
+    /// 'us-east-1', 'us-east-2', 'us-west-1', or 'us-west-2'
+    #[arg(long)]
+    jetstream: String,
+    /// don't request zstd-compressed jetstream events
+    ///
+    /// reduces CPU at the expense of more ingress bandwidth
+    #[arg(long, action)]
+    jetstream_no_zstd: bool,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), String> {
+    env_logger::init();
+
+    // tokio broadcast keeps a single main output queue for all subscribers.
+    // each subscriber clones off a copy of an individual value for each recv.
+    // since there's no large per-client buffer, we can make this one kind of
+    // big and accommodate more slow/bursty clients.
+    //
+    // in fact, we *could* even keep lagging clients alive, inserting lag-
+    // indicating messages to their output.... but for now we'll drop them to
+    // avoid accumulating zombies.
+    //
+    // events on the channel are individual links as they are discovered. a link
+    // contains a source and a target. the target is an at-uri, so it's up to
+    // ~1KB max; source is a collection + link path, which can be more but in
+    // practice the whole link rarely approaches 1KB total.
+    //
+    // TODO: determine if a pathological case could blow this up (eg 1MB link
+    // paths + slow subscriber -> 16GiB queue)
+    let (b, _) = broadcast::channel(16_384);
+    let consumer_sender = b.clone();
+    let (d, _) = broadcast::channel(16_384);
+    let consumer_delayed_sender = d.clone();
+
+    let delay = Duration::from_secs(21);
+    let (delay_queue_sender, delay_queue_receiver) = removable_delay_queue(delay);
+
+    let shutdown = CancellationToken::new();
+
+    let ctrlc_shutdown = shutdown.clone();
+    ctrlc::set_handler(move || ctrlc_shutdown.cancel()).expect("failed to set ctrl-c handler");
+
+    let args = Args::parse();
+
+    if let Err(e) = install_metrics_server() {
+        log::error!("failed to install metrics server: {e:?}");
+    };
+
+    let mut tasks: tokio::task::JoinSet<Result<(), MainTaskError>> = tokio::task::JoinSet::new();
+
+    let server_shutdown = shutdown.clone();
+    tasks.spawn(async move {
+        server::serve(b, d, server_shutdown).await?;
+        Ok(())
+    });
+
+    let consumer_shutdown = shutdown.clone();
+    tasks.spawn(async move {
+        consumer::consume(
+            consumer_sender,
+            delay_queue_sender,
+            args.jetstream,
+            None,
+            args.jetstream_no_zstd,
+            consumer_shutdown
+        )
+            .await?;
+        Ok(())
+    });
+
+    let delay_shutdown = shutdown.clone();
+    tasks.spawn(async move {
+        delay::to_broadcast(delay_queue_receiver, consumer_delayed_sender, delay_shutdown).await?;
+        Ok(())
+    });
+
+    tokio::select! {
+        _ = shutdown.cancelled() => log::warn!("shutdown requested"),
+        Some(r) = tasks.join_next() => {
+            log::warn!("a task exited, shutting down: {r:?}");
+            shutdown.cancel();
+        }
+    }
+
+    tokio::select! {
+        _ = async {
+            while let Some(completed) = tasks.join_next().await {
+                log::info!("shutdown: task completed: {completed:?}");
+            }
+        } => {},
+        _ = tokio::time::sleep(std::time::Duration::from_secs(3)) => {
+            log::info!("shutdown: not all tasks completed on time. aborting...");
+            tasks.shutdown().await;
+        },
+    }
+
+    log::info!("bye!");
+
+    Ok(())
+}
+
+fn install_metrics_server() -> Result<(), metrics_exporter_prometheus::BuildError> {
+    log::info!("installing metrics server...");
+    let host = [0, 0, 0, 0];
+    let port = 8765;
+    PrometheusBuilder::new()
+        .set_quantiles(&[0.5, 0.9, 0.99, 1.0])?
+        .set_bucket_duration(std::time::Duration::from_secs(300))?
+        .set_bucket_count(std::num::NonZero::new(12).unwrap()) // count * duration = 60 mins. stuff doesn't happen that fast here.
+        .set_enable_unit_suffix(false) // this seemed buggy for constellation (sometimes wouldn't engage)
+        .with_http_listener((host, port))
+        .install()?;
+    log::info!(
+        "metrics server installed! listening on http://{}.{}.{}.{}:{port}",
+        host[0],
+        host[1],
+        host[2],
+        host[3]
+    );
+    Ok(())
+}

spacedust/src/removable_delay_queue.rs

···
+use std::ops::RangeBounds;
+use std::collections::{BTreeMap, VecDeque};
+use std::time::{Duration, Instant};
+use tokio::sync::Mutex;
+use std::sync::Arc;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum EnqueueError<T> {
+    #[error("queue ouput dropped")]
+    OutputDropped(T),
+}
+
+pub trait Key: Eq + Ord + Clone {}
+impl<T: Eq + Ord + Clone> Key for T {}
+
+#[derive(Debug)]
+struct Queue<K: Key, T> {
+    queue: VecDeque<(Instant, K)>,
+    items: BTreeMap<K, T>
+}
+
+pub struct Input<K: Key, T> {
+    q: Arc<Mutex<Queue<K, T>>>,
+}
+
+impl<K: Key, T> Input<K, T> {
+    /// if a key is already present, its previous item will be overwritten and
+    /// its delay time will be reset for the new item.
+    ///
+    /// errors if the remover has been dropped
+    pub async fn enqueue(&self, key: K, item: T) -> Result<(), EnqueueError<T>> {
+        if Arc::strong_count(&self.q) == 1 {
+            return Err(EnqueueError::OutputDropped(item));
+        }
+        // TODO: try to push out an old element first
+        // for now we just hope there's a listener
+        let now = Instant::now();
+        let mut q = self.q.lock().await;
+        q.queue.push_back((now, key.clone()));
+        q.items.insert(key, item);
+        Ok(())
+    }
+    /// remove an item from the queue, by key
+    ///
+    /// the item itself is removed, but the key will remain in the queue -- it
+    /// will simply be skipped over when a new output item is requested. this
+    /// keeps the removal cheap (=btreemap remove), for a bit of space overhead
+    pub async fn remove_range(&self, range: impl RangeBounds<K>) {
+        let n = {
+            let mut q = self.q.lock().await;
+            let keys = q.items.range(range).map(|(k, _)| k).cloned().collect::<Vec<_>>();
+            for k in &keys {
+                q.items.remove(k);
+            }
+            keys.len()
+        };
+        if n == 0 {
+            metrics::counter!("delay_queue_remove_not_found").increment(1);
+        } else {
+            metrics::counter!("delay_queue_remove_total_records").increment(1);
+            metrics::counter!("delay_queue_remove_total_links").increment(n as u64);
+        }
+    }
+}
+
+pub struct Output<K: Key, T> {
+    delay: Duration,
+    q: Arc<Mutex<Queue<K, T>>>,
+}
+
+impl<K: Key, T> Output<K, T> {
+    pub async fn next(&self) -> Option<T> {
+        let get = || async {
+            let mut q = self.q.lock().await;
+            metrics::gauge!("delay_queue_queue_len").set(q.queue.len() as f64);
+            metrics::gauge!("delay_queue_queue_capacity").set(q.queue.capacity() as f64);
+            while let Some((t, k)) = q.queue.pop_front() {
+                // skip over queued keys that were removed from items
+                if let Some(item) = q.items.remove(&k) {
+                    return Some((t, item));
+                }
+            }
+            None
+        };
+        loop {
+            if let Some((t, item)) = get().await {
+                let now = Instant::now();
+                let expected_release = t + self.delay;
+                if expected_release.saturating_duration_since(now) > Duration::from_millis(1) {
+                    tokio::time::sleep_until(expected_release.into()).await;
+                    metrics::counter!("delay_queue_emit_total", "early" => "yes").increment(1);
+                    metrics::histogram!("delay_queue_emit_overshoot").record(0);
+                } else {
+                    let overshoot = now.saturating_duration_since(expected_release);
+                    metrics::counter!("delay_queue_emit_total", "early" => "no").increment(1);
+                    metrics::histogram!("delay_queue_emit_overshoot").record(overshoot.as_secs_f64());
+                }
+                return Some(item)
+            } else if Arc::strong_count(&self.q) == 1 {
+                return None;
+            }
+            // the queue is *empty*, so we need to wait at least as long as the current delay
+            tokio::time::sleep(self.delay).await;
+            metrics::counter!("delay_queue_entirely_empty_total").increment(1);
+        };
+    }
+}
+
+pub fn removable_delay_queue<K: Key, T>(
+    delay: Duration,
+) -> (Input<K, T>, Output<K, T>) {
+    let q: Arc<Mutex<Queue<K, T>>> = Arc::new(Mutex::new(Queue {
+        queue: VecDeque::new(),
+        items: BTreeMap::new(),
+    }));
+
+    let input = Input::<K, T> { q: q.clone() };
+    let output = Output::<K, T> { q, delay };
+    (input, output)
+}

spacedust/src/server.rs

···
+use crate::error::ServerError;
+use crate::subscriber::Subscriber;
+use metrics::{histogram, counter};
+use std::sync::Arc;
+use crate::LinkEvent;
+use http::{
+    header::{ORIGIN, USER_AGENT},
+    Response, StatusCode,
+};
+use dropshot::{
+    Body,
+    ApiDescription, ConfigDropshot, ConfigLogging, ConfigLoggingLevel, Query, RequestContext,
+    ServerBuilder, WebsocketConnection, channel, endpoint, HttpResponse,
+    ApiEndpointBodyContentType, ExtractorMetadata, HttpError, ServerContext,
+    SharedExtractor,
+};
+
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use tokio::sync::broadcast;
+use tokio::time::Instant;
+use tokio_tungstenite::tungstenite::protocol::Role;
+use tokio_util::sync::CancellationToken;
+use async_trait::async_trait;
+use std::collections::HashSet;
+
+const INDEX_HTML: &str = include_str!("../static/index.html");
+const FAVICON: &[u8] = include_bytes!("../static/favicon.ico");
+
+pub async fn serve(
+    b: broadcast::Sender<LinkEvent>,
+    d: broadcast::Sender<LinkEvent>,
+    shutdown: CancellationToken
+) -> Result<(), ServerError> {
+    let config_logging = ConfigLogging::StderrTerminal {
+        level: ConfigLoggingLevel::Info,
+    };
+
+    let log = config_logging
+        .to_logger("example-basic")
+        .map_err(ServerError::ConfigLogError)?;
+
+    let mut api = ApiDescription::new();
+    api.register(index).unwrap();
+    api.register(favicon).unwrap();
+    api.register(openapi).unwrap();
+    api.register(subscribe).unwrap();
+
+    // TODO: put spec in a once cell / lazy lock thing?
+    let spec = Arc::new(
+        api.openapi(
+            "Spacedust",
+            env!("CARGO_PKG_VERSION")
+                .parse()
+                .inspect_err(|e| {
+                    eprintln!("failed to parse cargo package version for openapi: {e:?}")
+                })
+                .unwrap_or(semver::Version::new(0, 0, 1)),
+        )
+        .description("A configurable ATProto notifications firehose.")
+        .contact_name("part of @microcosm.blue")
+        .contact_url("https://microcosm.blue")
+        .json()
+        .map_err(ServerError::OpenApiJsonFail)?,
+    );
+
+    let sub_shutdown = shutdown.clone();
+    let ctx = Context { spec, b, d, shutdown: sub_shutdown };
+
+    let server = ServerBuilder::new(api, ctx, log)
+        .config(ConfigDropshot {
+            bind_address: "0.0.0.0:9998".parse().unwrap(),
+            ..Default::default()
+        })
+        .start()?;
+
+    tokio::select! {
+        s = server.wait_for_shutdown() => {
+            s.map_err(ServerError::ServerExited)?;
+            log::info!("server shut down normally.");
+        },
+        _ = shutdown.cancelled() => {
+            log::info!("shutting down: closing server");
+            server.close().await.map_err(ServerError::BadClose)?;
+        },
+    }
+    Ok(())
+}
+
+#[derive(Debug, Clone)]
+struct Context {
+    pub spec: Arc<serde_json::Value>,
+    pub b: broadcast::Sender<LinkEvent>,
+    pub d: broadcast::Sender<LinkEvent>,
+    pub shutdown: CancellationToken,
+}
+
+async fn instrument_handler<T, H, R>(ctx: &RequestContext<T>, handler: H) -> Result<R, HttpError>
+where
+    R: HttpResponse,
+    H: Future<Output = Result<R, HttpError>>,
+    T: ServerContext,
+{
+    let start = Instant::now();
+    let result = handler.await;
+    let latency = start.elapsed();
+    let status_code = match &result {
+        Ok(response) => response.status_code(),
+        Err(e) => e.status_code.as_status(),
+    }
+    .as_str() // just the number (.to_string()'s Display does eg `200 OK`)
+    .to_string();
+    let endpoint = ctx.endpoint.operation_id.clone();
+    let headers = ctx.request.headers();
+    let origin = headers
+        .get(ORIGIN)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or("")
+        .to_string();
+    let ua = headers
+        .get(USER_AGENT)
+        .and_then(|v| v.to_str().ok())
+        .map(|ua| {
+            if ua.starts_with("Mozilla/5.0 ") {
+                "browser"
+            } else {
+                ua
+            }
+        })
+        .unwrap_or("")
+        .to_string();
+    counter!("server_requests_total",
+        "endpoint" => endpoint.clone(),
+        "origin" => origin,
+        "ua" => ua,
+        "status_code" => status_code,
+    )
+    .increment(1);
+    histogram!("server_handler_latency", "endpoint" => endpoint).record(latency.as_micros() as f64);
+    result
+}
+
+use dropshot::{HttpResponseHeaders, HttpResponseOk};
+
+pub type OkCorsResponse<T> = Result<HttpResponseHeaders<HttpResponseOk<T>>, HttpError>;
+
+/// Helper for constructing Ok responses: return OkCors(T).into()
+/// (not happy with this yet)
+pub struct OkCors<T: Serialize + JsonSchema + Send + Sync>(pub T);
+
+impl<T> From<OkCors<T>> for OkCorsResponse<T>
+where
+    T: Serialize + JsonSchema + Send + Sync,
+{
+    fn from(ok: OkCors<T>) -> OkCorsResponse<T> {
+        let mut res = HttpResponseHeaders::new_unnamed(HttpResponseOk(ok.0));
+        res.headers_mut()
+            .insert("access-control-allow-origin", "*".parse().unwrap());
+        Ok(res)
+    }
+}
+
+// TODO: cors for HttpError
+
+
+/// Serve index page as html
+#[endpoint {
+    method = GET,
+    path = "/",
+    /*
+     * not useful to have this in openapi
+     */
+    unpublished = true,
+}]
+async fn index(ctx: RequestContext<Context>) -> Result<Response<Body>, HttpError> {
+    instrument_handler(&ctx, async {
+        Ok(Response::builder()
+            .status(StatusCode::OK)
+            .header(http::header::CONTENT_TYPE, "text/html")
+            .body(INDEX_HTML.into())?)
+    })
+    .await
+}
+
+/// Serve index page as html
+#[endpoint {
+    method = GET,
+    path = "/favicon.ico",
+    /*
+     * not useful to have this in openapi
+     */
+    unpublished = true,
+}]
+async fn favicon(ctx: RequestContext<Context>) -> Result<Response<Body>, HttpError> {
+    instrument_handler(&ctx, async {
+        Ok(Response::builder()
+            .status(StatusCode::OK)
+            .header(http::header::CONTENT_TYPE, "image/x-icon")
+            .body(FAVICON.to_vec().into())?)
+    })
+    .await
+}
+
+/// Meta: get the openapi spec for this api
+#[endpoint {
+    method = GET,
+    path = "/openapi",
+    /*
+     * not useful to have this in openapi
+     */
+    unpublished = true,
+}]
+async fn openapi(ctx: RequestContext<Context>) -> OkCorsResponse<serde_json::Value> {
+    instrument_handler(&ctx, async {
+        let spec = (*ctx.context().spec).clone();
+        OkCors(spec).into()
+    })
+    .await
+}
+
+/// The real type that gets deserialized
+#[derive(Debug, Deserialize, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct MultiSubscribeQuery {
+    #[serde(default)]
+    pub wanted_subjects: HashSet<String>,
+    #[serde(default)]
+    pub wanted_subject_dids: HashSet<String>,
+    #[serde(default)]
+    pub wanted_sources: HashSet<String>,
+}
+/// The fake corresponding type for docs that dropshot won't freak out about a
+/// vec for
+#[derive(Deserialize, JsonSchema)]
+#[allow(dead_code)]
+#[serde(rename_all = "camelCase")]
+struct MultiSubscribeQueryForDocs {
+    /// One or more at-uris to receive links about
+    ///
+    /// The at-uri must be url-encoded
+    ///
+    /// Pass this parameter multiple times to specify multiple collections, like
+    /// `wantedSubjects=[...]&wantedSubjects=[...]`
+    pub wanted_subjects: String,
+    /// One or more DIDs to receive links about
+    ///
+    /// Pass this parameter multiple times to specify multiple collections
+    pub wanted_subject_dids: String,
+    /// One or more link sources to receive links about
+    ///
+    /// TODO: docs about link sources
+    ///
+    /// eg, a bluesky like's link source: `app.bsky.feed.like:subject.uri`
+    ///
+    /// Pass this parameter multiple times to specify multiple sources
+    pub wanted_sources: String,
+}
+
+// The `SharedExtractor` implementation for Query<QueryType> describes how to
+// construct an instance of `Query<QueryType>` from an HTTP request: namely, by
+// parsing the query string to an instance of `QueryType`.
+#[async_trait]
+impl SharedExtractor for MultiSubscribeQuery {
+    async fn from_request<Context: ServerContext>(
+        ctx: &RequestContext<Context>,
+    ) -> Result<MultiSubscribeQuery, HttpError> {
+        let raw_query = ctx.request.uri().query().unwrap_or("");
+        let q = serde_qs::from_str(raw_query).map_err(|e| {
+            HttpError::for_bad_request(None, format!("unable to parse query string: {}", e))
+        })?;
+        Ok(q)
+    }
+
+    fn metadata(body_content_type: ApiEndpointBodyContentType) -> ExtractorMetadata {
+        // HACK: query type switcheroo: passing MultiSubscribeQuery to
+        // `metadata` would "helpfully" panic because dropshot believes we can
+        // only have scalar types in a  query.
+        //
+        // so instead we have a fake second type whose only job is to look the
+        // same as MultiSubscribeQuery exept that it has `String` instead of
+        // `Vec<String>`, which dropshot will accept, and generate ~close-enough
+        // docs for.
+        <Query<MultiSubscribeQueryForDocs> as SharedExtractor>::metadata(body_content_type)
+    }
+}
+
+#[derive(Deserialize, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+struct ScalarSubscribeQuery {
+    #[serde(default)]
+    pub instant: bool,
+}
+
+#[channel {
+    protocol = WEBSOCKETS,
+    path = "/subscribe",
+}]
+async fn subscribe(
+    reqctx: RequestContext<Context>,
+    query: MultiSubscribeQuery,
+    scalar_query: Query<ScalarSubscribeQuery>,
+    upgraded: WebsocketConnection,
+) -> dropshot::WebsocketChannelResult {
+    let ws = tokio_tungstenite::WebSocketStream::from_raw_socket(
+        upgraded.into_inner(),
+        Role::Server,
+        None,
+    )
+    .await;
+
+    let Context { b, d, shutdown, .. } = reqctx.context();
+    let sub_token = shutdown.child_token();
+
+    let q = scalar_query.into_inner();
+    let subscription = if q.instant { b } else { d }.subscribe();
+    log::info!("starting subscriber with broadcast: instant={}", q.instant);
+
+    Subscriber::new(query, sub_token)
+        .start(ws, subscription)
+        .await
+        .map_err(|e| format!("boo: {e:?}"))?;
+
+    Ok(())
+}

spacedust/src/subscriber.rs

···
+use tokio::time::interval;
+use std::time::Duration;
+use futures::StreamExt;
+use crate::ClientEvent;
+use crate::LinkEvent;
+use crate::server::MultiSubscribeQuery;
+use futures::SinkExt;
+use std::error::Error;
+use tokio::sync::broadcast::{self, error::RecvError};
+use tokio_tungstenite::{WebSocketStream, tungstenite::Message};
+use tokio_util::sync::CancellationToken;
+use dropshot::WebsocketConnectionRaw;
+
+const PING_PERIOD: Duration = Duration::from_secs(30);
+
+pub struct Subscriber {
+    query: MultiSubscribeQuery,
+    shutdown: CancellationToken,
+}
+
+impl Subscriber {
+    pub fn new(
+        query: MultiSubscribeQuery,
+        shutdown: CancellationToken,
+    ) -> Self {
+        Self { query, shutdown }
+    }
+
+    pub async fn start(
+        self,
+        ws: WebSocketStream<WebsocketConnectionRaw>,
+        mut receiver: broadcast::Receiver<LinkEvent>
+    ) -> Result<(), Box<dyn Error>> {
+        let mut ping_state = None;
+        let (mut ws_sender, mut ws_receiver) = ws.split();
+        let mut ping_interval = interval(PING_PERIOD);
+        let _guard = self.shutdown.clone().drop_guard();
+
+        // TODO: do we need to timeout ws sends??
+
+        metrics::counter!("subscribers_connected_total").increment(1);
+        metrics::gauge!("subscribers_connected").increment(1);
+
+        loop {
+            tokio::select! {
+                l = receiver.recv() => match l {
+                    Ok(link) => if let Some(message) = self.filter(link) {
+                        if let Err(e) = ws_sender.send(message).await {
+                            log::warn!("failed to send link, dropping subscriber: {e:?}");
+                            break;
+                        }
+                    },
+                    Err(RecvError::Closed) => self.shutdown.cancel(),
+                    Err(RecvError::Lagged(n)) => {
+                        log::warn!("dropping lagging subscriber (missed {n} messages already)");
+                        self.shutdown.cancel();
+                    }
+                },
+                cm = ws_receiver.next() => match cm {
+                    Some(Ok(Message::Ping(state))) => {
+                        if let Err(e) = ws_sender.send(Message::Pong(state)).await {
+                            log::error!("failed to reply pong to subscriber: {e:?}");
+                            break;
+                        }
+                    }
+                    Some(Ok(Message::Pong(state))) => {
+                        if let Some(expected_state) = ping_state {
+                            if *state == expected_state {
+                                ping_state = None; // good
+                            } else {
+                                log::error!("subscriber returned a pong with the wrong state, dropping");
+                                self.shutdown.cancel();
+                            }
+                        } else {
+                            log::error!("subscriber sent a pong when none was expected");
+                            self.shutdown.cancel();
+                        }
+                    }
+                    Some(Ok(m)) => log::trace!("subscriber sent an unexpected message: {m:?}"),
+                    Some(Err(e)) => {
+                        log::error!("failed to receive subscriber message: {e:?}");
+                        break;
+                    }
+                    None => {
+                        log::trace!("end of subscriber messages. bye!");
+                        break;
+                    }
+                },
+                _ = ping_interval.tick() => {
+                    if ping_state.is_some() {
+                        log::warn!("did not recieve pong within {PING_PERIOD:?}, dropping subscriber");
+                        self.shutdown.cancel();
+                    } else {
+                        let new_state: [u8; 8] = rand::random();
+                        let ping = new_state.to_vec().into();
+                        ping_state = Some(new_state);
+                        if let Err(e) = ws_sender.send(Message::Ping(ping)).await {
+                            log::error!("failed to send ping to subscriber, dropping: {e:?}");
+                            self.shutdown.cancel();
+                        }
+                    }
+                }
+                _ = self.shutdown.cancelled() => {
+                    log::info!("subscriber shutdown requested, bye!");
+                    if let Err(e) = ws_sender.close().await {
+                        log::warn!("failed to close subscriber: {e:?}");
+                    }
+                    break;
+                },
+            }
+        }
+        log::trace!("end of subscriber. bye!");
+        metrics::gauge!("subscribers_connected").decrement(1);
+        Ok(())
+    }
+
+    fn filter(
+        &self,
+        link: LinkEvent,
+        // mut sender: impl Sink<Message> + Unpin
+    ) -> Option<Message> {
+        let query = &self.query;
+
+        // subject + subject DIDs are logical OR
+        let target_did = if link.target.starts_with("did:") {
+            link.target.clone()
+        } else {
+            let rest = link.target.strip_prefix("at://")?;
+            if let Some((did, _)) = rest.split_once("/") {
+                did
+            } else {
+                rest
+            }.to_string()
+        };
+        if !(query.wanted_subjects.contains(&link.target) || query.wanted_subject_dids.contains(&target_did) || query.wanted_subjects.is_empty() && query.wanted_subject_dids.is_empty()) {
+            // wowwww ^^ fix that
+            return None
+        }
+
+        // subjects together with sources are logical AND
+
+        if !query.wanted_sources.is_empty() {
+            let undotted = link.path.strip_prefix('.').unwrap_or_else(|| {
+                eprintln!("link path did not have expected '.' prefix: {}", link.path);
+                ""
+            });
+            let source = format!("{}:{undotted}", link.collection);
+            if !query.wanted_sources.contains(&source) {
+                return None
+            }
+        }
+
+        let ev = ClientEvent {
+            kind: "link".to_string(),
+            origin: "live".to_string(),
+            link: link.into(),
+        };
+
+        let json = serde_json::to_string(&ev).unwrap();
+
+        Some(Message::Text(json.into()))
+    }
+}

spacedust/static/index.html

···
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8" />
+    <title>Spacedust documentation</title>
+    <meta name="viewport" content="width=device-width, initial-scale=1" />
+    <meta name="description" content="API Documentation for Spacedust, a configurable ATProto notifications firehose" />
+    <style>
+      .custom-header {
+        height: 42px;
+        background-color: #221828;
+        box-shadow: inset 0 -1px 0 var(--scalar-border-color);
+        color: var(--scalar-color-1);
+        font-size: var(--scalar-font-size-3);
+        font-family: 'Iowan Old Style', 'Palatino Linotype', 'URW Palladio L', P052, serif;
+        padding: 0 18px;
+        justify-content: space-between;
+      }
+      .custom-header,
+      .custom-header nav {
+        display: flex;
+        align-items: center;
+        gap: 18px;
+      }
+      .custom-header a:hover {
+        color: var(--scalar-color-2);
+      }
+    </style>
+  </head>
+  <body>
+    <header class="custom-header scalar-app">
+      <p>
+        TODO: pdsls jetstream link
+        <a href="https://ufos.microcosm.blue">Launch 🛸 UFOs app</a>: Explore lexicons
+      </p>
+      <nav>
+        <b>a <a href="https://microcosm.blue">microcosm</a> project</b>
+        <a href="https://bsky.app/profile/microcosm.blue">@microcosm.blue</a>
+        <a href="https://github.com/at-microcosm">github</a>
+      </nav>
+    </header>
+
+    <script id="api-reference" type="application/json" data-url="/openapi""></script>
+
+    <script>
+      var configuration = {
+        theme: 'purple',
+      }
+      document.getElementById('api-reference').dataset.configuration = JSON.stringify(configuration)
+    </script>
+
+    <script src="https://cdn.jsdelivr.net/npm/@scalar/api-reference"></script>
+  </body>
+</html>

ufos/src/index_html.rs

···
 <html lang="en">
   <head>
     <meta charset="utf-8" />
-    <title>UFOs API Documentation</title>
+    <title>UFOs API documentation</title>
     <meta name="viewport" content="width=device-width, initial-scale=1" />
     <meta name="description" content="API Documentation for UFOs: Samples and stats for all atproto lexicons." />
     <style>

ufos/src/storage_fjall.rs

···
 ///
 /// new data format, roughly:
 ///
-/// Partion: 'global'
+/// Partition: 'global'
 ///
 ///  - Global sequence counter (is the jetstream cursor -- monotonic with many gaps)
 ///      - key: "js_cursor" (literal)