diff --git a/Cargo.lock b/Cargo.lock index 336335b..b7a574a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -76,7 +76,7 @@ checksum = "d1eb7c4fcde1858a6796c18a729b661346d38e05a207e2d9028bce822fc20283" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -171,6 +171,18 @@ version = "1.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457" +[[package]] +name = "arrayref" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76a2e8124351fda1ef8aaaa3bbd7ebbcb486bbcd4225aca0aa0d84bb2db8fecb" + +[[package]] +name = "arrayvec" +version = "0.7.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" + [[package]] name = "async-compression" version = "0.4.27" @@ -195,7 +207,7 @@ checksum = "e539d3fca749fcee5236ab05e93a52867dd549cc157c8cb7f99595f3cedffdb5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -204,6 +216,29 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "aws-lc-rs" +version = "1.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.45.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", + "pkg-config", +] + [[package]] name = "axum" version = "0.8.4" @@ -331,6 +366,20 @@ dependencies = [ "serde", ] +[[package]] +name = "blake3" +version = "1.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76ae7bad254120e9e4c63bafc385310756f90c484eac0e36b8317cf09cb92a77" +dependencies = [ + "arrayref", + "arrayvec", + "cc", + "cfg-if", + "constant_time_eq", + "cpufeatures 0.3.0", +] + [[package]] name = "block-buffer" version = "0.10.4" @@ -384,10 +433,11 @@ checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a" [[package]] name = "cc" -version = "1.2.29" +version = "1.4.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c1599538de2394445747c8cf7935946e3cc27e9625f889d979bfb2aaf569362" +checksum = "54413ede23c2daf518f35156dfde027feb2374004d63bd497f983c8db9c0e313" dependencies = [ + "find-msvc-tools", "jobserver", "libc", "shlex", @@ -457,7 +507,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -484,7 +534,7 @@ dependencies = [ "lz4_flex", "quanta", "replace_with", - "sealed", + "sealed 0.6.0", "serde", "static_assertions", "thiserror 1.0.69", @@ -502,7 +552,16 @@ dependencies = [ "proc-macro2", "quote", "serde_derive_internals", - "syn", + "syn 2.0.104", +] + +[[package]] +name = "cmake" +version = "0.1.58" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" +dependencies = [ + "cc", ] [[package]] @@ -511,6 +570,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" +[[package]] +name = "constant_time_eq" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" + [[package]] name = "core-foundation-sys" version = "0.8.7" @@ -526,6 +591,15 @@ dependencies = [ "libc", ] +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + [[package]] name = "crc32fast" version = "1.5.0" @@ -535,6 +609,15 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossbeam-channel" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-deque" version = "0.8.6" @@ -547,9 +630,9 @@ dependencies = [ [[package]] name = "crossbeam-epoch" -version = "0.9.18" +version = "0.9.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" dependencies = [ "crossbeam-utils", ] @@ -630,9 +713,15 @@ checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] +[[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" @@ -654,7 +743,7 @@ dependencies = [ "once_cell", "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -663,6 +752,12 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "find-msvc-tools" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef25905e51abafe4dcea6c15fec58c57b601cdbd0ee53d22ea1d3016c587d39b" + [[package]] name = "flate2" version = "1.1.2" @@ -685,6 +780,12 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + [[package]] name = "form_urlencoded" version = "1.2.1" @@ -694,6 +795,12 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "futures" version = "0.3.31" @@ -750,7 +857,7 @@ checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -838,7 +945,16 @@ version = "0.15.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5971ac85611da7067dbfcabef3c70ebb5606018acd9e2a3903a0da507521e0d5" dependencies = [ - "foldhash", + "foldhash 0.1.5", +] + +[[package]] +name = "hashbrown" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +dependencies = [ + "foldhash 0.2.0", ] [[package]] @@ -946,12 +1062,12 @@ dependencies = [ "http", "hyper", "hyper-util", - "rustls 0.23.29", + "rustls", "rustls-pki-types", "tokio", - "tokio-rustls 0.26.2", + "tokio-rustls", "tower-service", - "webpki-roots 1.0.1", + "webpki-roots", ] [[package]] @@ -1292,42 +1408,44 @@ dependencies = [ [[package]] name = "metrics-prometheus" -version = "0.8.0" +version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba3115d0e859470455e1587af70aed7935163339f959f01d6cbd245605fe7e67" +checksum = "5ba1c653d1714b0057c0581e4ef8d50600470132d311fbed69cdcc1804e69788" dependencies = [ "arc-swap", "metrics", - "metrics-util 0.18.0", + "metrics-util 0.20.1", "prometheus", - "sealed", + "sealed 0.7.0", "smallvec", - "thiserror 1.0.69", ] [[package]] name = "metrics-util" -version = "0.18.0" +version = "0.19.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "15b482df36c13dd1869d73d14d28cd4855fbd6cfc32294bee109908a9f4a4ed7" +checksum = "b8496cc523d1f94c1385dd8f0f0c2c480b2b8aeccb5b7e4485ad6365523ae376" dependencies = [ "crossbeam-epoch", "crossbeam-utils", "hashbrown 0.15.4", "metrics", + "quanta", + "rand 0.9.1", + "rand_xoshiro", + "sketches-ddsketch", ] [[package]] name = "metrics-util" -version = "0.19.1" +version = "0.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8496cc523d1f94c1385dd8f0f0c2c480b2b8aeccb5b7e4485ad6365523ae376" +checksum = "cdfb1365fea27e6dd9dc1dbc19f570198bc86914533ad639dae939635f096be4" dependencies = [ "crossbeam-epoch", "crossbeam-utils", - "hashbrown 0.15.4", + "hashbrown 0.16.1", "metrics", - "quanta", "rand 0.9.1", "rand_xoshiro", "sketches-ddsketch", @@ -1378,6 +1496,23 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "moka" +version = "0.12.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4293f18e7567a1caf3c584855554377025c65e0aa445344d04171f5ad63d19b9" +dependencies = [ + "crossbeam-channel", + "crossbeam-epoch", + "crossbeam-utils", + "equivalent", + "parking_lot", + "portable-atomic", + "smallvec", + "tagptr", + "uuid", +] + [[package]] name = "nu-ansi-term" version = "0.46.0" @@ -1476,7 +1611,7 @@ checksum = "6e918e4ff8c4549eb882f14b3a4bc8c8bc93de829416eacf579f1207a8fbf861" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -1542,9 +1677,9 @@ dependencies = [ [[package]] name = "prometheus" -version = "0.13.4" +version = "0.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d33c28a30771f7f96db69893f78b857f7450d7e0237e9c8fc6427a81bae7ed1" +checksum = "3ca5326d8d0b950a9acd87e6a3f94745394f62e4dae1b1ee22b2bc0c394af43a" dependencies = [ "cfg-if", "fnv", @@ -1552,14 +1687,28 @@ dependencies = [ "memchr", "parking_lot", "protobuf", - "thiserror 1.0.69", + "thiserror 2.0.12", ] [[package]] name = "protobuf" -version = "2.28.0" +version = "3.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d65a1d4ddae7d8b5de68153b48f6aa3bba8cb002b243dbdbc55a5afbc98f99f4" +dependencies = [ + "once_cell", + "protobuf-support", + "thiserror 1.0.69", +] + +[[package]] +name = "protobuf-support" +version = "3.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "106dd99e98437432fed6519dedecfade6a06a73bb7b2a1e019fdd2bee5778d94" +checksum = "3e36c2f31e0a47f9280fb347ef5e461ffcd2c52dd520d8e216b52f93b0b0d7d6" +dependencies = [ + "thiserror 1.0.69", +] [[package]] name = "quanta" @@ -1588,7 +1737,7 @@ dependencies = [ "quinn-proto", "quinn-udp", "rustc-hash", - "rustls 0.23.29", + "rustls", "socket2", "thiserror 2.0.12", "tokio", @@ -1608,7 +1757,7 @@ dependencies = [ "rand 0.9.1", "ring", "rustc-hash", - "rustls 0.23.29", + "rustls", "rustls-pki-types", "slab", "thiserror 2.0.12", @@ -1780,7 +1929,7 @@ checksum = "1165225c21bff1f3bbce98f5a1f889949bc902d3575308cc7b0de30b4f6d27c7" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -1853,14 +2002,14 @@ dependencies = [ "percent-encoding", "pin-project-lite", "quinn", - "rustls 0.23.29", + "rustls", "rustls-pki-types", "serde", "serde_json", "serde_urlencoded", "sync_wrapper", "tokio", - "tokio-rustls 0.26.2", + "tokio-rustls", "tower 0.5.2", "tower-http", "tower-service", @@ -1868,7 +2017,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", - "webpki-roots 1.0.1", + "webpki-roots", ] [[package]] @@ -1906,7 +2055,7 @@ dependencies = [ "quote", "rust-embed-utils", "shellexpand", - "syn", + "syn 2.0.104", "walkdir", ] @@ -1942,6 +2091,7 @@ dependencies = [ "axum-extra", "axum-prometheus", "bitflags", + "blake3", "chrono", "clap", "clickhouse", @@ -1953,12 +2103,14 @@ dependencies = [ "metrics-prometheus", "mimalloc", "mime_guess", + "moka", "pretty_assertions", "prometheus", "rand 0.9.1", "rayon", "reqwest", "rust-embed", + "rustls", "schemars", "serde", "serde_json", @@ -1975,28 +2127,18 @@ dependencies = [ "uuid", ] -[[package]] -name = "rustls" -version = "0.21.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f56a14d1f48b391359b22f731fd4bd7e43c97f3c50eee276f3aa09c94784d3e" -dependencies = [ - "log", - "ring", - "rustls-webpki 0.101.7", - "sct", -] - [[package]] name = "rustls" version = "0.23.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2491382039b29b9b11ff08b76ff6c97cf287671dbb74f0be44bda389fffe9bd1" dependencies = [ + "aws-lc-rs", + "log", "once_cell", "ring", "rustls-pki-types", - "rustls-webpki 0.103.4", + "rustls-webpki", "subtle", "zeroize", ] @@ -2011,22 +2153,13 @@ dependencies = [ "zeroize", ] -[[package]] -name = "rustls-webpki" -version = "0.101.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b6275d1ee7a1cd780b64aca7726599a1dbc893b1e64144529e55c3c2f745765" -dependencies = [ - "ring", - "untrusted", -] - [[package]] name = "rustls-webpki" version = "0.103.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0a17884ae0c1b773f1ccd2bd4a8c72f16da897310a98b0e84bf349ad5ead92fc" dependencies = [ + "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", @@ -2077,7 +2210,7 @@ dependencies = [ "proc-macro2", "quote", "serde_derive_internals", - "syn", + "syn 2.0.104", ] [[package]] @@ -2087,24 +2220,25 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" [[package]] -name = "sct" -version = "0.7.1" +name = "sealed" +version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da046153aa2352493d6cb7da4b6e5c0c057d8a1d0a9aa8560baffdd945acd414" +checksum = "22f968c5ea23d555e670b449c1c5e7b2fc399fdaec1d304a17cd48e288abc107" dependencies = [ - "ring", - "untrusted", + "proc-macro2", + "quote", + "syn 2.0.104", ] [[package]] name = "sealed" -version = "0.6.0" +version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22f968c5ea23d555e670b449c1c5e7b2fc399fdaec1d304a17cd48e288abc107" +checksum = "9b68e2ea526d9fb32f23ca8894fb5da9e743f34c2f41701f0501dc8a25c4b343" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.6", ] [[package]] @@ -2124,7 +2258,7 @@ checksum = "5b0276cf7f2c73365f7157c8123c21cd9a50fbbd844757af28ca1f5925fc2a00" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -2135,7 +2269,7 @@ checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -2182,7 +2316,7 @@ checksum = "175ee3e80ae9982737ca543e96133087cbd9a485eecc3bc4de9c1a37b47ea59c" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -2204,7 +2338,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", ] @@ -2215,7 +2349,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", ] @@ -2239,9 +2373,9 @@ dependencies = [ [[package]] name = "shlex" -version = "1.3.0" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" [[package]] name = "signal-hook-registry" @@ -2317,7 +2451,7 @@ dependencies = [ "proc-macro2", "quote", "rustversion", - "syn", + "syn 2.0.104", ] [[package]] @@ -2337,6 +2471,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "1.0.2" @@ -2354,9 +2499,15 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] +[[package]] +name = "tagptr" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" + [[package]] name = "thiserror" version = "1.0.69" @@ -2383,7 +2534,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -2394,7 +2545,7 @@ checksum = "7f7cf42b4507d8ea322120659672cf1b9dbb93f8f2d4ecfd6e51350ff5b17a1d" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -2467,17 +2618,7 @@ checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8" dependencies = [ "proc-macro2", "quote", - "syn", -] - -[[package]] -name = "tokio-rustls" -version = "0.24.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c28327cf380ac148141087fbfb9de9d7bd4e84ab5d2c28fbc911d753de8a7081" -dependencies = [ - "rustls 0.21.12", - "tokio", + "syn 2.0.104", ] [[package]] @@ -2486,7 +2627,7 @@ version = "0.26.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e727b36a1a0e8b74c376ac2211e40c2c8af09fb4013c60d910495810f008e9b" dependencies = [ - "rustls 0.23.29", + "rustls", "tokio", ] @@ -2596,7 +2737,7 @@ checksum = "81383ab64e72a7a8b8e13130c49e3dab29def6d0c7d76a03087b3cf71c5c6903" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -2646,9 +2787,9 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "twitch-irc" -version = "5.0.1" +version = "6.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba8e5095c3a4d4d72decb36b2ceb1e40085e7d8d93282b1d2f3cf2bba0dee4db" +checksum = "40f18ad59bffa7d9456f585f3baeff9063286bb6765290e2875602c292b2cee9" dependencies = [ "async-trait", "bytes", @@ -2658,13 +2799,13 @@ dependencies = [ "futures-util", "prometheus", "smallvec", - "thiserror 1.0.69", + "thiserror 2.0.12", "tokio", - "tokio-rustls 0.24.1", + "tokio-rustls", "tokio-stream", "tokio-util", "tracing", - "webpki-roots 0.25.4", + "webpki-roots", ] [[package]] @@ -2852,7 +2993,7 @@ dependencies = [ "log", "proc-macro2", "quote", - "syn", + "syn 2.0.104", "wasm-bindgen-shared", ] @@ -2887,7 +3028,7 @@ checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", "wasm-bindgen-backend", "wasm-bindgen-shared", ] @@ -2921,12 +3062,6 @@ dependencies = [ "wasm-bindgen", ] -[[package]] -name = "webpki-roots" -version = "0.25.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" - [[package]] name = "webpki-roots" version = "1.0.1" @@ -2988,7 +3123,7 @@ checksum = "a47fddd13af08290e67f4acabf4b459f647552718f683a7b415d290ac744a836" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -2999,7 +3134,7 @@ checksum = "bd9211b69f8dcdfa817bfd14bf1c97c9188afa36f4750130fcdf3f400eca9fa8" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -3222,7 +3357,7 @@ checksum = "38da3c9736e16c5d3c8c597a9aaa5d1fa565d0532ae05e27c24aa62fb32c0ab6" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", "synstructure", ] @@ -3243,7 +3378,7 @@ checksum = "9ecf5b4cc5364572d7f4c329661bcc82724222973f2cab6f050a4e5c22f75181" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] @@ -3263,7 +3398,7 @@ checksum = "d71e5d6e06ab090c67b5e44993ec16b72dcbaabc526db883a360057678b48502" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", "synstructure", ] @@ -3303,7 +3438,7 @@ checksum = "5b96237efa0c878c64bd89c436f661be4e46b2f3eff1ebb976f7ef2321d2f58f" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.104", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 76745ca..4f9ba11 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,7 @@ aide = { version = "0.15.0", features = [ ] } anyhow = "1.0.75" axum = { version = "0.8.4", features = ["tokio"] } +blake3 = "1.8.2" chrono = { version = "0.4.27", features = ["serde"] } clap = { version = "4.4.1", features = ["derive"] } clickhouse = { version = "0.13.0", default-features = false, features = [ @@ -27,12 +28,14 @@ indexmap = "2.2.6" lazy_static = "1.4.0" mimalloc = { version = "0.1.38", default-features = false } mime_guess = "2.0.4" -prometheus = "0.13.3" +prometheus = "0.14" rand = "0.9.0" rayon = "1.7.0" reqwest = { version = "0.12.4", features = [ + "json", "rustls-tls", ], default-features = false } +rustls = { version = "0.23", default-features = false, features = ["ring"] } rust-embed = { version = "8.0.0", features = ["interpolate-folder-path"] } schemars = { version = "0.9", features = ["chrono04"] } serde = { version = "1.0.188", features = ["derive"] } @@ -49,7 +52,7 @@ tower-http = { version = "0.6.1", features = [ ] } tracing = "0.1.37" tracing-subscriber = { version = "0.3.17", features = ["env-filter"] } -twitch-irc = { version = "5.0.1", default-features = false, features = [ +twitch-irc = { version = "6.1.2", default-features = false, features = [ "metrics-collection", "transport-tcp-rustls-webpki-roots", ] } @@ -61,7 +64,8 @@ twitch_api = { version = "0.7.0", features = [ ] } tmi = { version = "0.7.0", default-features = false, features = ["simd"] } axum-prometheus = "0.8.0" -metrics-prometheus = "0.8.0" +metrics-prometheus = "0.11.2" +moka = { version = "0.12.14", features = ["sync"] } axum-extra = { version = "0.10.0", features = ["typed-header"] } bitflags = { version = "2.5.0", features = ["serde"] } diff --git a/config.dist.json b/config.dist.json index 829c658..24b24db 100644 --- a/config.dist.json +++ b/config.dist.json @@ -4,6 +4,9 @@ "clickhouseUsername": "user", "clickhousePassword": "SuperSecretPassword", "listenAddress": "0.0.0.0:8025", + "recentMessagesEnabled": false, + "recentMessagesUrl": "https://recent-messages.robotty.de/api/v2/recent-messages", + "recentMessagesLimit": 800, "channels": ["12345"], "clientID": "id", "clientSecret": "secret", diff --git a/docs/CONFIG.md b/docs/CONFIG.md index f2334fa..258dc77 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -9,6 +9,9 @@ Available options: - `clickhousePassword` (string): Clickhouse password. - `clickhouseFlushInterval` (number): Interval (in seconds) of how often messages should be flushed to the database. A lower value means that logs are available sooner at the expensive of higher database load. Defaults to 10. - `listenAddress` (string): Listening address for the web server. Defaults to `0.0.0.0:8025`. +- `recentMessagesEnabled` (boolean): Enables best-effort Robotty recent-message recovery. Defaults to `false`. +- `recentMessagesUrl` (string): Robotty-compatible recent-messages base URL. Defaults to `https://recent-messages.robotty.de/api/v2/recent-messages`. Only HTTP and HTTPS URLs are accepted. +- `recentMessagesLimit` (number): Maximum number of recent messages requested per channel. Defaults to `800`. - `channels` (array of strings): List of channel ids to be logged. - `clientId` (string): Twitch client id. - `clientSecret` (string): Twitch client secret. @@ -16,6 +19,10 @@ Available options: - `optOut` (object of strings: booleans): List of user ids who opted out from being logged. - `adminAPIKey` (string): API key for admin requests +When enabled, rustlog requests recent messages after it successfully joins a channel at startup, after reconnecting, and after a runtime channel addition. Backfill is best-effort: request, response, and individual IRC parsing failures are logged without stopping live Twitch ingest. + +Replayed messages never execute chat commands. A live command still executes if recovery has already stored its message. Removing a channel stops further recovery requests and prevents pending responses from being queued for storage. + Example config: ```json { @@ -24,6 +31,9 @@ Example config: "clickhouseUsername": "user", "clickhousePassword": "SuperSecretPassword", "listenAddress": "0.0.0.0:8025", + "recentMessagesEnabled": false, + "recentMessagesUrl": "https://recent-messages.robotty.de/api/v2/recent-messages", + "recentMessagesLimit": 800, "channels": ["12345"], "clientID": "id", "clientSecret": "secret", diff --git a/src/bot.rs b/src/bot.rs index 55ee46d..d1af8c2 100644 --- a/src/bot.rs +++ b/src/bot.rs @@ -2,16 +2,28 @@ use crate::{ app::App, db::schema::{StructuredMessage, UnstructuredMessage}, logs::extract::{extract_channel_and_user_from_raw, extract_raw_timestamp}, + recent_messages::RecentMessagesClient, ShutdownRx, }; use anyhow::{anyhow, Context}; use chrono::Utc; +use dashmap::DashSet; use lazy_static::lazy_static; +use moka::sync::Cache; use prometheus::{register_int_counter_vec, IntCounterVec}; -use std::time::Duration; +use std::{ + borrow::Cow, + collections::{HashMap, HashSet}, + sync::Arc, + time::Duration, +}; use tokio::{ - sync::mpsc::{Receiver, Sender}, - time::sleep, + sync::{ + mpsc::{Receiver, Sender}, + Mutex, Notify, OwnedSemaphorePermit, Semaphore, + }, + task::JoinHandle, + time::{sleep, sleep_until, Instant}, }; use tracing::{debug, error, info, log::warn, trace}; use twitch_irc::{ @@ -21,7 +33,13 @@ use twitch_irc::{ }; const CHANNEL_REJOIN_INTERVAL_SECONDS: u64 = 3600; -const CHANENLS_REFETCH_RETRY_INTERVAL_SECONDS: u64 = 5; +const CHANNELS_REFETCH_RETRY_INTERVAL_SECONDS: u64 = 5; +const RECENT_MESSAGE_DEDUPE_TTL_SECONDS: u64 = 30; +const RECENT_MESSAGE_DEDUPE_CAPACITY: u64 = 50_000; +const RECENT_MESSAGE_MAX_CONCURRENT_REQUESTS: usize = 4; +const RECENT_MESSAGE_REQUEST_SPACING_MILLIS: u64 = 250; +const RECENT_MESSAGE_STARTUP_DELAYS_SECONDS: [u64; 5] = [0, 5, 15, 30, 60]; +const RECENT_MESSAGE_RECONNECT_DELAYS_SECONDS: [u64; 6] = [0, 5, 15, 30, 60, 120]; type TwitchClient = TwitchIRCClient; @@ -57,11 +75,73 @@ pub async fn run( struct Bot { app: App, writer_tx: Sender>, + recent_messages: RecentMessagesClient, + recent_message_dedupe: Cache, + live_command_dedupe: Cache, + backfill_states: Arc>>, + backfill_tasks: Arc>>>, + seen_channel_joins: Arc>, + recent_message_request_slots: Arc, + recent_message_next_request: Arc>, + recent_message_request_spacing: Duration, + startup_backfill_delays: Arc<[Duration]>, + reconnect_backfill_delays: Arc<[Duration]>, + #[cfg(test)] + skip_existing_message_lookup: bool, +} + +struct PreparedMessage { + message: StructuredMessage<'static>, + dedupe_key: String, +} + +struct BackfillState { + restart_requested: bool, + wakeup: Arc, } impl Bot { pub fn new(app: App, writer_tx: Sender>) -> Bot { - Self { app, writer_tx } + let recent_messages = RecentMessagesClient::from_config(&app.config); + Self::new_with_recent_messages(app, writer_tx, recent_messages) + } + + fn new_with_recent_messages( + app: App, + writer_tx: Sender>, + recent_messages: RecentMessagesClient, + ) -> Bot { + Self { + app, + writer_tx, + recent_messages, + recent_message_dedupe: Cache::builder() + .time_to_live(Duration::from_secs(RECENT_MESSAGE_DEDUPE_TTL_SECONDS)) + .max_capacity(RECENT_MESSAGE_DEDUPE_CAPACITY) + .build(), + live_command_dedupe: Cache::builder() + .time_to_live(Duration::from_secs(RECENT_MESSAGE_DEDUPE_TTL_SECONDS)) + .max_capacity(RECENT_MESSAGE_DEDUPE_CAPACITY) + .build(), + backfill_states: Arc::new(Mutex::new(HashMap::new())), + backfill_tasks: Arc::new(Mutex::new(Vec::new())), + seen_channel_joins: Arc::new(DashSet::new()), + recent_message_request_slots: Arc::new(Semaphore::new( + RECENT_MESSAGE_MAX_CONCURRENT_REQUESTS, + )), + recent_message_next_request: Arc::new(Mutex::new(Instant::now())), + recent_message_request_spacing: Duration::from_millis( + RECENT_MESSAGE_REQUEST_SPACING_MILLIS, + ), + startup_backfill_delays: RECENT_MESSAGE_STARTUP_DELAYS_SECONDS + .map(Duration::from_secs) + .into(), + reconnect_backfill_delays: RECENT_MESSAGE_RECONNECT_DELAYS_SECONDS + .map(Duration::from_secs) + .into(), + #[cfg(test)] + skip_existing_message_lookup: false, + } } pub async fn run( @@ -70,6 +150,13 @@ impl Bot { mut shutdown_rx: ShutdownRx, mut command_rx: Receiver, ) { + let own_login = match login_credentials.get_credentials().await { + Ok(credentials) => Some(credentials.login), + Err(err) => { + warn!("Could not determine IRC login for recent-message backfill: {err}"); + None + } + }; let client_config = ClientConfig::new_simple(login_credentials); let (mut receiver, client) = TwitchIRCClient::::new(client_config); @@ -95,7 +182,7 @@ impl Bot { } Err(err) => { error!("Could not fetch users list: {err}"); - CHANENLS_REFETCH_RETRY_INTERVAL_SECONDS + CHANNELS_REFETCH_RETRY_INTERVAL_SECONDS } }; sleep(Duration::from_secs(interval)).await; @@ -138,12 +225,13 @@ impl Bot { loop { tokio::select! { Some(msg) = receiver.recv() => { - if let Err(e) = self.handle_message(msg, &client).await { + if let Err(e) = self.handle_message(msg, &client, own_login.as_deref()).await { error!("Could not handle message: {e}"); } } _ = shutdown_rx.changed() => { debug!("Shutting down bot task"); + self.stop_backfill_tasks().await; break; } } @@ -154,24 +242,59 @@ impl Bot { &self, msg: ServerMessage, client: &TwitchClient, + own_login: Option<&str>, ) -> anyhow::Result<()> { + if let Some(own_login) = own_login { + self.trigger_backfill_for_own_join(&msg, own_login).await; + } + + let raw_irc = msg.as_raw_irc(); + let prepared = self.prepare_message(IRCMessage::from(msg.clone()), &raw_irc)?; + let Some(prepared) = prepared else { + return Ok(()); + }; if let ServerMessage::Privmsg(privmsg) = &msg { trace!("Processing message {}", privmsg.message_text); if let Some(cmd) = privmsg.message_text.strip_prefix(COMMAND_PREFIX) { - if let Err(err) = self - .handle_command(cmd, client, &privmsg.sender.id, &privmsg.sender.login) - .await + // A replay may store the message first, but only live IRC may execute commands. + if self + .live_command_dedupe + .entry(prepared.dedupe_key.clone()) + .or_insert_with(|| ()) + .is_fresh() { - warn!("Could not handle command {cmd}: {err:#}"); + if let Err(err) = self + .handle_command(cmd, client, &privmsg.sender.id, &privmsg.sender.login) + .await + { + warn!("Could not handle command {cmd}: {err:#}"); + } } } } - self.write_message(msg).await?; + if self.claim_message(&prepared.dedupe_key) { + self.write_prepared(prepared).await?; + } Ok(()) } + async fn trigger_backfill_for_own_join(&self, msg: &ServerMessage, own_login: &str) { + if let ServerMessage::Join(join) = msg { + if join.user_login.eq_ignore_ascii_case(own_login) { + let channel_login = normalize_channel_login(&join.channel_login); + let schedule = if self.seen_channel_joins.insert(channel_login.clone()) { + BackfillSchedule::Startup + } else { + BackfillSchedule::Reconnect + }; + self.trigger_recent_messages_fetch(channel_login, "successful join", schedule) + .await; + } + } + } + fn check_admin(&self, user_login: &str) -> anyhow::Result<()> { if self .app @@ -186,49 +309,453 @@ impl Bot { } } - async fn write_message(&self, msg: ServerMessage) -> anyhow::Result<()> { - // Ignore - if matches!(msg, ServerMessage::RoomState(_)) { - return Ok(()); + fn prepare_message( + &self, + irc_message: IRCMessage, + raw_irc: &str, + ) -> anyhow::Result> { + if irc_message.command == "ROOMSTATE" { + return Ok(None); } - - let irc_message = IRCMessage::from(msg); - if let Some((channel_id, maybe_user_id)) = extract_channel_and_user_from_raw(&irc_message) { - if !channel_id.is_empty() { - MESSAGES_RECEIVED_COUNTERS - .with_label_values(&[channel_id]) - .inc(); - } - let timestamp = extract_raw_timestamp(&irc_message) .unwrap_or_else(|| Utc::now().timestamp_millis().try_into().unwrap()); let user_id = maybe_user_id.unwrap_or_default().to_owned(); - - if self.app.config.opt_out.contains_key(&user_id) { - return Ok(()); - } - - let raw_irc = irc_message.as_raw_irc(); let unstructured = UnstructuredMessage { channel_id, user_id: &user_id, timestamp, - raw: &raw_irc, + raw: raw_irc, }; - match StructuredMessage::from_unstructured(&unstructured) { - Ok(msg) => { - self.writer_tx.send(msg.into_owned()).await?; + let mut message = StructuredMessage::from_unstructured(&unstructured)?.into_owned(); + message.channel_login = normalize_channel_login(&message.channel_login).into(); + let dedupe_key = structured_event_identity(&message); + return Ok(Some(PreparedMessage { + message, + dedupe_key, + })); + } + + Ok(None) + } + + fn claim_message(&self, dedupe_key: &str) -> bool { + self.recent_message_dedupe + .entry(dedupe_key.to_owned()) + .or_insert_with(|| ()) + .is_fresh() + } + + async fn write_prepared(&self, prepared: PreparedMessage) -> anyhow::Result<()> { + self.write_prepared_for_channel(prepared, false).await + } + + async fn write_prepared_for_channel( + &self, + prepared: PreparedMessage, + require_joined: bool, + ) -> anyhow::Result<()> { + let message = prepared.message; + if !self.should_write_message(&message, require_joined) { + return Ok(()); + } + let permit = self.writer_tx.reserve().await?; + if !self.should_write_message(&message, require_joined) { + return Ok(()); + } + + if !message.channel_id.is_empty() { + MESSAGES_RECEIVED_COUNTERS + .with_label_values(&[message.channel_id.as_ref()]) + .inc(); + } + permit.send(message); + Ok(()) + } + + fn should_write_message(&self, message: &StructuredMessage<'_>, require_joined: bool) -> bool { + if require_joined + && !self + .app + .config + .channels + .read() + .unwrap() + .contains(message.channel_id.as_ref()) + { + return false; + } + !self + .app + .config + .opt_out + .contains_key(message.channel_id.as_ref()) + && !self + .app + .config + .opt_out + .contains_key(message.user_id.as_ref()) + } + + async fn trigger_recent_messages_fetch( + &self, + channel_login: String, + reason: &'static str, + schedule: BackfillSchedule, + ) { + if !self.recent_messages.enabled() { + return; + } + + let channel_login = normalize_channel_login(&channel_login); + let wakeup = { + let mut states = self.backfill_states.lock().await; + if let Some(state) = states.get_mut(&channel_login) { + if matches!(schedule, BackfillSchedule::Reconnect) { + state.restart_requested = true; + // Retain a notification if the task has not started waiting yet. + state.wakeup.notify_one(); + } + return; + } + let wakeup = Arc::new(Notify::new()); + states.insert( + channel_login.clone(), + BackfillState { + restart_requested: false, + wakeup: wakeup.clone(), + }, + ); + wakeup + }; + + let bot = self.clone(); + let mut delays: Arc<[Duration]> = match schedule { + #[cfg(test)] + BackfillSchedule::OneShot => [Duration::ZERO].into(), + BackfillSchedule::Startup => self.startup_backfill_delays.clone(), + BackfillSchedule::Reconnect => self.reconnect_backfill_delays.clone(), + }; + let reconnect_delays = self.reconnect_backfill_delays.clone(); + let handle = tokio::spawn(async move { + 'restart_schedule: loop { + let started_at = Instant::now(); + let active_delays = delays.clone(); + for (attempt, delay) in active_delays.iter().enumerate() { + let deadline = started_at + *delay; + loop { + if bot.take_backfill_restart(&channel_login).await { + delays = reconnect_delays.clone(); + info!( + "Restarting recent-message recovery schedule for {channel_login} after another successful join" + ); + continue 'restart_schedule; + } + tokio::select! { + _ = sleep_until(deadline) => break, + _ = wakeup.notified() => continue, + } + } + + let expected_channel_id = match bot + .resolve_expected_channel_id(&channel_login) + .await + { + Ok(channel_id) => channel_id, + Err(err) => { + warn!( + "Recent-message backfill could not verify {channel_login} after {reason} (attempt {} of {}): {err:#}", + attempt + 1, + active_delays.len() + ); + continue; + } + }; + if !bot + .app + .config + .channels + .read() + .unwrap() + .contains(&expected_channel_id) + { + break; + } + if let Err(err) = bot + .fetch_recent_messages_for_channel(&channel_login, &expected_channel_id) + .await + { + warn!( + "Recent-message backfill failed for {channel_login} after {reason} (attempt {} of {}): {err:#}", + attempt + 1, + active_delays.len() + ); + } + + if bot.take_backfill_restart(&channel_login).await { + delays = reconnect_delays.clone(); + info!( + "Restarting recent-message recovery schedule for {channel_login} after another successful join" + ); + continue 'restart_schedule; + } } - Err(err) => { - error!("Could not convert message {unstructured:?} to be logged: {err}"); + + // Finish and reconnect registration share a lock, so a join cannot + // leave a restart request behind after the task has exited. + let mut states = bot.backfill_states.lock().await; + let state = states + .get_mut(&channel_login) + .expect("active backfill state"); + if std::mem::take(&mut state.restart_requested) { + delays = reconnect_delays.clone(); + continue; } + states.remove(&channel_login); + break; } + }); + let mut tasks = self.backfill_tasks.lock().await; + tasks.retain(|task| !task.is_finished()); + tasks.push(handle); + } + + async fn take_backfill_restart(&self, channel_login: &str) -> bool { + let mut states = self.backfill_states.lock().await; + states + .get_mut(channel_login) + .map(|state| std::mem::take(&mut state.restart_requested)) + .unwrap_or(false) + } + + async fn resolve_expected_channel_id(&self, channel_login: &str) -> anyhow::Result { + match self.app.users.get_id(channel_login) { + Some(Some(channel_id)) => return Ok(channel_id), + Some(None) => return Err(anyhow!("Twitch reports the channel is unavailable")), + None => {} } + self.app + .get_users(vec![], vec![channel_login.to_owned()], false) + .await? + .into_iter() + .find_map(|(channel_id, login)| { + (normalize_channel_login(&login) == channel_login).then_some(channel_id) + }) + .context("Twitch did not return the expected channel ID") + } + + async fn fetch_recent_messages_for_channel( + &self, + channel_login: &str, + expected_channel_id: &str, + ) -> anyhow::Result<()> { + let channel_login = normalize_channel_login(channel_login); + let _request_permit = self.acquire_recent_message_request_permit().await?; + if !self + .app + .config + .channels + .read() + .unwrap() + .contains(expected_channel_id) + { + return Ok(()); + } + let response = self.recent_messages.fetch(&channel_login).await?; + if !self + .app + .config + .channels + .read() + .unwrap() + .contains(expected_channel_id) + { + return Ok(()); + } + if response.error.is_some() || response.error_code.is_some() { + warn!( + "Recent-message backfill skipped for {channel_login}: error={:?} error_code={:?}", + response.error, response.error_code + ); + return Ok(()); + } + + let response_count = response.messages.len(); + let mut candidates = Vec::new(); + let mut parse_failures = 0_usize; + let mut missing_timestamps = 0_usize; + let mut channel_login_mismatches = 0_usize; + let mut channel_id_mismatches = 0_usize; + for (response_index, raw) in response.messages.into_iter().enumerate() { + let normalized_raw = normalize_robotty_message(&raw); + let parsed = IRCMessage::parse(normalized_raw.as_ref()).map_err(anyhow::Error::from); + match parsed { + Ok(message) => { + if extract_raw_timestamp(&message).is_none() { + missing_timestamps += 1; + continue; + } + match self.prepare_message(message, normalized_raw.as_ref()) { + Ok(Some(prepared)) => { + if prepared.message.channel_login != channel_login { + channel_login_mismatches += 1; + } else if prepared.message.channel_id != expected_channel_id { + channel_id_mismatches += 1; + } else { + candidates.push((response_index, prepared)); + } + } + Ok(None) => {} + Err(_) => parse_failures += 1, + } + } + Err(_) => parse_failures += 1, + } + } + + candidates.sort_by_key(|(response_index, prepared)| { + (prepared.message.timestamp, *response_index) + }); + let mut response_keys = HashSet::new(); + let before_response_dedupe = candidates.len(); + candidates.retain(|(_, prepared)| response_keys.insert(prepared.dedupe_key.clone())); + let response_duplicates = before_response_dedupe - candidates.len(); + let candidates = candidates + .into_iter() + .map(|(_, prepared)| prepared) + .collect::>(); + + let existing_keys = self.existing_message_keys(&candidates).await?; + let mut existing_duplicates = 0_usize; + let mut recent_duplicates = 0_usize; + let mut stored = 0_usize; + let mut write_failures = 0_usize; + for prepared in candidates { + // The channel can be removed while the HTTP request or database lookup is pending. + if !self + .app + .config + .channels + .read() + .unwrap() + .contains(expected_channel_id) + { + break; + } + if existing_keys.contains(&prepared.dedupe_key) { + existing_duplicates += 1; + continue; + } + if !self.claim_message(&prepared.dedupe_key) { + recent_duplicates += 1; + continue; + } + if let Err(err) = self.write_prepared_for_channel(prepared, true).await { + write_failures += 1; + warn!("Could not store recent-message backfill event for {channel_login}: {err}"); + } else { + stored += 1; + } + } + + info!( + "Recent-message backfill summary for {channel_login}: received={response_count} stored={stored} response_duplicates={response_duplicates} existing_duplicates={existing_duplicates} recent_duplicates={recent_duplicates} malformed={parse_failures} missing_timestamp={missing_timestamps} channel_login_mismatch={channel_login_mismatches} channel_id_mismatch={channel_id_mismatches} write_failures={write_failures}" + ); + Ok(()) } + async fn acquire_recent_message_request_permit(&self) -> anyhow::Result { + let permit = self + .recent_message_request_slots + .clone() + .acquire_owned() + .await + .context("Recent-message request gate was closed")?; + let mut next_request = self.recent_message_next_request.lock().await; + let now = Instant::now(); + if *next_request > now { + sleep_until(*next_request).await; + } + *next_request = Instant::now() + self.recent_message_request_spacing; + Ok(permit) + } + + async fn existing_message_keys( + &self, + candidates: &[PreparedMessage], + ) -> anyhow::Result> { + #[cfg(test)] + if self.skip_existing_message_lookup { + return Ok(HashSet::new()); + } + + let candidate_keys = candidates + .iter() + .map(|prepared| prepared.dedupe_key.clone()) + .collect::>(); + if candidate_keys.is_empty() { + return Ok(HashSet::new()); + } + + let mut existing_keys = HashSet::new(); + let mut groups = HashMap::::new(); + for prepared in candidates { + let entry = groups + .entry(prepared.message.channel_id.to_string()) + .or_insert((prepared.message.timestamp, prepared.message.timestamp)); + entry.0 = entry.0.min(prepared.message.timestamp); + entry.1 = entry.1.max(prepared.message.timestamp); + } + + for (channel_id, (from_millis, to_millis)) in groups { + let buffered = self + .app + .flush_buffer + .messages_by_channel(from_millis..to_millis.saturating_add(1), &channel_id) + .await; + existing_keys.extend( + buffered + .iter() + .map(structured_event_identity) + .filter(|key| candidate_keys.contains(key)), + ); + + let stored_messages = self + .app + .db + .query( + "SELECT ?fields FROM message_structured WHERE channel_id = ? AND timestamp >= ? AND timestamp <= ?", + ) + .bind(channel_id) + .bind(from_millis as f64 / 1000.0) + .bind(to_millis as f64 / 1000.0) + .fetch_all::>() + .await + .context("could not check stored recent-message identities")?; + existing_keys.extend( + stored_messages + .iter() + .map(structured_event_identity) + .filter(|key| candidate_keys.contains(key)), + ); + } + + Ok(existing_keys) + } + + async fn stop_backfill_tasks(&self) { + let tasks = std::mem::take(&mut *self.backfill_tasks.lock().await); + for task in &tasks { + task.abort(); + } + for task in tasks { + let _ = task.await; + } + self.backfill_states.lock().await.clear(); + } + async fn handle_command( &self, cmd: &str, @@ -316,6 +843,8 @@ impl Bot { ChannelAction::Part => { info!("Parting channel {channel_name}"); config_channels.remove(&channel_id); + self.seen_channel_joins + .remove(&normalize_channel_login(&channel_name)); client.part(channel_name); } } @@ -332,3 +861,819 @@ enum ChannelAction { Join, Part, } + +#[derive(Clone, Copy)] +enum BackfillSchedule { + #[cfg(test)] + OneShot, + Startup, + Reconnect, +} + +fn normalize_channel_login(channel_login: &str) -> String { + channel_login + .trim_matches(|character: char| character.is_ascii_whitespace() || character == '\0') + .trim_start_matches('#') + .to_ascii_lowercase() +} + +fn trim_irc_framing(raw: &str) -> &str { + raw.trim_matches(|character: char| character.is_ascii_whitespace() || character == '\0') +} + +fn normalize_robotty_message(raw: &str) -> Cow<'_, str> { + let raw = trim_irc_framing(raw); + let Some(tagged) = raw.strip_prefix('@') else { + return Cow::Borrowed(raw); + }; + let Some((tags, remainder)) = tagged.split_once(' ') else { + return Cow::Borrowed(raw); + }; + if tags + .split(';') + .all(|tag| tag.is_empty() || tag.contains('=')) + { + return Cow::Borrowed(raw); + } + + let mut normalized = String::with_capacity(raw.len() + 8); + normalized.push('@'); + for (index, tag) in tags.split(';').enumerate() { + if index > 0 { + normalized.push(';'); + } + normalized.push_str(tag); + if !tag.is_empty() && !tag.contains('=') { + normalized.push('='); + } + } + normalized.push(' '); + normalized.push_str(remainder); + Cow::Owned(normalized) +} + +fn structured_event_identity(message: &StructuredMessage<'_>) -> String { + if let Some(message_id) = message.id() { + format!("uuid:{message_id}") + } else { + let mut canonical_message = message.clone(); + // IRC tag order can differ between live messages and recent-message replay. + canonical_message.extra_tags.sort_unstable(); + let canonical = serde_json::to_vec(&canonical_message) + .expect("serializing a structured message for deduplication cannot fail"); + format!("row:{}", blake3::hash(&canonical).to_hex()) + } +} + +#[cfg(test)] +mod tests { + use super::{structured_event_identity, BackfillSchedule, Bot, ServerMessage}; + use crate::{ + app::{cache::UsersCache, App}, + config::Config, + db::writer::FlushBuffer, + recent_messages::{RecentMessagesClient, RecentMessagesRuntimeSummary}, + }; + use axum::{ + extract::Path, http::StatusCode, response::IntoResponse, routing::get, Json, Router, + }; + use dashmap::DashSet; + use std::{ + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex as StdMutex, + }, + time::{Duration, Instant as StdInstant}, + }; + use tokio::{net::TcpListener, sync::mpsc, time::timeout}; + use twitch_api::{ + twitch_oauth2::{AccessToken, AppAccessToken, ClientId, ClientSecret}, + HelixClient, + }; + use twitch_irc::message::IRCMessage; + + const REPLAYED_COMMAND: &str = "@room-id=1;user-id=200;tmi-sent-ts=1704067200000;id=robotty-command;display-name=admin;badge-info=;badges=;color=;user-type=;emotes=;flags= :admin!admin@admin.tmi.twitch.tv PRIVMSG #channelone :!rustlog optout secret-code"; + const UUID_MESSAGE: &str = "@room-id=1;user-id=200;tmi-sent-ts=1704067200000;id=272e342c-5864-4c59-b730-25908cdb7f57;display-name=user;badges=;color=;user-type=;emotes=;flags= :user!user@user.tmi.twitch.tv PRIVMSG #channelone :hello"; + const ROBOTTY_VALUELESS_MESSAGE: &str = "@display-name=user;flags;badge-info;id=272e342c-5864-4c59-b730-25908cdb7f57;badges;user-type;color=#00FF7F;emotes;room-id=1;tmi-sent-ts=1704067200000;user-id=200 :user!user@user.tmi.twitch.tv PRIVMSG #channelone :hello"; + + fn test_app() -> App { + let config: Config = serde_json::from_value(serde_json::json!({ + "clickhouseUrl": "http://127.0.0.1:9", + "clickhouseDb": "rustlog", + "channels": ["1"], + "clientID": "client", + "clientSecret": "secret", + "admins": ["admin"], + "optOut": {} + })) + .unwrap(); + let optout_codes = Arc::new(DashSet::new()); + optout_codes.insert("secret-code".to_string()); + let users = UsersCache::default(); + users.insert("1".to_string(), "channelone".to_string()); + App { + helix_client: HelixClient::default(), + token: Arc::new(AppAccessToken::from_existing_unchecked( + AccessToken::new("token".to_string()), + None, + ClientId::new("client".to_string()), + ClientSecret::new("secret".to_string()), + None, + None, + )), + users, + optout_codes, + db: Arc::new(clickhouse::Client::default().with_url("http://127.0.0.1:9")), + config: Arc::new(config), + flush_buffer: FlushBuffer::default(), + } + } + + fn recent_messages_client(address: std::net::SocketAddr) -> RecentMessagesClient { + RecentMessagesClient::from_summary(RecentMessagesRuntimeSummary { + enabled: true, + base_url: Some(format!("http://{address}/api/v2/recent-messages")), + limit: 800, + warnings: Vec::new(), + }) + } + + async fn bot_with_server( + router: Router, + ) -> ( + Bot, + mpsc::Receiver>, + tokio::task::JoinHandle>, + ) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await }); + let (writer_tx, writer_rx) = mpsc::channel(10); + let mut bot = + Bot::new_with_recent_messages(test_app(), writer_tx, recent_messages_client(address)); + bot.recent_message_request_spacing = Duration::ZERO; + bot.skip_existing_message_lookup = true; + (bot, writer_rx, server) + } + + async fn wait_for_backfills(bot: &Bot) { + let tasks = std::mem::take(&mut *bot.backfill_tasks.lock().await); + for task in tasks { + task.await.unwrap(); + } + } + + #[tokio::test] + async fn replay_stores_command_once_without_command_side_effects() { + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get(|| async { + Json(serde_json::json!({ + "messages": ["not raw irc", REPLAYED_COMMAND, REPLAYED_COMMAND], + "error": null, + "error_code": null + })) + }), + ); + let (bot, mut writer_rx, server) = bot_with_server(router).await; + + bot.fetch_recent_messages_for_channel("channelone", "1") + .await + .unwrap(); + + let stored = timeout(Duration::from_secs(1), writer_rx.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(stored.user_friendly_text(), "!rustlog optout secret-code"); + assert!(bot.app.optout_codes.contains("secret-code")); + assert!(!bot.app.config.opt_out.contains_key("200")); + assert!(timeout(Duration::from_millis(50), writer_rx.recv()) + .await + .is_err()); + server.abort(); + } + + #[tokio::test] + async fn replay_does_not_suppress_a_later_live_command() { + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get(|| async { Json(serde_json::json!({"messages": [REPLAYED_COMMAND]})) }), + ); + let (mut bot, mut writer_rx, server) = bot_with_server(router).await; + let config_path = + std::env::temp_dir().join(format!("rustlog-command-{}.json", uuid::Uuid::new_v4())); + std::fs::write( + &config_path, + serde_json::to_vec(bot.app.config.as_ref()).unwrap(), + ) + .unwrap(); + bot.app.config = Arc::new(Config::load(&config_path).unwrap()); + bot.fetch_recent_messages_for_channel("channelone", "1") + .await + .unwrap(); + assert!(bot.app.optout_codes.contains("secret-code")); + let (_, client) = super::TwitchClient::new(twitch_irc::ClientConfig::new_simple( + twitch_irc::login::StaticLoginCredentials::anonymous(), + )); + let live = ServerMessage::try_from(IRCMessage::parse(REPLAYED_COMMAND).unwrap()).unwrap(); + bot.handle_message(live.clone(), &client, None) + .await + .unwrap(); + assert!(!bot.app.optout_codes.contains("secret-code")); + assert!(Config::load(&config_path) + .unwrap() + .opt_out + .contains_key("200")); + bot.app.optout_codes.insert("secret-code".to_string()); + bot.handle_message(live, &client, None).await.unwrap(); + assert!(bot.app.optout_codes.contains("secret-code")); + assert!(writer_rx.try_recv().is_ok()); + assert!(writer_rx.try_recv().is_err()); + std::fs::remove_file(config_path).unwrap(); + server.abort(); + } + + #[tokio::test] + async fn replay_rechecks_channel_after_waiting_for_writer_capacity() { + let (mut bot, _, server) = bot_with_server(Router::new()).await; + let (writer_tx, mut writer_rx) = mpsc::channel(1); + bot.writer_tx = writer_tx; + let prepared = bot + .prepare_message(IRCMessage::parse(UUID_MESSAGE).unwrap(), UUID_MESSAGE) + .unwrap() + .unwrap(); + bot.writer_tx.send(prepared.message.clone()).await.unwrap(); + let pending = bot.write_prepared_for_channel(prepared, true); + tokio::pin!(pending); + assert!(futures::poll!(&mut pending).is_pending()); + bot.app.config.channels.write().unwrap().remove("1"); + writer_rx.recv().await.unwrap(); + pending.await.unwrap(); + assert!(writer_rx.try_recv().is_err()); + server.abort(); + } + + #[tokio::test] + async fn opted_out_message_does_not_wait_for_writer_capacity() { + let (mut bot, _writer_rx, server) = bot_with_server(Router::new()).await; + let (writer_tx, _writer_rx) = mpsc::channel(1); + bot.writer_tx = writer_tx; + let prepared = bot + .prepare_message(IRCMessage::parse(UUID_MESSAGE).unwrap(), UUID_MESSAGE) + .unwrap() + .unwrap(); + bot.writer_tx.send(prepared.message.clone()).await.unwrap(); + bot.app.config.opt_out.insert("200".to_string(), true); + timeout(Duration::from_secs(1), bot.write_prepared(prepared)) + .await + .unwrap() + .unwrap(); + server.abort(); + } + + #[tokio::test] + async fn removed_channel_is_not_replayed_or_requested_again() { + let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let requests = Arc::new(AtomicUsize::new(0)); + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get({ + let started = started.clone(); + let release = release.clone(); + let requests = requests.clone(); + move || { + let started = started.clone(); + let release = release.clone(); + let requests = requests.clone(); + async move { + requests.fetch_add(1, Ordering::SeqCst); + started.notify_one(); + release.notified().await; + Json(serde_json::json!({"messages": [UUID_MESSAGE]})) + } + } + }), + ); + let (bot, mut writer_rx, server) = bot_with_server(router).await; + let replay = tokio::spawn({ + let bot = bot.clone(); + async move { + bot.fetch_recent_messages_for_channel("channelone", "1") + .await + } + }); + timeout(Duration::from_secs(1), started.notified()) + .await + .unwrap(); + bot.app.config.channels.write().unwrap().remove("1"); + release.notify_one(); + replay.await.unwrap().unwrap(); + bot.fetch_recent_messages_for_channel("channelone", "1") + .await + .unwrap(); + assert!(writer_rx.try_recv().is_err()); + assert_eq!(requests.load(Ordering::SeqCst), 1); + server.abort(); + } + + #[tokio::test] + async fn normalizes_valid_items_and_rejects_untrusted_robotty_metadata() { + let valid = format!( + " \r\n\0{}\0\n ", + ROBOTTY_VALUELESS_MESSAGE.replace("#channelone", "#ChAnNeLoNe") + ); + let wrong_login = UUID_MESSAGE + .replace("#channelone", "#otherchannel") + .replace( + "272e342c-5864-4c59-b730-25908cdb7f57", + "372e342c-5864-4c59-b730-25908cdb7f57", + ); + let wrong_room_id = UUID_MESSAGE.replace("room-id=1", "room-id=2").replace( + "272e342c-5864-4c59-b730-25908cdb7f57", + "472e342c-5864-4c59-b730-25908cdb7f57", + ); + let missing_timestamp = UUID_MESSAGE + .replace("tmi-sent-ts=1704067200000;", "") + .replace( + "272e342c-5864-4c59-b730-25908cdb7f57", + "572e342c-5864-4c59-b730-25908cdb7f57", + ); + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get(move || { + let messages = vec![ + valid.clone(), + wrong_login.clone(), + wrong_room_id.clone(), + missing_timestamp.clone(), + "not raw irc".to_string(), + ]; + async move { + Json(serde_json::json!({ + "messages": messages, + "error": null, + "error_code": null + })) + } + }), + ); + let (bot, mut writer_rx, server) = bot_with_server(router).await; + + bot.fetch_recent_messages_for_channel(" \0#CHANNELONE\r\n", "1") + .await + .unwrap(); + + let stored = timeout(Duration::from_secs(1), writer_rx.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(stored.channel_login, "channelone"); + assert_eq!(stored.channel_id, "1"); + assert_eq!(stored.timestamp, 1_704_067_200_000); + assert_eq!( + stored.id().as_deref(), + Some("272e342c-5864-4c59-b730-25908cdb7f57") + ); + assert_eq!(stored.color, Some(0x00ff7f)); + assert!(timeout(Duration::from_millis(50), writer_rx.recv()) + .await + .is_err()); + server.abort(); + } + + #[tokio::test] + async fn robotty_items_are_stored_in_timestamp_order() { + let later = UUID_MESSAGE + .replace("1704067200000", "1704067202000") + .replace( + "272e342c-5864-4c59-b730-25908cdb7f57", + "672e342c-5864-4c59-b730-25908cdb7f57", + ); + let earlier = UUID_MESSAGE + .replace("1704067200000", "1704067201000") + .replace( + "272e342c-5864-4c59-b730-25908cdb7f57", + "772e342c-5864-4c59-b730-25908cdb7f57", + ); + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get(move || { + let messages = vec![later.clone(), earlier.clone()]; + async move { + Json(serde_json::json!({ + "messages": messages, + "error": null, + "error_code": null + })) + } + }), + ); + let (bot, mut writer_rx, server) = bot_with_server(router).await; + + bot.fetch_recent_messages_for_channel("channelone", "1") + .await + .unwrap(); + + assert_eq!(writer_rx.recv().await.unwrap().timestamp, 1_704_067_201_000); + assert_eq!(writer_rx.recv().await.unwrap().timestamp, 1_704_067_202_000); + server.abort(); + } + + #[tokio::test] + async fn robotty_requests_are_spaced_and_concurrency_is_bounded() { + let requests = Arc::new(AtomicUsize::new(0)); + let active = Arc::new(AtomicUsize::new(0)); + let max_active = Arc::new(AtomicUsize::new(0)); + let starts = Arc::new(StdMutex::new(Vec::new())); + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get({ + let requests = requests.clone(); + let active = active.clone(); + let max_active = max_active.clone(); + let starts = starts.clone(); + move || { + let requests = requests.clone(); + let active = active.clone(); + let max_active = max_active.clone(); + let starts = starts.clone(); + async move { + requests.fetch_add(1, Ordering::SeqCst); + let active_now = active.fetch_add(1, Ordering::SeqCst) + 1; + max_active.fetch_max(active_now, Ordering::SeqCst); + starts.lock().unwrap().push(StdInstant::now()); + tokio::time::sleep(Duration::from_millis(75)).await; + active.fetch_sub(1, Ordering::SeqCst); + Json(serde_json::json!({ + "messages": [], + "error": null, + "error_code": null + })) + } + } + }), + ); + let (mut bot, _writer_rx, server) = bot_with_server(router).await; + bot.recent_message_request_spacing = Duration::from_millis(25); + bot.recent_message_request_slots = Arc::new(tokio::sync::Semaphore::new(2)); + + for index in 0..8 { + let channel_id = (index + 10).to_string(); + let channel_login = format!("channel{index}"); + bot.app + .config + .channels + .write() + .unwrap() + .insert(channel_id.clone()); + bot.app.users.insert(channel_id, channel_login.clone()); + bot.trigger_recent_messages_fetch(channel_login, "test", BackfillSchedule::OneShot) + .await; + } + wait_for_backfills(&bot).await; + + let mut starts = starts.lock().unwrap().clone(); + starts.sort_unstable(); + assert_eq!(requests.load(Ordering::SeqCst), 8); + assert_eq!(max_active.load(Ordering::SeqCst), 2); + assert!(starts.last().unwrap().duration_since(starts[0]) >= Duration::from_millis(140)); + server.abort(); + } + + #[tokio::test] + async fn successful_own_joins_trigger_backfill_and_suppress_concurrent_fetches() { + let requests = Arc::new(AtomicUsize::new(0)); + let handler_requests = requests.clone(); + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get(move || { + let requests = handler_requests.clone(); + async move { + requests.fetch_add(1, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(50)).await; + Json(serde_json::json!({ + "messages": [], + "error": null, + "error_code": null + })) + } + }), + ); + let (mut bot, _writer_rx, server) = bot_with_server(router).await; + bot.startup_backfill_delays = vec![Duration::ZERO; 5].into(); + bot.reconnect_backfill_delays = vec![Duration::ZERO; 6].into(); + let own_join = ServerMessage::try_from( + IRCMessage::parse( + ":justinfan12345!justinfan12345@justinfan12345.tmi.twitch.tv JOIN #channelone", + ) + .unwrap(), + ) + .unwrap(); + let other_join = ServerMessage::try_from( + IRCMessage::parse(":viewer!viewer@viewer.tmi.twitch.tv JOIN #channelone").unwrap(), + ) + .unwrap(); + + bot.trigger_backfill_for_own_join(&own_join, "justinfan12345") + .await; + wait_for_backfills(&bot).await; + assert_eq!(requests.load(Ordering::SeqCst), 5); + + bot.trigger_backfill_for_own_join(&own_join, "justinfan12345") + .await; + bot.trigger_backfill_for_own_join(&own_join, "justinfan12345") + .await; + bot.trigger_backfill_for_own_join(&other_join, "justinfan12345") + .await; + wait_for_backfills(&bot).await; + let after_coalesced_reconnect = requests.load(Ordering::SeqCst); + assert!((11..=12).contains(&after_coalesced_reconnect)); + + bot.trigger_backfill_for_own_join(&own_join, "justinfan12345") + .await; + wait_for_backfills(&bot).await; + assert_eq!( + requests.load(Ordering::SeqCst), + after_coalesced_reconnect + 6 + ); + server.abort(); + } + + #[tokio::test] + async fn reconnect_wakes_delayed_recovery_and_releases_channel_state() { + let requests = Arc::new(AtomicUsize::new(0)); + let handler_requests = requests.clone(); + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get(move || { + let requests = handler_requests.clone(); + async move { + requests.fetch_add(1, Ordering::SeqCst); + Json(serde_json::json!({"messages": [], "error": null, "error_code": null})) + } + }), + ); + let (mut bot, _writer_rx, server) = bot_with_server(router).await; + bot.startup_backfill_delays = [Duration::from_secs(3600)].into(); + bot.reconnect_backfill_delays = [Duration::ZERO].into(); + + bot.trigger_recent_messages_fetch( + "channelone".to_owned(), + "test", + BackfillSchedule::Startup, + ) + .await; + tokio::task::yield_now().await; + bot.trigger_recent_messages_fetch( + "channelone".to_owned(), + "test", + BackfillSchedule::Reconnect, + ) + .await; + assert_eq!(bot.backfill_tasks.lock().await.len(), 1); + timeout(Duration::from_secs(1), wait_for_backfills(&bot)) + .await + .unwrap(); + assert_eq!(requests.load(Ordering::SeqCst), 1); + assert!(bot.backfill_states.lock().await.is_empty()); + + // A join after completion starts a fresh task instead of stranding a restart. + bot.trigger_recent_messages_fetch( + "channelone".to_owned(), + "test", + BackfillSchedule::Reconnect, + ) + .await; + timeout(Duration::from_secs(1), wait_for_backfills(&bot)) + .await + .unwrap(); + assert_eq!(requests.load(Ordering::SeqCst), 2); + assert!(bot.backfill_states.lock().await.is_empty()); + server.abort(); + } + + #[tokio::test] + async fn reconnect_during_final_request_restarts_the_existing_task() { + let requests = Arc::new(AtomicUsize::new(0)); + let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get({ + let requests = requests.clone(); + let started = started.clone(); + let release = release.clone(); + move || { + let requests = requests.clone(); + let started = started.clone(); + let release = release.clone(); + async move { + if requests.fetch_add(1, Ordering::SeqCst) == 0 { + started.notify_one(); + release.notified().await; + } + Json(serde_json::json!({"messages": [], "error": null, "error_code": null})) + } + } + }), + ); + let (mut bot, _writer_rx, server) = bot_with_server(router).await; + bot.startup_backfill_delays = [Duration::ZERO].into(); + bot.reconnect_backfill_delays = [Duration::ZERO].into(); + bot.trigger_recent_messages_fetch( + "channelone".to_owned(), + "test", + BackfillSchedule::Startup, + ) + .await; + timeout(Duration::from_secs(1), started.notified()) + .await + .unwrap(); + bot.trigger_recent_messages_fetch( + "channelone".to_owned(), + "test", + BackfillSchedule::Reconnect, + ) + .await; + assert_eq!(bot.backfill_tasks.lock().await.len(), 1); + release.notify_one(); + timeout(Duration::from_secs(1), wait_for_backfills(&bot)) + .await + .unwrap(); + + assert_eq!(requests.load(Ordering::SeqCst), 2); + assert!(bot.backfill_states.lock().await.is_empty()); + server.abort(); + } + + #[tokio::test] + async fn stored_id_lookup_failure_prevents_backfill_insert() { + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get(|| async { + Json(serde_json::json!({ + "messages": [UUID_MESSAGE], + "error": null, + "error_code": null + })) + }), + ); + let (mut bot, mut writer_rx, server) = bot_with_server(router).await; + bot.skip_existing_message_lookup = false; + + assert!(bot + .fetch_recent_messages_for_channel("channelone", "1") + .await + .is_err()); + assert!(writer_rx.try_recv().is_err()); + + server.abort(); + } + + #[tokio::test] + async fn one_channel_failure_does_not_stop_another_backfill() { + let working_message = REPLAYED_COMMAND.replace("#channelone", "#working"); + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get(move |Path(channel): Path| { + let working_message = working_message.clone(); + async move { + if channel == "broken" { + StatusCode::BAD_GATEWAY.into_response() + } else { + Json(serde_json::json!({ + "messages": [working_message], + "error": null, + "error_code": null + })) + .into_response() + } + } + }), + ); + let (bot, mut writer_rx, server) = bot_with_server(router).await; + bot.app.users.insert("1".to_string(), "working".to_string()); + bot.app.users.insert("2".to_string(), "broken".to_string()); + bot.app + .config + .channels + .write() + .unwrap() + .insert("2".to_string()); + + bot.trigger_recent_messages_fetch("broken".to_string(), "test", BackfillSchedule::OneShot) + .await; + bot.trigger_recent_messages_fetch("working".to_string(), "test", BackfillSchedule::OneShot) + .await; + wait_for_backfills(&bot).await; + + assert!(writer_rx.recv().await.is_some()); + server.abort(); + } + + #[tokio::test] + async fn replay_dedupes_against_a_live_message_and_skips_remote_errors() { + let router = Router::new().route( + "/api/v2/recent-messages/{channel}", + get(|Path(channel): Path| async move { + if channel == "remoteerror" { + Json(serde_json::json!({ + "messages": [REPLAYED_COMMAND], + "error": "temporarily unavailable", + "error_code": "503" + })) + } else { + Json(serde_json::json!({ + "messages": [REPLAYED_COMMAND], + "error": null, + "error_code": null + })) + } + }), + ); + let (bot, mut writer_rx, server) = bot_with_server(router).await; + let message = IRCMessage::parse(REPLAYED_COMMAND).unwrap(); + let prepared = bot + .prepare_message(message, REPLAYED_COMMAND) + .unwrap() + .unwrap(); + assert!(bot.claim_message(&prepared.dedupe_key)); + bot.write_prepared(prepared).await.unwrap(); + + bot.fetch_recent_messages_for_channel("channelone", "1") + .await + .unwrap(); + bot.fetch_recent_messages_for_channel("remoteerror", "1") + .await + .unwrap(); + + assert!(writer_rx.recv().await.is_some()); + assert!(timeout(Duration::from_millis(50), writer_rx.recv()) + .await + .is_err()); + server.abort(); + } + + #[test] + fn event_identity_ignores_extra_tag_order_without_changing_stored_tags() { + let parse = |raw: &str| { + crate::db::schema::StructuredMessage::from_unstructured( + &crate::db::schema::UnstructuredMessage { + channel_id: "1", + user_id: "200", + timestamp: 1_704_067_200_000, + raw, + }, + ) + .unwrap() + .into_owned() + }; + let live = parse("@room-id=1;target-user-id=200;ban-duration=60;tmi-sent-ts=1704067200000 :tmi.twitch.tv CLEARCHAT #channelone :user"); + let replay = parse("@ban-duration=60;tmi-sent-ts=1704067200000;room-id=1;target-user-id=200 :tmi.twitch.tv CLEARCHAT #channelone :user"); + let different_duration = parse("@ban-duration=120;tmi-sent-ts=1704067200000;room-id=1;target-user-id=200 :tmi.twitch.tv CLEARCHAT #channelone :user"); + let original_tags = live.extra_tags.clone(); + + assert_ne!(live.extra_tags, replay.extra_tags); + assert_eq!( + structured_event_identity(&live), + structured_event_identity(&replay) + ); + assert_ne!( + structured_event_identity(&live), + structured_event_identity(&different_duration) + ); + assert_eq!(live.extra_tags, original_tags); + } + + #[test] + fn event_identity_uses_uuid_and_structured_hash_fallback() { + let with_id = crate::db::schema::StructuredMessage::from_unstructured( + &crate::db::schema::UnstructuredMessage { + channel_id: "1", + user_id: "200", + timestamp: 1_704_067_200_000, + raw: UUID_MESSAGE, + }, + ) + .unwrap(); + assert_eq!( + structured_event_identity(&with_id), + "uuid:272e342c-5864-4c59-b730-25908cdb7f57" + ); + + let without_id = crate::db::schema::StructuredMessage::from_unstructured( + &crate::db::schema::UnstructuredMessage { + channel_id: "1", + user_id: "200", + timestamp: 1_704_067_200_000, + raw: REPLAYED_COMMAND, + }, + ) + .unwrap(); + let identity = structured_event_identity(&without_id); + assert!(identity.starts_with("row:")); + assert_eq!(identity, structured_event_identity(&without_id)); + } +} diff --git a/src/config.rs b/src/config.rs index c270e80..6b7b703 100644 --- a/src/config.rs +++ b/src/config.rs @@ -5,6 +5,10 @@ use std::fs; use std::{collections::HashSet, sync::RwLock}; use tracing::info; +pub(crate) const DEFAULT_RECENT_MESSAGES_URL: &str = + "https://recent-messages.robotty.de/api/v2/recent-messages"; +pub(crate) const DEFAULT_RECENT_MESSAGES_LIMIT: usize = 800; + #[derive(Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct Config { @@ -16,6 +20,12 @@ pub struct Config { pub clickhouse_flush_interval: u64, #[serde(default = "default_listen_address")] pub listen_address: String, + #[serde(default)] + pub recent_messages_enabled: bool, + #[serde(default = "default_recent_messages_url")] + pub recent_messages_url: String, + #[serde(default = "default_recent_messages_limit")] + pub recent_messages_limit: usize, pub channels: RwLock>, #[serde(rename = "clientID")] pub client_id: String, @@ -41,7 +51,12 @@ impl Config { pub fn save(&self) -> anyhow::Result<()> { info!("Updating config"); let json = serde_json::to_string_pretty(self)?; - fs::write(self.config_path.as_ref().expect("config path should always be available"), json)?; + fs::write( + self.config_path + .as_ref() + .expect("config path should always be available"), + json, + )?; Ok(()) } @@ -54,3 +69,59 @@ fn default_listen_address() -> String { fn clickhouse_flush_interval() -> u64 { 10 } + +fn default_recent_messages_url() -> String { + DEFAULT_RECENT_MESSAGES_URL.to_string() +} + +fn default_recent_messages_limit() -> usize { + DEFAULT_RECENT_MESSAGES_LIMIT +} + +#[cfg(test)] +mod tests { + use super::{Config, DEFAULT_RECENT_MESSAGES_LIMIT, DEFAULT_RECENT_MESSAGES_URL}; + use serde_json::json; + + fn base_config() -> serde_json::Value { + json!({ + "clickhouseUrl": "http://clickhouse:8123", + "clickhouseDb": "rustlog", + "clickhouseUsername": "rustlog", + "clickhousePassword": "password", + "channels": [], + "clientID": "client-id", + "clientSecret": "client-secret", + "admins": [], + "optOut": {}, + "adminAPIKey": "admin-key" + }) + } + + #[test] + fn recent_messages_defaults_match_documented_values() { + let config: Config = serde_json::from_value(base_config()).unwrap(); + + assert!(!config.recent_messages_enabled); + assert_eq!(config.recent_messages_url, DEFAULT_RECENT_MESSAGES_URL); + assert_eq!(config.recent_messages_limit, DEFAULT_RECENT_MESSAGES_LIMIT); + } + + #[test] + fn recent_messages_fields_round_trip() { + let mut value = base_config(); + value["recentMessagesEnabled"] = json!(true); + value["recentMessagesUrl"] = json!("https://example.test/recent"); + value["recentMessagesLimit"] = json!(25); + + let config: Config = serde_json::from_value(value).unwrap(); + let serialized = serde_json::to_value(&config).unwrap(); + + assert_eq!(serialized["recentMessagesEnabled"], json!(true)); + assert_eq!( + serialized["recentMessagesUrl"], + json!("https://example.test/recent") + ); + assert_eq!(serialized["recentMessagesLimit"], json!(25)); + } +} diff --git a/src/logs/extract.rs b/src/logs/extract.rs index 96ed197..823e538 100644 --- a/src/logs/extract.rs +++ b/src/logs/extract.rs @@ -5,12 +5,28 @@ pub trait MessageWithTags { fn get_tag(&self, key: Tag) -> Option<&str>; } +trait TagValue { + fn as_optional_str(&self) -> Option<&str>; +} + +impl TagValue for String { + fn as_optional_str(&self) -> Option<&str> { + Some(self) + } +} + +impl TagValue for Option { + fn as_optional_str(&self) -> Option<&str> { + self.as_deref() + } +} + impl MessageWithTags for IRCMessage { fn get_tag(&self, key: Tag) -> Option<&str> { self.tags .0 .get(key.as_str()) - .and_then(|value| value.as_deref()) + .and_then(TagValue::as_optional_str) } } diff --git a/src/main.rs b/src/main.rs index 8aac305..81f527e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,6 +6,7 @@ mod db; mod error; mod logs; mod migrator; +mod recent_messages; mod web; pub type Result = std::result::Result; @@ -45,8 +46,14 @@ const SHUTDOWN_TIMEOUT_SECONDS: u64 = 8; #[global_allocator] static GLOBAL: MiMalloc = MiMalloc; +fn install_rustls_crypto_provider() { + let _ = rustls::crypto::ring::default_provider().install_default(); +} + #[tokio::main] async fn main() -> anyhow::Result<()> { + install_rustls_crypto_provider(); + let use_ansi = env::var("RUST_LOG_ANSI") .ok() .and_then(|ansi| ansi.parse().ok()) diff --git a/src/recent_messages.rs b/src/recent_messages.rs new file mode 100644 index 0000000..cc0725d --- /dev/null +++ b/src/recent_messages.rs @@ -0,0 +1,274 @@ +use crate::config::{Config, DEFAULT_RECENT_MESSAGES_LIMIT}; +use anyhow::{anyhow, Context, Result}; +use reqwest::Url; +use serde::Deserialize; +use std::time::Duration; +use tracing::warn; + +pub(crate) const RECENT_MESSAGES_TIMEOUT: Duration = Duration::from_secs(10); + +#[derive(Clone)] +pub struct RecentMessagesClient { + enabled: bool, + base_url: Option, + limit: usize, + client: reqwest::Client, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RecentMessagesRuntimeSummary { + pub enabled: bool, + pub base_url: Option, + pub limit: usize, + pub warnings: Vec, +} + +#[derive(Debug, Deserialize)] +pub struct RecentMessagesResponse { + pub messages: Vec, + pub error: Option, + pub error_code: Option, +} + +impl RecentMessagesClient { + pub fn from_config(config: &Config) -> Self { + Self::from_summary(Self::summary_from_config(config)) + } + + pub fn from_summary(summary: RecentMessagesRuntimeSummary) -> Self { + crate::install_rustls_crypto_provider(); + + for warning in &summary.warnings { + warn!("{warning}"); + } + + let base_url = summary + .base_url + .as_deref() + .and_then(|url| Url::parse(url).ok()); + Self { + enabled: summary.enabled && base_url.is_some(), + base_url, + limit: summary.limit, + client: reqwest::Client::builder() + .timeout(RECENT_MESSAGES_TIMEOUT) + .build() + .expect("failed to build recent-messages HTTP client"), + } + } + + pub fn summary_from_config(config: &Config) -> RecentMessagesRuntimeSummary { + let mut warnings = Vec::new(); + let base_url = match normalize_base_url(&config.recent_messages_url) { + Ok(url) => Some(url), + Err(error) => { + warnings.push(format!( + "recentMessagesUrl is invalid; recent-message backfill is disabled: {error}" + )); + None + } + }; + let limit = (config.recent_messages_limit > 0) + .then_some(config.recent_messages_limit) + .unwrap_or(DEFAULT_RECENT_MESSAGES_LIMIT); + + if config.recent_messages_enabled && base_url.is_none() { + warnings.push( + "recentMessagesEnabled is set but recentMessagesUrl is unusable; recent-message backfill is disabled".to_string(), + ); + } + + RecentMessagesRuntimeSummary { + enabled: config.recent_messages_enabled, + base_url, + limit, + warnings, + } + } + + pub fn enabled(&self) -> bool { + self.enabled + } + + pub async fn fetch(&self, channel_login: &str) -> Result { + let mut url = self + .base_url + .clone() + .context("recent-message backfill is disabled")?; + url.path_segments_mut() + .map_err(|_| anyhow!("recent-messages URL cannot contain path segments"))? + .push(&channel_login.to_ascii_lowercase()); + url.query_pairs_mut() + .append_pair("limit", &self.limit.to_string()); + + self.client + .get(url) + .send() + .await + .context("recent-messages request failed")? + .error_for_status() + .context("recent-messages request returned an error status")? + .json::() + .await + .context("recent-messages response was not valid JSON") + } +} + +fn normalize_base_url(value: &str) -> Result { + let trimmed = value.trim().trim_end_matches('/'); + if trimmed.is_empty() { + return Err("empty URL".to_string()); + } + + let parsed = Url::parse(trimmed).map_err(|error| error.to_string())?; + match parsed.scheme() { + "http" | "https" => Ok(parsed.to_string().trim_end_matches('/').to_string()), + scheme => Err(format!("unsupported scheme {scheme}")), + } +} + +#[cfg(test)] +mod tests { + use super::{RecentMessagesClient, RecentMessagesRuntimeSummary}; + use crate::config::{Config, DEFAULT_RECENT_MESSAGES_LIMIT, DEFAULT_RECENT_MESSAGES_URL}; + use axum::{ + extract::{Path, Query}, + http::StatusCode, + response::IntoResponse, + routing::get, + Json, Router, + }; + use std::collections::HashMap; + use tokio::net::TcpListener; + + fn config(enabled: bool, url: &str, limit: usize) -> Config { + serde_json::from_value(serde_json::json!({ + "clickhouseUrl": "http://clickhouse:8123", + "clickhouseDb": "rustlog", + "clickhouseUsername": "rustlog", + "clickhousePassword": "password", + "channels": [], + "clientID": "client-id", + "clientSecret": "client-secret", + "admins": [], + "optOut": {}, + "adminAPIKey": "admin-key", + "recentMessagesEnabled": enabled, + "recentMessagesUrl": url, + "recentMessagesLimit": limit + })) + .unwrap() + } + + #[test] + fn defaults_to_disabled_with_robotty_url() { + let config: Config = serde_json::from_value(serde_json::json!({ + "clickhouseUrl": "http://clickhouse:8123", + "clickhouseDb": "rustlog", + "channels": [], + "clientID": "client-id", + "clientSecret": "client-secret", + "admins": [], + "optOut": {} + })) + .unwrap(); + let summary = RecentMessagesClient::summary_from_config(&config); + + assert!(!summary.enabled); + assert_eq!( + summary.base_url.as_deref(), + Some(DEFAULT_RECENT_MESSAGES_URL) + ); + assert_eq!(summary.limit, DEFAULT_RECENT_MESSAGES_LIMIT); + assert!(summary.warnings.is_empty()); + } + + #[test] + fn normalizes_configured_url_and_limit() { + let config = config(true, " https://example.test/api/v2/recent-messages/ ", 25); + let summary = RecentMessagesClient::summary_from_config(&config); + + assert!(summary.enabled); + assert_eq!( + summary.base_url.as_deref(), + Some("https://example.test/api/v2/recent-messages") + ); + assert_eq!(summary.limit, 25); + } + + #[test] + fn rejects_non_http_url_and_defaults_invalid_limit() { + let config = config(true, "file:///tmp/messages", 0); + let summary = RecentMessagesClient::summary_from_config(&config); + + assert!(summary.enabled); + assert!(summary.base_url.is_none()); + assert_eq!(summary.limit, DEFAULT_RECENT_MESSAGES_LIMIT); + assert!(!summary.warnings.is_empty()); + assert!(!RecentMessagesClient::from_summary(summary).enabled()); + } + + #[tokio::test] + async fn fetch_uses_lowercase_channel_and_configured_limit() { + async fn handler( + Path(channel): Path, + Query(query): Query>, + ) -> impl IntoResponse { + assert_eq!(channel, "channelone"); + assert_eq!(query.get("limit").map(String::as_str), Some("25")); + Json(serde_json::json!({ + "messages": ["raw message"], + "error": null, + "error_code": null + })) + } + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve( + listener, + Router::new().route("/api/v2/recent-messages/{channel}", get(handler)), + ) + .await + }); + let client = RecentMessagesClient::from_summary(RecentMessagesRuntimeSummary { + enabled: true, + base_url: Some(format!("http://{address}/api/v2/recent-messages")), + limit: 25, + warnings: Vec::new(), + }); + + let response = client.fetch("ChannelOne").await.unwrap(); + assert_eq!(response.messages, ["raw message"]); + server.abort(); + } + + #[tokio::test] + async fn fetch_rejects_error_status_and_malformed_json() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve( + listener, + Router::new() + .route( + "/api/v2/recent-messages/status", + get(|| async { StatusCode::BAD_GATEWAY }), + ) + .route("/api/v2/recent-messages/json", get(|| async { "not json" })), + ) + .await + }); + let client = RecentMessagesClient::from_summary(RecentMessagesRuntimeSummary { + enabled: true, + base_url: Some(format!("http://{address}/api/v2/recent-messages")), + limit: 800, + warnings: Vec::new(), + }); + + assert!(client.fetch("status").await.is_err()); + assert!(client.fetch("json").await.is_err()); + server.abort(); + } +}