diff --git a/Cargo.lock b/Cargo.lock index 7805c84..8974e3a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -771,6 +771,12 @@ dependencies = [ "objc2", ] +[[package]] +name = "bnum" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "119771309b95163ec7aaf79810da82f7cd0599c19722d48b9c03894dca833966" + [[package]] name = "bon" version = "3.8.1" @@ -796,12 +802,6 @@ dependencies = [ "syn 2.0.111", ] -[[package]] -name = "borrow-or-share" -version = "0.2.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc0b364ead1874514c8c2855ab558056ebfeb775653e7ae45ff72f28f8f3166c" - [[package]] name = "borsh" version = "1.6.0" @@ -1096,7 +1096,7 @@ dependencies = [ "multihash", "serde", "serde_bytes", - "unsigned-varint", + "unsigned-varint 0.8.0", ] [[package]] @@ -1111,6 +1111,12 @@ dependencies = [ "zeroize", ] +[[package]] +name = "cityhash-rs" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93a719913643003b84bd13022b4b7e703c09342cd03b679c4641c7d2e50dc34d" + [[package]] name = "clap" version = "4.5.53" @@ -1153,6 +1159,57 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a1d728cc89cf3aee9ff92b05e62b19ee65a02b5702cff7d5a377e32c6ae29d8d" +[[package]] +name = "clickhouse" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4916807143285e80eefcaf741a29a98bbd687e4581352631a74ab223e66558b" +dependencies = [ + "bnum", + "bstr", + "bytes", + "chrono", + "cityhash-rs", + "clickhouse-macros", + "clickhouse-types", + "futures-channel", + "futures-util", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "lz4_flex", + "polonius-the-crab", + "quanta", + "rustls", + "serde", + "thiserror 2.0.17", + "tokio", + "url", +] + +[[package]] +name = "clickhouse-macros" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff6669899e23cb87b43daf7996f0ea3b9c07d0fb933d745bb7b815b052515ae3" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn 2.0.111", +] + +[[package]] +name = "clickhouse-types" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "358fbfd439fb0bed02a3e2ecc5131f6a9d039ba5639aed650cf0e845f6ebfc16" +dependencies = [ + "bytes", + "thiserror 2.0.17", +] + [[package]] name = "cobs" version = "0.3.0" @@ -1618,7 +1675,7 @@ dependencies = [ "crypto_secretbox", "curve25519-dalek 5.0.0-pre.1", "salsa20", - "serdect 0.4.1", + "serdect", "subtle", "zeroize", ] @@ -1840,24 +1897,6 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be1e0bca6c3637f992fc1cc7cbc52a78c1ef6db076dbf1059c4323d6a2048376" -[[package]] -name = "deadpool" -version = "0.12.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b" -dependencies = [ - "deadpool-runtime", - "lazy_static", - "num_cpus", - "tokio", -] - -[[package]] -name = "deadpool-runtime" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b" - [[package]] name = "deflate" version = "1.0.0" @@ -1974,68 +2013,6 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ab03c107fafeb3ee9f5925686dbb7a73bc76e3932abb0d2b365cb64b169cf04c" -[[package]] -name = "diesel" -version = "2.3.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c415189028b232660655e4893e8bc25ca7aee8e96888db66d9edb400535456a" -dependencies = [ - "chrono", - "diesel_derives", - "downcast-rs 2.0.2", - "libsqlite3-sys", - "serde_json", - "sqlite-wasm-rs", - "time", -] - -[[package]] -name = "diesel-async" -version = "0.7.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "13096fb8dae53f2d411c4b523bec85f45552ed3044a2ab4d85fb2092d9cb4f34" -dependencies = [ - "deadpool", - "diesel", - "futures-core", - "futures-util", - "scoped-futures", - "tokio", -] - -[[package]] -name = "diesel_derives" -version = "2.3.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8587cbca3c929fb198e7950d761d31ca72b80aa6e07c1b7bec5879d187720436" -dependencies = [ - "diesel_table_macro_syntax", - "dsl_auto_type", - "proc-macro2", - "quote", - "syn 2.0.111", -] - -[[package]] -name = "diesel_migrations" -version = "2.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "745fd255645f0f1135f9ec55c7b00e0882192af9683ab4731e4bba3da82b8f9c" -dependencies = [ - "diesel", - "migrations_internals", - "migrations_macros", -] - -[[package]] -name = "diesel_table_macro_syntax" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fe2444076b48641147115697648dc743c2c00b61adade0f01ce67133c7babe8c" -dependencies = [ - "syn 2.0.111", -] - [[package]] name = "diff" version = "0.1.13" @@ -3232,32 +3209,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2" -[[package]] -name = "downcast-rs" -version = "2.0.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "117240f60069e65410b3ae1bb213295bd828f707b5bec6596a1afc8793ce0cbc" - [[package]] name = "dpi" version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d8b14ccef22fc6f5a8f4d7d768562a182c04ce9a3b3157b91390b52ddfdf1a76" -[[package]] -name = "dsl_auto_type" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dd122633e4bef06db27737f21d3738fb89c8f6d5360d6d9d7635dda142a7757e" -dependencies = [ - "darling 0.21.3", - "either", - "heck 0.5.0", - "proc-macro2", - "quote", - "syn 2.0.111", -] - [[package]] name = "dtoa" version = "1.0.10" @@ -3305,6 +3262,7 @@ version = "2.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53" dependencies = [ + "pkcs8 0.10.2", "signature 2.2.0", ] @@ -3328,6 +3286,7 @@ dependencies = [ "curve25519-dalek 4.1.3", "ed25519 2.2.3", "rand_core 0.6.4", + "serde", "sha2 0.10.9", "subtle", "zeroize", @@ -3362,7 +3321,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47" dependencies = [ "base16ct 0.2.0", - "base64ct", "crypto-bigint", "digest 0.10.7", "ff", @@ -3372,8 +3330,6 @@ dependencies = [ "pkcs8 0.10.2", "rand_core 0.6.4", "sec1", - "serde_json", - "serdect 0.2.0", "subtle", "zeroize", ] @@ -3640,16 +3596,6 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "98de4bbd547a563b716d8dfa9aad1cb19bfab00f4fa09a6a4ed21dbcf44ce9c4" -[[package]] -name = "fluent-uri" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1918b65d96df47d3591bed19c5cca17e3fa5d0707318e4b5ef2eae01764df7e5" -dependencies = [ - "borrow-or-share", - "ref-cast", -] - [[package]] name = "fnv" version = "1.0.7" @@ -4026,7 +3972,6 @@ version = "0.14.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2" dependencies = [ - "serde", "typenum", "version_check", "zeroize", @@ -4473,10 +4418,7 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ - "allocator-api2", - "equivalent", "foldhash 0.1.5", - "serde", ] [[package]] @@ -4685,6 +4627,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "higher-kinded-types" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e690f8474c6c5d8ff99656fcbc195a215acc3949481a8b0b3351c838972dc776" +dependencies = [ + "macro_rules_attribute", + "never-say-never", + "paste", +] + [[package]] name = "hmac" version = "0.12.1" @@ -5265,6 +5218,22 @@ dependencies = [ "zeroize_derive", ] +[[package]] +name = "iroh-car" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f8cd4cb9aa083fba8b52e921764252d0b4dcb1cd6d120b809dbfe1106e81a" +dependencies = [ + "anyhow", + "cid", + "futures", + "serde", + "serde_ipld_dagcbor", + "thiserror 1.0.69", + "tokio", + "unsigned-varint 0.7.2", +] + [[package]] name = "iroh-gossip" version = "0.95.0" @@ -5592,6 +5561,7 @@ dependencies = [ "chrono", "ciborium", "cid", + "ed25519-dalek 2.2.0", "futures", "getrandom 0.2.16", "getrandom 0.3.4", @@ -5724,6 +5694,31 @@ dependencies = [ "webbrowser", ] +[[package]] +name = "jacquard-repo" +version = "0.9.4" +dependencies = [ + "bytes", + "cid", + "ed25519-dalek 2.2.0", + "iroh-car", + "jacquard-common", + "jacquard-derive", + "k256", + "miette 7.6.0", + "multihash", + "n0-future 0.1.3", + "p256", + "serde", + "serde_bytes", + "serde_ipld_dagcbor", + "sha2 0.10.9", + "smol_str", + "thiserror 2.0.17", + "tokio", + "trait-variant", +] + [[package]] name = "javascriptcore-rs" version = "1.1.2" @@ -5779,39 +5774,6 @@ dependencies = [ "libc", ] -[[package]] -name = "jose" -version = "0.0.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "de791f8b277ef903d49b0c6f3f3b866569ae8cb1d0829825eea28374cc4c803c" -dependencies = [ - "base64ct", - "cfg-if", - "digest 0.10.7", - "ecdsa", - "ed25519-dalek 2.2.0", - "elliptic-curve", - "fluent-uri", - "generic-array", - "hashbrown 0.15.5", - "hmac", - "k256", - "mediatype", - "p256", - "p384", - "rand_core 0.6.4", - "rsa", - "secrecy", - "serde", - "serde-value", - "serde_json", - "sha2 0.10.9", - "signature 2.2.0", - "subtle", - "thiserror 2.0.17", - "zeroize", -] - [[package]] name = "jose-b64" version = "0.1.2" @@ -5867,7 +5829,9 @@ dependencies = [ "cfg-if", "ecdsa", "elliptic-curve", + "once_cell", "sha2 0.10.9", + "signature 2.2.0", ] [[package]] @@ -6020,17 +5984,6 @@ dependencies = [ "redox_syscall", ] -[[package]] -name = "libsqlite3-sys" -version = "0.35.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "133c182a6a2c87864fe97778797e46c7e999672690dc9fa3ee8e241aa4a9c13f" -dependencies = [ - "cc", - "pkg-config", - "vcpkg", -] - [[package]] name = "libxdo" version = "0.6.0" @@ -6313,6 +6266,22 @@ dependencies = [ "syn 2.0.111", ] +[[package]] +name = "macro_rules_attribute" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65049d7923698040cd0b1ddcced9b0eb14dd22c5f86ae59c3740eab64a676520" +dependencies = [ + "macro_rules_attribute-proc_macro", + "paste", +] + +[[package]] +name = "macro_rules_attribute-proc_macro" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "670fdfda89751bc4a84ac13eaa63e205cf0fd22b4c9a5fbfa085b63c1f1d3a30" + [[package]] name = "malloc_buf" version = "0.0.6" @@ -6500,15 +6469,6 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "490cc448043f947bae3cbee9c203358d62dbee0db12107a74be5c30ccfd09771" -[[package]] -name = "mediatype" -version = "0.19.20" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33746aadcb41349ec291e7f2f0a3aa6834d1d7c58066fb4b01f68efc4c4b7631" -dependencies = [ - "serde", -] - [[package]] name = "memchr" version = "2.7.6" @@ -6568,7 +6528,6 @@ dependencies = [ "supports-color", "supports-hyperlinks", "supports-unicode", - "syntect", "terminal_size", "textwrap", "unicode-width 0.1.14", @@ -6596,27 +6555,6 @@ dependencies = [ "syn 2.0.111", ] -[[package]] -name = "migrations_internals" -version = "2.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "36c791ecdf977c99f45f23280405d7723727470f6689a5e6dbf513ac547ae10d" -dependencies = [ - "serde", - "toml 0.9.8", -] - -[[package]] -name = "migrations_macros" -version = "2.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "36fc5ac76be324cfd2d3f2cf0fdf5d5d3c4f14ed8aaebadb09e304ba42282703" -dependencies = [ - "migrations_internals", - "proc-macro2", - "quote", -] - [[package]] name = "mime" version = "0.3.17" @@ -6670,25 +6608,6 @@ dependencies = [ "web-time", ] -[[package]] -name = "minijinja" -version = "2.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0adbe6e92a6ce0fd6c4aac593fdfd3e3950b0f61b1a63aa9731eb6fd85776fa3" -dependencies = [ - "serde", -] - -[[package]] -name = "minijinja-contrib" -version = "2.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a915f5cc17b954d5252b6373cc5fd97eb86d75e29b30ac2ee7932126451eef24" -dependencies = [ - "minijinja", - "serde", -] - [[package]] name = "minimal-lexical" version = "0.2.1" @@ -6792,7 +6711,7 @@ checksum = "6b430e7953c29dd6a09afc29ff0bb69c6e306329ee6794700aee27b76a1aea8d" dependencies = [ "core2", "serde", - "unsigned-varint", + "unsigned-varint 0.8.0", ] [[package]] @@ -7034,6 +6953,12 @@ dependencies = [ "wmi", ] +[[package]] +name = "never-say-never" +version = "6.6.666" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf5a574dadd7941adeaa71823ecba5e28331b8313fb2e1c6a5c7e5981ea53ad6" + [[package]] name = "new_debug_unreachable" version = "1.0.6" @@ -7386,28 +7311,6 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" -[[package]] -name = "onig" -version = "6.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "336b9c63443aceef14bea841b899035ae3abe89b7c486aaf4c5bd8aafedac3f0" -dependencies = [ - "bitflags 2.10.0", - "libc", - "once_cell", - "onig_sys", -] - -[[package]] -name = "onig_sys" -version = "69.9.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c7f86c6eef3d6df15f23bcfb6af487cbd2fed4e5581d58d5bf1f5f8b7f6727dc" -dependencies = [ - "cc", - "pkg-config", -] - [[package]] name = "openssl" version = "0.10.75" @@ -7458,15 +7361,6 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" -[[package]] -name = "ordered-float" -version = "2.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68f19d67e5a2795c94e73e0bb1cc1a7edeb2e28efd39e2e1c9b7a40c1108b11c" -dependencies = [ - "num-traits", -] - [[package]] name = "ordered-stream" version = "0.2.0" @@ -7525,10 +7419,8 @@ version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fe42f1670a52a47d448f14b6a5c61dd78fce51856e68edaa38f7ae3a46b8d6b6" dependencies = [ - "ecdsa", "elliptic-curve", "primeorder", - "sha2 0.10.9", ] [[package]] @@ -7951,6 +7843,16 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2f3a9f18d041e6d0e102a0a46750538147e5e8992d3b4873aaafee2520b00ce3" +[[package]] +name = "polonius-the-crab" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec242d7eccbb2fd8b3b5b6e3cf89f94a91a800f469005b44d154359609f8af72" +dependencies = [ + "higher-kinded-types", + "never-say-never", +] + [[package]] name = "poly1305" version = "0.9.0-rc.2" @@ -8196,6 +8098,21 @@ dependencies = [ "bumpalo", ] +[[package]] +name = "quanta" +version = "0.12.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3ab5a9d756f0d97bdc89019bd2e4ea098cf9cde50ee7564dde6b81ccc8f06c7" +dependencies = [ + "crossbeam-utils", + "libc", + "once_cell", + "raw-cpuid", + "wasi 0.11.1+wasi-snapshot-preview1", + "web-sys", + "winapi", +] + [[package]] name = "quick-error" version = "1.2.3" @@ -8433,6 +8350,15 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d20581732dd76fa913c7dff1a2412b714afe3573e94d41c34719de73337cc8ab" +[[package]] +name = "raw-cpuid" +version = "11.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "498cd0dc59d73224351ee52a95fee0f1a617a2eae0e7d9d720cc622c73a54186" +dependencies = [ + "bitflags 2.10.0", +] + [[package]] name = "raw-window-handle" version = "0.5.2" @@ -8901,15 +8827,6 @@ dependencies = [ "serde_json", ] -[[package]] -name = "scoped-futures" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b24aae2d0636530f359e9d5ef0c04669d11c5e756699b27a6a6d845d8329091" -dependencies = [ - "pin-project-lite", -] - [[package]] name = "scoped-tls" version = "1.0.1" @@ -8932,21 +8849,10 @@ dependencies = [ "der 0.7.10", "generic-array", "pkcs8 0.10.2", - "serdect 0.2.0", "subtle", "zeroize", ] -[[package]] -name = "secrecy" -version = "0.10.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e891af845473308773346dc847b2c23ee78fe442e0472ac50e22a18a93d3ae5a" -dependencies = [ - "serde", - "zeroize", -] - [[package]] name = "security-framework" version = "2.11.1" @@ -9038,16 +8944,6 @@ dependencies = [ "serde_derive", ] -[[package]] -name = "serde-value" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3a1a3341211875ef120e117ea7fd5228530ae7e7036a779fdc9117be6b3282c" -dependencies = [ - "ordered-float", - "serde", -] - [[package]] name = "serde-wasm-bindgen" version = "0.4.5" @@ -9136,6 +9032,17 @@ dependencies = [ "syn 2.0.111", ] +[[package]] +name = "serde_derive_internals" +version = "0.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.111", +] + [[package]] name = "serde_html_form" version = "0.2.8" @@ -9228,15 +9135,6 @@ dependencies = [ "serde", ] -[[package]] -name = "serde_spanned" -version = "1.0.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e24345aa0fe688594e73770a5f6d1b216508b4f93484c0026d521acd30134392" -dependencies = [ - "serde_core", -] - [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -9280,16 +9178,6 @@ dependencies = [ "syn 2.0.111", ] -[[package]] -name = "serdect" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a84f14a19e9a014bb9f4512488d9829a68e04ecabffb0f9904cd1ace94598177" -dependencies = [ - "base16ct 0.2.0", - "serde", -] - [[package]] name = "serdect" version = "0.4.1" @@ -9693,21 +9581,6 @@ dependencies = [ "der 0.8.0-rc.10", ] -[[package]] -name = "sqlite-wasm-rs" -version = "0.4.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "60bdd87fcb4c9764b024805fb2df5f1d659bea6e629fdbdcdcfc4042b9a640d0" -dependencies = [ - "js-sys", - "once_cell", - "thiserror 2.0.17", - "tokio", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -10113,7 +9986,6 @@ dependencies = [ "flate2", "fnv", "once_cell", - "onig", "plist", "regex-syntax", "serde", @@ -10154,7 +10026,7 @@ dependencies = [ "cfg-expr", "heck 0.5.0", "pkg-config", - "toml 0.8.23", + "toml", "version-compare", ] @@ -10575,26 +10447,12 @@ version = "0.8.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362" dependencies = [ - "indexmap 2.12.1", "serde", - "serde_spanned 0.6.9", + "serde_spanned", "toml_datetime 0.6.11", "toml_edit 0.22.27", ] -[[package]] -name = "toml" -version = "0.9.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0dc8b1fb61449e27716ec0e1bdf0f6b8f3e8f6b05391e8497b8b6d7804ea6d8" -dependencies = [ - "serde_core", - "serde_spanned 1.0.3", - "toml_datetime 0.7.3", - "toml_parser", - "winnow 0.7.14", -] - [[package]] name = "toml_datetime" version = "0.6.11" @@ -10643,9 +10501,8 @@ checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" dependencies = [ "indexmap 2.12.1", "serde", - "serde_spanned 0.6.9", + "serde_spanned", "toml_datetime 0.6.11", - "toml_write", "winnow 0.7.14", ] @@ -10670,12 +10527,6 @@ dependencies = [ "winnow 0.7.14", ] -[[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" @@ -10692,22 +10543,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "tower-cookies" -version = "0.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "151b5a3e3c45df17466454bb74e9ecedecc955269bdedbf4d150dfa393b55a36" -dependencies = [ - "axum-core", - "cookie", - "futures-util", - "http", - "parking_lot", - "pin-project-lite", - "tower-layer", - "tower-service", -] - [[package]] name = "tower-http" version = "0.6.8" @@ -10734,7 +10569,6 @@ dependencies = [ "tower-layer", "tower-service", "tracing", - "uuid", ] [[package]] @@ -10761,19 +10595,6 @@ dependencies = [ "tracing-core", ] -[[package]] -name = "tracing-appender" -version = "0.2.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "786d480bce6247ab75f005b14ae1624ad978d3029d9113f0a22fa1ac773faeaf" -dependencies = [ - "crossbeam-channel", - "parking_lot", - "thiserror 2.0.17", - "time", - "tracing-subscriber", -] - [[package]] name = "tracing-attributes" version = "0.1.31" @@ -10816,16 +10637,6 @@ dependencies = [ "tracing-core", ] -[[package]] -name = "tracing-serde" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" -dependencies = [ - "serde", - "tracing-core", -] - [[package]] name = "tracing-subscriber" version = "0.3.22" @@ -10836,15 +10647,12 @@ dependencies = [ "nu-ansi-term", "once_cell", "regex-automata", - "serde", - "serde_json", "sharded-slab", "smallvec", "thread_local", "tracing", "tracing-core", "tracing-log", - "tracing-serde", ] [[package]] @@ -11136,6 +10944,12 @@ dependencies = [ "subtle", ] +[[package]] +name = "unsigned-varint" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6889a77d49f1f013504cec6bf97a2c730394adedaeb1deb5ea08949a50541105" + [[package]] name = "unsigned-varint" version = "0.8.0" @@ -11441,7 +11255,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "673a33c33048a5ade91a6b139580fa174e19fb0d23f396dca9fa15f2e1e49b35" dependencies = [ "cc", - "downcast-rs 1.2.1", + "downcast-rs", "rustix", "scoped-tls", "smallvec", @@ -11643,47 +11457,30 @@ dependencies = [ name = "weaver-index" version = "0.1.0" dependencies = [ - "axum", + "base64 0.22.1", + "bytes", "chrono", + "ciborium", + "cid", "clap", - "dashmap 6.1.0", - "diesel", - "diesel-async", - "diesel_migrations", + "clickhouse", "dotenvy", - "futures-util", - "hyper", + "humansize", "jacquard", - "jacquard-axum", - "jose", - "jose-jwk", - "js-sys", - "libsqlite3-sys", + "jacquard-common", + "jacquard-repo", "miette 7.6.0", - "minijinja", - "minijinja-contrib", - "pin-project", - "reqwest", - "send_wrapper", + "n0-future 0.1.3", "serde", + "serde_ipld_dagcbor", "serde_json", + "smol_str", "thiserror 2.0.17", "tokio", - "toml 0.8.23", - "tower", - "tower-cookies", - "tower-http", "tracing", - "tracing-appender", "tracing-subscriber", "url", - "uuid", - "wasm-bindgen", - "wasm-bindgen-futures", - "wasmworker", - "wasmworker-proc-macro", - "weaver-common", - "web-time", + "weaver-api", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index a78abb3..df2e9b8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -47,6 +47,7 @@ jacquard-common = { path = "../jacquard/crates/jacquard-common" } jacquard-axum = {path = "../jacquard/crates/jacquard-axum" } jacquard-derive = { path = "../jacquard/crates/jacquard-derive" } jacquard-lexicon = { path = "../jacquard/crates/jacquard-lexicon", default-features = false } +jacquard-repo = { path = "../jacquard/crates/jacquard-repo" } # jacquard = { path = "../jacquard-facet/crates/jacquard", default-features = false, features = ["derive", "api_bluesky", "tracing", "serde"] } # jacquard-identity = { path = "../jacquard-facet/crates/jacquard-identity", features = ["cache"] } diff --git a/crates/weaver-index/Cargo.toml b/crates/weaver-index/Cargo.toml index d9b0477..dde2787 100644 --- a/crates/weaver-index/Cargo.toml +++ b/crates/weaver-index/Cargo.toml @@ -4,75 +4,63 @@ version.workspace = true edition.workspace = true license.workspace = true publish = false +default-run = "indexer" [features] default = [] -embedded-db = ["diesel/sqlite", "diesel-async/sync-connection-wrapper"] -[dependencies] -weaver-common = { path = "../weaver-common", default-features = false } -clap = { version = "4.5", features = ["derive", "env", "cargo", "unicode"] } -dotenvy = "0.15" +[[bin]] +name = "storage-benchmark" +path = "src/bin/storage_benchmark.rs" + +[[bin]] +name = "indexer" +path = "src/bin/weaver_indexer.rs" -tracing = "0.1.40" -tracing-subscriber = { version = "0.3.18", features = ["json", "env-filter"] } -tracing-appender = { version = "0.2.3", features = ["parking_lot"] } -chrono = "0.4.41" -serde = { version = "1.0.210", features = ["derive"] } -serde_json = "1.0.128" -uuid = { version = "1.8.0", features = ["v7", "serde"] } +[dependencies] +# Internal +weaver-api = { path = "../weaver-api", features = ["streaming"] } +# AT Protocol / Jacquard jacquard = { workspace = true, features = ["websocket", "zstd"] } -jacquard-axum = { workspace = true } +jacquard-common = { workspace = true } +jacquard-repo = { workspace = true } -axum = "0.8" -hyper = "1.4.1" -tower = { version = "0.5", features = [] } -tower-cookies = "0.11" -tower-http = { version = "0.6", features = [ -"fs", - "trace", - "timeout", - "request-id", - "cors", - "normalize-path", -] } +# ClickHouse +clickhouse = { version = "0.14", features = ["inserter", "chrono", "rustls-tls-ring", "rustls-tls-webpki-roots"] } -reqwest = { version = "0.12.7", default-features = false, features = [ - "json", - "rustls-tls", -] } +# Serialization +serde = { workspace = true } +serde_json = "1.0" +serde_ipld_dagcbor = "0.6" +ciborium = "0.2" +# Async runtime tokio = { version = "1.44", features = ["full"] } +n0-future = { workspace = true } -miette = { workspace = true, features = ["fancy", "syntect-highlighter"] } +# Error handling +miette = { workspace = true, features = ["fancy"] } thiserror = { workspace = true } -diesel = { version = "2.3.0", features = ["sqlite", "serde_json", "chrono", "returning_clauses_for_sqlite_3_35"] } -diesel-async = { version = "0.7", features = ["sqlite", "deadpool"] } -diesel_migrations = "2.3.0" +# Config / CLI +clap = { version = "4.5", features = ["derive", "env"] } +dotenvy = "0.15" -libsqlite3-sys = {version = ">=0.30.1,<0.36.0", features = ["bundled"] } +# Tracing +tracing = { workspace = true } +tracing-subscriber = { version = "0.3", features = ["env-filter"] } -toml = { version = "0.8.22", features = ["preserve_order"] } -jose = { version = "0.0.2", features = ["crypto-rustcrypto"] } -jose-jwk = "0.1.2" +# Data types +bytes = { workspace = true } url = "2" +chrono = { version = "0.4", features = ["serde"] } +smol_str = "0.3" -minijinja = { version = "2.9.0" } -minijinja-contrib = { version = "2.9.0" } -dashmap = "6.1.0" - +# CID handling (for CAR block lookups) +cid = "0.11" -# wasm-in-browser dependencies -[target.'cfg(all(target_family = "wasm", target_os = "unknown"))'.dependencies] -futures-util = "0.3" -js-sys = "0.3" -pin-project = "1" -wasm-bindgen = "0.2" -wasm-bindgen-futures = "0.4" -web-time = "1" -send_wrapper = "0.6" -wasmworker = "0.1" -wasmworker-proc-macro = "0.1" +# Utilities +humansize = "2.0" +base64 = "0.22" diff --git a/crates/weaver-index/appview-config.toml b/crates/weaver-index/appview-config.toml deleted file mode 100644 index e69de29..0000000 diff --git a/crates/weaver-index/build.rs b/crates/weaver-index/build.rs deleted file mode 100644 index 3a8149e..0000000 --- a/crates/weaver-index/build.rs +++ /dev/null @@ -1,3 +0,0 @@ -fn main() { - println!("cargo:rerun-if-changed=migrations"); -} diff --git a/crates/weaver-index/diesel.toml b/crates/weaver-index/diesel.toml deleted file mode 100644 index 609961e..0000000 --- a/crates/weaver-index/diesel.toml +++ /dev/null @@ -1,11 +0,0 @@ -# For documentation on how to configure this file, -# see https://diesel.rs/guides/configuring-diesel-cli - -[print_schema] -file = "src/schema.rs" -custom_type_derives = ["diesel::query_builder::QueryId", "Clone", "std::fmt::Debug", "jacquard::IntoStatic"] -patch_file = "src/schema.patch" - - -[migrations_directory] -dir = "/home/orual/Projects/weaver.sh/crates/weaver-index/migrations" diff --git a/crates/weaver-index/migrations/.keep b/crates/weaver-index/migrations/.keep deleted file mode 100644 index e69de29..0000000 diff --git a/crates/weaver-index/migrations/2025-05-15-230036_create_appviewdb/down.sql b/crates/weaver-index/migrations/2025-05-15-230036_create_appviewdb/down.sql deleted file mode 100644 index e002360..0000000 --- a/crates/weaver-index/migrations/2025-05-15-230036_create_appviewdb/down.sql +++ /dev/null @@ -1,28 +0,0 @@ --- This file should undo anything in `up.sql` -drop index if exists idx_oauth_auth_requests_did; - -drop index if exists idx_oauth_auth_requests_expires; - -drop index if exists idx_oauth_sessions_did; - -drop index if exists idx_oauth_sessions_did_session; - -drop table if exists oauth_auth_requests; - -drop table if exists oauth_sessions; - -drop table if exists profile_pronouns; - -drop table if exists profile_links; - -drop table if exists profile; - -drop table if exists emails; - -drop table if exists _jetstream; - -drop table if exists follows; - -drop table if exists public_keys; - -drop table if exists registrations; diff --git a/crates/weaver-index/migrations/2025-05-15-230036_create_appviewdb/up.sql b/crates/weaver-index/migrations/2025-05-15-230036_create_appviewdb/up.sql deleted file mode 100644 index 3d7d207..0000000 --- a/crates/weaver-index/migrations/2025-05-15-230036_create_appviewdb/up.sql +++ /dev/null @@ -1,118 +0,0 @@ -create table if not exists registrations ( - id integer not null primary key autoincrement, - domain text not null unique, - did text not null, - secret text not null, - created timestamp not null default (datetime('now')), - registered text -); - -create table if not exists public_keys ( - id integer not null primary key autoincrement, - did text not null, - name text not null, - key_contents text not null, - rkey text not null, - created timestamp not null default (datetime('now')), - unique (did, name, key_contents) -); - -create table if not exists follows ( - user_did text not null, - subject_did text not null, - rkey text not null, - followed_at timestamp not null default (datetime('now')), - primary key (user_did, subject_did), - check (user_did <> subject_did) -); - -create table if not exists _jetstream ( - id integer not null primary key autoincrement, - last_time_us integer not null -); - -create table if not exists emails ( - id integer not null primary key autoincrement, - did text not null, - email text not null, - verified boolean not null default false, - verification_code text not null, - last_sent timestamp not null default (datetime('now')), - is_primary boolean not null default false, - created timestamp not null default (datetime('now')), - unique (did, email) -); - -create table if not exists profile ( - -- id - id integer not null primary key autoincrement, - did text not null, - -- data - avatar text, - description text not null, - include_bluesky boolean not null default false, - include_tangled boolean not null default false, - location text, - pinned_post text, - created_at timestamp default (datetime('now')), - -- constraints - unique (did) -); - -create table if not exists profile_links ( - -- id - id integer not null primary key autoincrement, - did text not null, - -- data - link text not null, - -- constraints - foreign key (did) references profile (did) on delete cascade -); - -create table if not exists profile_pronouns ( - -- id - id integer not null primary key autoincrement, - did text not null, - -- data - pronoun text not null, - -- constraints - foreign key (did) references profile (did) on delete cascade -); - --- OAuth sessions table for jacquard ClientSessionData -create table if not exists oauth_sessions ( - id integer not null primary key autoincrement, - -- Extracted from ClientSessionData for indexing - did text not null, - session_id text not null, - -- Full ClientSessionData as JSON - session_data blob not null, - created_at timestamp not null default (datetime('now')), - updated_at timestamp not null default (datetime('now')), - unique (did, session_id) -); - --- OAuth authorization requests table for jacquard AuthRequestData -create table if not exists oauth_auth_requests ( - id integer not null primary key autoincrement, - -- Extracted from AuthRequestData for indexing - state text not null unique, - -- Optional DID if known at auth request time - account_did text, - -- Full AuthRequestData as JSON - auth_req_data blob not null, - created_at timestamp not null default (datetime('now')), - expires_at timestamp not null default (datetime('now', '+10 minutes')) -); - --- Index for quick session lookups -create index if not exists idx_oauth_sessions_did_session on oauth_sessions(did, session_id); - --- Index for DID lookups -create index if not exists idx_oauth_sessions_did on oauth_sessions(did); - --- Index for auth request cleanup -create index if not exists idx_oauth_auth_requests_expires on oauth_auth_requests(expires_at); - --- Index for DID lookups in auth requests -create index if not exists idx_oauth_auth_requests_did on oauth_auth_requests(account_did) where account_did is not null; diff --git a/crates/weaver-index/migrations/clickhouse/000_migrations.sql b/crates/weaver-index/migrations/clickhouse/000_migrations.sql new file mode 100644 index 0000000..7981635 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/000_migrations.sql @@ -0,0 +1,12 @@ +-- Migration tracking table +-- Tracks which migrations have been applied + +CREATE TABLE IF NOT EXISTS _migrations ( + -- Migration filename (e.g., '001_raw_records.sql') + name String, + + -- When this migration was applied + applied_at DateTime64(3) DEFAULT now64(3) +) +ENGINE = MergeTree() +ORDER BY (name); diff --git a/crates/weaver-index/migrations/clickhouse/001_raw_records.sql b/crates/weaver-index/migrations/clickhouse/001_raw_records.sql new file mode 100644 index 0000000..2e1d6ce --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/001_raw_records.sql @@ -0,0 +1,35 @@ +-- Raw records from firehose/jetstream +-- Core table for all AT Protocol records before denormalization +-- +-- Uses ReplacingMergeTree to deduplicate on (collection, did, rkey) keeping latest indexed_at +-- JSON column stores full record, extract fields only when needed for ORDER BY/WHERE/JOINs + +CREATE TABLE IF NOT EXISTS raw_records ( + -- Decomposed AT URI components (at://did/collection/rkey) + did String, + collection LowCardinality(String), + rkey String, + + -- Content identifier from the record + cid String, + + -- Full record as native JSON (schema-flexible, queryable with record.field.subfield) + record JSON, + + -- Operation: 'create', 'update', 'delete' + operation LowCardinality(String), + + -- Firehose sequence number (metadata only, not for ordering - can jump on relay restart) + seq UInt64, + + -- Event timestamp from firehose + event_time DateTime64(3), + + -- When we indexed this record + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Materialized AT URI for convenience + uri String MATERIALIZED concat('at://', did, '/', collection, '/', rkey) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (collection, did, rkey, indexed_at); diff --git a/crates/weaver-index/migrations/clickhouse/002_identity_events.sql b/crates/weaver-index/migrations/clickhouse/002_identity_events.sql new file mode 100644 index 0000000..b42017e --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/002_identity_events.sql @@ -0,0 +1,21 @@ +-- Identity events from firehose (#identity messages) +-- Tracks handle changes, key rotation, etc. + +CREATE TABLE IF NOT EXISTS raw_identity_events ( + -- The DID this identity event is about + did String, + + -- Handle (may be empty if cleared) + handle String, + + -- Sequence number from firehose + seq UInt64, + + -- Event timestamp from firehose + event_time DateTime64(3), + + -- When we indexed this event + indexed_at DateTime64(3) DEFAULT now64(3) +) +ENGINE = MergeTree() +ORDER BY (did, indexed_at); diff --git a/crates/weaver-index/migrations/clickhouse/003_account_events.sql b/crates/weaver-index/migrations/clickhouse/003_account_events.sql new file mode 100644 index 0000000..e626a3f --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/003_account_events.sql @@ -0,0 +1,24 @@ +-- Account events from firehose (#account messages) +-- Tracks account status changes: active, deactivated, deleted, suspended, takendown + +CREATE TABLE IF NOT EXISTS raw_account_events ( + -- The DID this account event is about + did String, + + -- Whether the account is active + active UInt8, + + -- Account status: 'active', 'deactivated', 'deleted', 'suspended', 'takendown' + status LowCardinality(String), + + -- Sequence number from firehose + seq UInt64, + + -- Event timestamp from firehose + event_time DateTime64(3), + + -- When we indexed this event + indexed_at DateTime64(3) DEFAULT now64(3) +) +ENGINE = MergeTree() +ORDER BY (did, indexed_at); diff --git a/crates/weaver-index/migrations/clickhouse/004_events_dlq.sql b/crates/weaver-index/migrations/clickhouse/004_events_dlq.sql new file mode 100644 index 0000000..955751d --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/004_events_dlq.sql @@ -0,0 +1,21 @@ +-- Dead-letter queue for malformed events +-- Events that couldn't be parsed or processed land here for debugging + +CREATE TABLE IF NOT EXISTS raw_events_dlq ( + -- Event type we attempted to parse (if known) + event_type LowCardinality(String), + + -- Raw event data (JSON string of whatever we received) + raw_data String, + + -- Error message describing why parsing failed + error_message String, + + -- Sequence number from firehose (if available) + seq UInt64, + + -- When we received this event + received_at DateTime64(3) DEFAULT now64(3) +) +ENGINE = MergeTree() +ORDER BY (received_at); diff --git a/crates/weaver-index/migrations/clickhouse/005_firehose_cursor.sql b/crates/weaver-index/migrations/clickhouse/005_firehose_cursor.sql new file mode 100644 index 0000000..b37c9e4 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/005_firehose_cursor.sql @@ -0,0 +1,18 @@ +-- Firehose cursor persistence +-- Tracks our position in the firehose stream for resumption after restart + +CREATE TABLE IF NOT EXISTS firehose_cursor ( + -- Consumer identifier (allows multiple consumers with different cursors) + consumer_id String, + + -- Last successfully processed sequence number + seq UInt64, + + -- Timestamp of the last processed event + event_time DateTime64(3), + + -- When we saved this cursor + updated_at DateTime64(3) DEFAULT now64(3) +) +ENGINE = ReplacingMergeTree(updated_at) +ORDER BY (consumer_id); diff --git a/crates/weaver-index/src/api_error.rs b/crates/weaver-index/src/api_error.rs deleted file mode 100644 index 0b71e8f..0000000 --- a/crates/weaver-index/src/api_error.rs +++ /dev/null @@ -1,90 +0,0 @@ -use axum::{ - Json, - extract::rejection::JsonRejection, - response::{IntoResponse, Response}, -}; -use hyper::StatusCode; -use miette::Diagnostic; -use serde::{Deserialize, Serialize}; -use thiserror::Error; -use tracing::error; - -/// Custom error type for the API. -/// The `#[from]` attribute allows for easy conversion from other error types. -#[derive(Error, Debug, Diagnostic)] -pub enum ApiError { - /// Converts from an Axum built-in extractor error. - #[diagnostic_source] - #[error("Invalid payload.")] - InvalidJsonBody(#[from] JsonRejection), - - /// For errors that occur during manual validation. - #[error("Invalid request: {0}")] - #[diagnostic()] - InvalidRequest(String), - - /// Converts from `sqlx::Error`. - #[error("A database error has occurred.")] - #[diagnostic_source] - DatabaseError(#[from] diesel::result::Error), - - #[error("A Weaver error has occurred.")] - #[diagnostic(transparent)] - WeaverError(#[from] weaver_common::error::WeaverError), - /// Converts from any `anyhow::Error`. - #[error("An internal server error has occurred.")] - #[diagnostic(transparent)] - InternalError(miette::Report), -} - -impl From for ApiError { - fn from(err: miette::Report) -> Self { - ApiError::InternalError(err) - } -} - -#[derive(Serialize, Deserialize)] -pub struct ApiErrorResp { - pub message: String, -} - -// The IntoResponse implementation for ApiError logs the error message. -// -// To avoid exposing implementation details to API consumers, we separate -// the message that we log from the API response message. -impl IntoResponse for ApiError { - fn into_response(self) -> Response { - // Log detailed error for telemetry. - let error_to_log = match &self { - ApiError::InvalidJsonBody(err) => match err { - JsonRejection::JsonDataError(e) => e.body_text(), - JsonRejection::JsonSyntaxError(e) => e.body_text(), - JsonRejection::MissingJsonContentType(_) => { - "Missing `Content-Type: application/json` header".to_string() - } - JsonRejection::BytesRejection(_) => "Failed to buffer request body".to_string(), - _ => "Unknown error".to_string(), - }, - ApiError::InvalidRequest(_) => format!("{}", self), - ApiError::WeaverError(err) => format!("{}", err), - ApiError::DatabaseError(err) => format!("{}", err), - ApiError::InternalError(err) => format!("{}", err), - }; - error!("{}", error_to_log); - - // Create a generic response to hide specific implementation details. - let resp = ApiErrorResp { - message: self.to_string(), - }; - - // Determine the appropriate status code. - let status = match self { - ApiError::InvalidJsonBody(_) | ApiError::InvalidRequest(_) => StatusCode::BAD_REQUEST, - ApiError::WeaverError(_) | ApiError::DatabaseError(_) | ApiError::InternalError(_) => { - StatusCode::INTERNAL_SERVER_ERROR - } - }; - - (status, Json(resp)).into_response() - } -} diff --git a/crates/weaver-index/src/bin/storage_benchmark.rs b/crates/weaver-index/src/bin/storage_benchmark.rs new file mode 100644 index 0000000..fb11f15 --- /dev/null +++ b/crates/weaver-index/src/bin/storage_benchmark.rs @@ -0,0 +1,483 @@ +use bytes::Bytes; +use chrono::{DateTime, Utc}; +use clap::Parser; +use clickhouse::Row; +use n0_future::StreamExt; +use smol_str::SmolStr; +use std::time::{Duration, Instant}; +use tracing::{info, warn}; +use weaver_index::clickhouse::Client; +use weaver_index::config::{ClickHouseConfig, FirehoseConfig}; +use weaver_index::firehose::{FirehoseConsumer, SubscribeReposMessage, extract_records}; + +// ============================================================================= +// Benchmark-specific schema (not part of production) +// ============================================================================= + +const TABLE_JSON: &str = "raw_records_json"; +const TABLE_CBOR: &str = "raw_records_cbor"; + +/// Row type for JSON benchmark records +#[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] +struct RawRecordJson { + did: SmolStr, + collection: SmolStr, + rkey: SmolStr, + cid: String, + record: String, + operation: SmolStr, + seq: u64, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + event_time: DateTime, +} + +/// Row type for CBOR benchmark records +#[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] +struct RawRecordCbor { + did: SmolStr, + collection: SmolStr, + rkey: SmolStr, + cid: String, + #[serde(with = "jacquard::serde_bytes_helper")] + record: Bytes, + operation: SmolStr, + seq: u64, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + event_time: DateTime, +} + +async fn create_benchmark_tables(client: &Client) -> miette::Result<()> { + client + .execute(&format!( + r#" + CREATE TABLE IF NOT EXISTS {} ( + did String, + collection LowCardinality(String), + rkey String, + cid String, + record JSON, + operation LowCardinality(String), + seq UInt64, + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3) + ) + ENGINE = MergeTree() + ORDER BY (collection, did, rkey, indexed_at) + "#, + TABLE_JSON + )) + .await?; + + client + .execute(&format!( + r#" + CREATE TABLE IF NOT EXISTS {} ( + did String, + collection LowCardinality(String), + rkey String, + cid String, + record String, + operation LowCardinality(String), + seq UInt64, + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3) + ) + ENGINE = MergeTree() + ORDER BY (collection, did, rkey, indexed_at) + "#, + TABLE_CBOR + )) + .await?; + + Ok(()) +} + +async fn drop_benchmark_tables(client: &Client) -> miette::Result<()> { + client + .execute(&format!("DROP TABLE IF EXISTS {}", TABLE_JSON)) + .await?; + client + .execute(&format!("DROP TABLE IF EXISTS {}", TABLE_CBOR)) + .await?; + Ok(()) +} + +// ============================================================================= +// Benchmark logic +// ============================================================================= + +/// Tracks firehose lag to detect if we're falling behind +#[derive(Default)] +struct LagStats { + min_ms: Option, + max_ms: Option, + current_ms: i64, + sample_count: u64, +} + +impl LagStats { + fn update(&mut self, event_time_ms: i64) { + let now_ms = Utc::now().timestamp_millis(); + let lag = now_ms - event_time_ms; + + self.current_ms = lag; + self.sample_count += 1; + + self.min_ms = Some(self.min_ms.map_or(lag, |m| m.min(lag))); + self.max_ms = Some(self.max_ms.map_or(lag, |m| m.max(lag))); + } + + fn reset_window(&mut self) { + // Keep current but reset min/max for next reporting window + self.min_ms = Some(self.current_ms); + self.max_ms = Some(self.current_ms); + } +} + +#[derive(Parser)] +#[command(name = "storage-benchmark")] +#[command(about = "Benchmark CBOR vs JSON storage in ClickHouse")] +struct Args { + /// Duration to run the benchmark in minutes + #[arg(short, long, default_value = "60")] + duration_minutes: u64, + + /// Batch size for ClickHouse inserts + #[arg(short, long, default_value = "1000")] + batch_size: usize, + + /// Report interval in seconds + #[arg(short, long, default_value = "30")] + report_interval_secs: u64, + + /// Drop and recreate tables before starting + #[arg(long)] + reset_tables: bool, +} + +#[tokio::main] +async fn main() -> miette::Result<()> { + dotenvy::dotenv().ok(); + + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::from_default_env() + .add_directive("weaver_index=info".parse().unwrap()) + .add_directive("storage_benchmark=info".parse().unwrap()), + ) + .init(); + + let args = Args::parse(); + + info!("Storage Benchmark: CBOR vs JSON in ClickHouse"); + info!("Duration: {} minutes", args.duration_minutes); + info!("Batch size: {}", args.batch_size); + + // Load configs + let ch_config = ClickHouseConfig::from_env()?; + let firehose_config = FirehoseConfig::from_env()?; + + info!( + "Connecting to ClickHouse at {} (database: {})", + ch_config.url, ch_config.database + ); + let client = Client::new(&ch_config)?; + + // Reset tables if requested + if args.reset_tables { + info!("Dropping existing benchmark tables..."); + drop_benchmark_tables(&client).await?; + } + + // Create tables + info!("Creating benchmark tables..."); + create_benchmark_tables(&client).await?; + + // Create inserters + let mut json_inserter = client.inserter::(TABLE_JSON); + let mut cbor_inserter = client.inserter::(TABLE_CBOR); + + // Connect to firehose + info!("Connecting to firehose at {}", firehose_config.relay_url); + let consumer = FirehoseConsumer::new(firehose_config); + let mut stream = consumer.connect().await?; + + // Tracking + let start = Instant::now(); + let duration = Duration::from_secs(args.duration_minutes * 60); + let report_interval = Duration::from_secs(args.report_interval_secs); + let mut last_report = Instant::now(); + let mut total_records = 0u64; + let mut total_commits = 0u64; + let mut errors = 0u64; + let mut lag_stats = LagStats::default(); + + info!("Starting benchmark..."); + + while start.elapsed() < duration { + // Check for report interval + if last_report.elapsed() >= report_interval { + // Flush inserters so size measurements are accurate + match json_inserter.commit().await { + Ok(stats) => info!( + " JSON flush: {} rows, {} transactions", + stats.rows, stats.transactions + ), + Err(e) => warn!("Failed to flush JSON inserter: {}", e), + } + match cbor_inserter.commit().await { + Ok(stats) => info!( + " CBOR flush: {} rows, {} transactions", + stats.rows, stats.transactions + ), + Err(e) => warn!("Failed to flush CBOR inserter: {}", e), + } + + report_progress( + &client, + total_records, + total_commits, + errors, + start.elapsed(), + &lag_stats, + ) + .await; + lag_stats.reset_window(); + last_report = Instant::now(); + } + + // Get next message with timeout + let msg = tokio::time::timeout(Duration::from_secs(30), stream.next()).await; + + let msg = match msg { + Ok(Some(Ok(msg))) => msg, + Ok(Some(Err(e))) => { + warn!("Stream error: {}", e); + errors += 1; + continue; + } + Ok(None) => { + warn!("Stream ended unexpectedly"); + break; + } + Err(_) => { + warn!("Timeout waiting for message"); + continue; + } + }; + + // Only process commits + let commit = match msg { + SubscribeReposMessage::Commit(c) => c, + _ => continue, + }; + + total_commits += 1; + + // Track lag + lag_stats.update(commit.time.as_ref().timestamp_millis()); + + // Extract records from the commit + let records = match extract_records(&commit).await { + Ok(r) => r, + Err(e) => { + warn!("Record extraction error: {}", e); + errors += 1; + continue; + } + }; + + // Insert to both tables + for record in records { + // Skip deletes (no record data) + let Some(cbor_bytes) = &record.cbor_bytes else { + continue; + }; + + // JSON table: decode CBOR to JSON + let json_str = match record.to_json() { + Ok(Some(j)) => j, + Ok(None) => continue, + Err(e) => { + warn!("JSON encode error: {}", e); + errors += 1; + continue; + } + }; + + let event_time = DateTime::from_timestamp_millis(record.event_time_ms).unwrap(); + // Insert JSON record + json_inserter + .write(&RawRecordJson { + did: record.did.clone(), + collection: record.collection.clone(), + rkey: record.rkey.clone(), + cid: record.cid.clone(), + record: json_str, + operation: record.operation.clone(), + seq: record.seq as u64, + event_time: event_time.clone(), + }) + .await + .map_err(|e| weaver_index::error::ClickHouseError::Insert { + message: "json insert failed".into(), + source: e, + })?; + + // Insert CBOR record (raw bytes, no base64) + cbor_inserter + .write(&RawRecordCbor { + did: record.did, + collection: record.collection, + rkey: record.rkey, + cid: record.cid, + record: cbor_bytes.clone(), + operation: record.operation, + seq: record.seq as u64, + event_time, + }) + .await + .map_err(|e| weaver_index::error::ClickHouseError::Insert { + message: "cbor insert failed".into(), + source: e, + })?; + + match json_inserter.commit().await { + Ok(_) => {} + Err(e) => warn!("Failed to flush JSON inserter: {}", e), + } + match cbor_inserter.commit().await { + Ok(_) => {} + Err(e) => warn!("Failed to flush CBOR inserter: {}", e), + } + total_records += 1; + } + } + + // Final flush + info!("Flushing remaining records..."); + json_inserter + .end() + .await + .map_err(|e| weaver_index::error::ClickHouseError::Insert { + message: "json flush failed".into(), + source: e, + })?; + cbor_inserter + .end() + .await + .map_err(|e| weaver_index::error::ClickHouseError::Insert { + message: "cbor flush failed".into(), + source: e, + })?; + + // Final report + info!("\n========== FINAL RESULTS =========="); + report_progress( + &client, + total_records, + total_commits, + errors, + start.elapsed(), + &lag_stats, + ) + .await; + + // Detailed size comparison + info!("\nStorage Comparison:"); + let sizes = client.table_sizes(&[TABLE_JSON, TABLE_CBOR]).await?; + + for size in &sizes { + info!( + " {}: {} compressed, {} uncompressed, {:.2}x ratio, {} rows", + size.table, + size.compressed_human(), + size.uncompressed_human(), + size.compression_ratio(), + size.row_count + ); + } + + if sizes.len() == 2 { + let json_size = sizes.iter().find(|s| s.table == TABLE_JSON); + let cbor_size = sizes.iter().find(|s| s.table == TABLE_CBOR); + + if let (Some(json), Some(cbor)) = (json_size, cbor_size) { + let compressed_diff = json.compressed_bytes as f64 / cbor.compressed_bytes as f64; + let uncompressed_diff = json.uncompressed_bytes as f64 / cbor.uncompressed_bytes as f64; + + info!("\nJSON vs CBOR:"); + info!( + " Compressed: JSON is {:.2}x the size of CBOR", + compressed_diff + ); + info!( + " Uncompressed: JSON is {:.2}x the size of CBOR", + uncompressed_diff + ); + + if compressed_diff < 1.0 { + info!( + " Winner (compressed): JSON ({:.1}% smaller)", + (1.0 - compressed_diff) * 100.0 + ); + } else { + info!( + " Winner (compressed): CBOR ({:.1}% smaller)", + (1.0 - 1.0 / compressed_diff) * 100.0 + ); + } + } + } + + info!("\nBenchmark complete!"); + + Ok(()) +} + +async fn report_progress( + client: &Client, + total_records: u64, + total_commits: u64, + errors: u64, + elapsed: Duration, + lag: &LagStats, +) { + let records_per_sec = total_records as f64 / elapsed.as_secs_f64(); + + info!( + "Progress: {} records from {} commits in {:.1}s ({:.1}/s), {} errors", + total_records, + total_commits, + elapsed.as_secs_f64(), + records_per_sec, + errors + ); + + // Lag info - critical for detecting if we're falling behind + if lag.sample_count > 0 { + info!( + " Lag: current={:.1}s, min={:.1}s, max={:.1}s (window)", + lag.current_ms as f64 / 1000.0, + lag.min_ms.unwrap_or(0) as f64 / 1000.0, + lag.max_ms.unwrap_or(0) as f64 / 1000.0, + ); + } + + // Try to get current sizes + match client.table_sizes(&[TABLE_JSON, TABLE_CBOR]).await { + Ok(sizes) => { + for size in sizes { + info!( + " {}: {} compressed ({} rows)", + size.table, + size.compressed_human(), + size.row_count + ); + } + } + Err(e) => { + warn!("Failed to query table sizes: {}", e); + } + } +} diff --git a/crates/weaver-index/src/bin/weaver_indexer.rs b/crates/weaver-index/src/bin/weaver_indexer.rs new file mode 100644 index 0000000..ffff545 --- /dev/null +++ b/crates/weaver-index/src/bin/weaver_indexer.rs @@ -0,0 +1,98 @@ +use clap::{Parser, Subcommand}; +use tracing::info; +use weaver_index::clickhouse::{Client, Migrator}; +use weaver_index::config::ClickHouseConfig; + +#[derive(Parser)] +#[command(name = "weaver-indexer")] +#[command(about = "Weaver index service - firehose ingestion and query serving")] +struct Args { + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand)] +enum Command { + /// Run database migrations + Migrate { + /// Show what would be run without executing + #[arg(long)] + dry_run: bool, + }, + + /// Check database connectivity + Health, + + /// Start the indexer service (not yet implemented) + Run, +} + +#[tokio::main] +async fn main() -> miette::Result<()> { + dotenvy::dotenv().ok(); + + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::from_default_env() + .add_directive("weaver_index=info".parse().unwrap()) + .add_directive("weaver_indexer=info".parse().unwrap()), + ) + .init(); + + let args = Args::parse(); + + match args.command { + Command::Migrate { dry_run } => run_migrate(dry_run).await, + Command::Health => run_health().await, + Command::Run => run_indexer().await, + } +} + +async fn run_migrate(dry_run: bool) -> miette::Result<()> { + let config = ClickHouseConfig::from_env()?; + info!( + "Connecting to ClickHouse at {} (database: {})", + config.url, config.database + ); + + let client = Client::new(&config)?; + let migrator = Migrator::new(&client); + + if dry_run { + let pending = migrator.pending().await?; + if pending.is_empty() { + info!("No pending migrations"); + } else { + info!("Pending migrations:"); + for name in pending { + info!(" - {}", name); + } + } + } else { + let result = migrator.run().await?; + info!("{}", result); + } + + Ok(()) +} + +async fn run_health() -> miette::Result<()> { + let config = ClickHouseConfig::from_env()?; + info!( + "Connecting to ClickHouse at {} (database: {})", + config.url, config.database + ); + + let client = Client::new(&config)?; + + // Simple connectivity check + client.execute("SELECT 1").await?; + info!("ClickHouse connection OK"); + + Ok(()) +} + +async fn run_indexer() -> miette::Result<()> { + info!("Indexer not yet implemented"); + Ok(()) +} diff --git a/crates/weaver-index/src/clickhouse.rs b/crates/weaver-index/src/clickhouse.rs new file mode 100644 index 0000000..5451d6a --- /dev/null +++ b/crates/weaver-index/src/clickhouse.rs @@ -0,0 +1,9 @@ +mod client; +mod migrations; +mod schema; + +pub use client::{Client, TableSize}; +pub use migrations::{MigrationResult, Migrator}; +pub use schema::{ + FirehoseCursor, RawAccountEvent, RawEventDlq, RawIdentityEvent, RawRecord, Tables, +}; diff --git a/crates/weaver-index/src/clickhouse/client.rs b/crates/weaver-index/src/clickhouse/client.rs new file mode 100644 index 0000000..3c05bb7 --- /dev/null +++ b/crates/weaver-index/src/clickhouse/client.rs @@ -0,0 +1,119 @@ +use crate::config::ClickHouseConfig; +use crate::error::{ClickHouseError, IndexError}; +use clickhouse::Row; +use clickhouse::inserter::Inserter; + +/// ClickHouse client wrapper with connection pooling and batched inserts +pub struct Client { + inner: clickhouse::Client, +} + +impl Client { + /// Create a new client from configuration + pub fn new(config: &ClickHouseConfig) -> Result { + let inner = clickhouse::Client::default() + .with_url(config.url.as_str()) + .with_database(&config.database) + .with_user(&config.user) + .with_password(&config.password); + + Ok(Self { inner }) + } + + /// Execute a DDL query (CREATE TABLE, etc.) + pub async fn execute(&self, query: &str) -> Result<(), IndexError> { + self.inner + .query(query) + .execute() + .await + .map_err(|e| ClickHouseError::Query { + message: "DDL execution failed".into(), + source: e, + })?; + Ok(()) + } + + /// Create a batched inserter for a table + /// + /// The inserter accumulates rows and flushes them in batches for efficiency. + pub fn inserter(&self, table: &str) -> Inserter { + self.inner + .inserter(table) + .with_max_rows(1000) + .with_period_bias(0.1) + .with_max_bytes(1_048_576) + } + + /// Query table sizes from system.parts + /// + /// Returns (table_name, compressed_bytes, uncompressed_bytes, row_count) + pub async fn table_sizes(&self, tables: &[&str]) -> Result, IndexError> { + let table_list = tables + .iter() + .map(|t| format!("'{}'", t)) + .collect::>() + .join(", "); + + let query = format!( + r#" + SELECT + table, + sum(bytes_on_disk) as compressed_bytes, + sum(data_uncompressed_bytes) as uncompressed_bytes, + sum(rows) as row_count + FROM system.parts + WHERE table IN ({}) + AND active + GROUP BY table + "#, + table_list + ); + + let rows = self + .inner + .query(&query) + .fetch_all::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to query table sizes".into(), + source: e, + })?; + + Ok(rows) + } + + /// Get reference to inner client for advanced operations + pub fn inner(&self) -> &clickhouse::Client { + &self.inner + } +} + +/// Table size statistics from system.parts +#[derive(Debug, Clone, Row, serde::Deserialize)] +pub struct TableSize { + pub table: String, + pub compressed_bytes: u64, + pub uncompressed_bytes: u64, + pub row_count: u64, +} + +impl TableSize { + /// Format compressed size as human-readable string + pub fn compressed_human(&self) -> String { + humansize::format_size(self.compressed_bytes, humansize::BINARY) + } + + /// Format uncompressed size as human-readable string + pub fn uncompressed_human(&self) -> String { + humansize::format_size(self.uncompressed_bytes, humansize::BINARY) + } + + /// Compression ratio (uncompressed / compressed) + pub fn compression_ratio(&self) -> f64 { + if self.compressed_bytes == 0 { + 0.0 + } else { + self.uncompressed_bytes as f64 / self.compressed_bytes as f64 + } + } +} diff --git a/crates/weaver-index/src/clickhouse/migrations.rs b/crates/weaver-index/src/clickhouse/migrations.rs new file mode 100644 index 0000000..194b647 --- /dev/null +++ b/crates/weaver-index/src/clickhouse/migrations.rs @@ -0,0 +1,151 @@ +use crate::error::{ClickHouseError, IndexError}; +use tracing::info; + +use super::Client; + +/// Embedded migrations - compiled into the binary +const MIGRATIONS: &[(&str, &str)] = &[ + ( + "000_migrations.sql", + include_str!("../../migrations/clickhouse/000_migrations.sql"), + ), + ( + "001_raw_records.sql", + include_str!("../../migrations/clickhouse/001_raw_records.sql"), + ), + ( + "002_identity_events.sql", + include_str!("../../migrations/clickhouse/002_identity_events.sql"), + ), + ( + "003_account_events.sql", + include_str!("../../migrations/clickhouse/003_account_events.sql"), + ), + ( + "004_events_dlq.sql", + include_str!("../../migrations/clickhouse/004_events_dlq.sql"), + ), + ( + "005_firehose_cursor.sql", + include_str!("../../migrations/clickhouse/005_firehose_cursor.sql"), + ), +]; + +/// Migration runner for ClickHouse +pub struct Migrator<'a> { + client: &'a Client, +} + +impl<'a> Migrator<'a> { + pub fn new(client: &'a Client) -> Self { + Self { client } + } + + /// Run all pending migrations + pub async fn run(&self) -> Result { + // First, ensure the migrations table exists (bootstrap) + self.ensure_migrations_table().await?; + + // Get list of already applied migrations + let applied = self.get_applied_migrations().await?; + + let mut applied_count = 0; + let mut skipped_count = 0; + + for (name, sql) in MIGRATIONS { + // Skip the bootstrap migration after first run + if *name == "000_migrations.sql" && applied.contains(&"000_migrations.sql".to_string()) + { + skipped_count += 1; + continue; + } + + if applied.contains(&name.to_string()) { + info!(migration = %name, "already applied, skipping"); + skipped_count += 1; + continue; + } + + info!(migration = %name, "applying migration"); + self.client.execute(sql).await?; + self.record_migration(name).await?; + applied_count += 1; + } + + Ok(MigrationResult { + applied: applied_count, + skipped: skipped_count, + }) + } + + /// Check which migrations would be applied without running them + pub async fn pending(&self) -> Result, IndexError> { + // Try to get applied migrations, but if table doesn't exist, all are pending + let applied = match self.get_applied_migrations().await { + Ok(list) => list, + Err(_) => vec![], + }; + + let pending: Vec = MIGRATIONS + .iter() + .filter(|(name, _)| !applied.contains(&name.to_string())) + .map(|(name, _)| name.to_string()) + .collect(); + + Ok(pending) + } + + async fn ensure_migrations_table(&self) -> Result<(), IndexError> { + // Run the bootstrap migration directly + let (_, sql) = MIGRATIONS + .iter() + .find(|(name, _)| *name == "000_migrations.sql") + .expect("bootstrap migration must exist"); + + self.client.execute(sql).await + } + + async fn get_applied_migrations(&self) -> Result, IndexError> { + let rows: Vec = self + .client + .inner() + .query("SELECT name FROM _migrations ORDER BY name") + .fetch_all() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to fetch applied migrations".into(), + source: e, + })?; + + Ok(rows.into_iter().map(|r| r.name).collect()) + } + + async fn record_migration(&self, name: &str) -> Result<(), IndexError> { + let query = format!("INSERT INTO _migrations (name) VALUES ('{}')", name); + self.client.execute(&query).await + } +} + +#[derive(Debug, Clone, clickhouse::Row, serde::Deserialize)] +struct MigrationRow { + name: String, +} + +/// Result of running migrations +#[derive(Debug, Clone)] +pub struct MigrationResult { + /// Number of migrations applied + pub applied: usize, + /// Number of migrations skipped (already applied) + pub skipped: usize, +} + +impl std::fmt::Display for MigrationResult { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "{} migrations applied, {} skipped", + self.applied, self.skipped + ) + } +} diff --git a/crates/weaver-index/src/clickhouse/schema.rs b/crates/weaver-index/src/clickhouse/schema.rs new file mode 100644 index 0000000..26162c0 --- /dev/null +++ b/crates/weaver-index/src/clickhouse/schema.rs @@ -0,0 +1,67 @@ +use chrono::{DateTime, Utc}; +use clickhouse::Row; + +/// Table names for production schema +pub struct Tables; + +impl Tables { + pub const RAW_RECORDS: &'static str = "raw_records"; + pub const RAW_IDENTITY_EVENTS: &'static str = "raw_identity_events"; + pub const RAW_ACCOUNT_EVENTS: &'static str = "raw_account_events"; + pub const RAW_EVENTS_DLQ: &'static str = "raw_events_dlq"; + pub const FIREHOSE_CURSOR: &'static str = "firehose_cursor"; +} + +/// Row type for raw_records table +/// Schema defined in migrations/clickhouse/001_raw_records.sql +#[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] +pub struct RawRecord { + pub did: String, + pub collection: String, + pub rkey: String, + pub cid: String, + pub record: String, // JSON string - ClickHouse JSON type accepts string + pub operation: String, + pub seq: u64, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub event_time: DateTime, +} + +/// Row type for raw_identity_events table +#[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] +pub struct RawIdentityEvent { + pub did: String, + pub handle: String, + pub seq: u64, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub event_time: DateTime, +} + +/// Row type for raw_account_events table +#[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] +pub struct RawAccountEvent { + pub did: String, + pub active: u8, + pub status: String, + pub seq: u64, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub event_time: DateTime, +} + +/// Row type for raw_events_dlq table +#[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] +pub struct RawEventDlq { + pub event_type: String, + pub raw_data: String, // JSON string + pub error_message: String, + pub seq: u64, +} + +/// Row type for firehose_cursor table +#[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] +pub struct FirehoseCursor { + pub consumer_id: String, + pub seq: u64, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub event_time: DateTime, +} diff --git a/crates/weaver-index/src/config.rs b/crates/weaver-index/src/config.rs index 115b262..48cf2ea 100644 --- a/crates/weaver-index/src/config.rs +++ b/crates/weaver-index/src/config.rs @@ -1,61 +1,102 @@ -use miette::miette; -use serde::{Deserialize, Serialize}; -use std::{env, fs}; +use crate::error::{ConfigError, IndexError}; +use url::Url; -#[derive(Deserialize, Serialize, Clone)] -pub struct OauthConfig { - pub jwks: Vec, +/// ClickHouse connection configuration +#[derive(Debug, Clone)] +pub struct ClickHouseConfig { + pub url: Url, + pub database: String, + pub user: String, + pub password: String, } -#[derive(Deserialize, Serialize, Clone)] -pub struct JetstreamConfig { - pub endpoint: String, -} +impl ClickHouseConfig { + /// Load configuration from environment variables. + /// + /// Required env vars: + /// - `CLICKHOUSE_URL`: Full URL including protocol (e.g., `https://xyz.clickhouse.cloud:8443`) + /// - `CLICKHOUSE_DATABASE`: Database name + /// - `CLICKHOUSE_USER`: Username + /// - `CLICKHOUSE_PASSWORD`: Password + pub fn from_env() -> Result { + let url_str = std::env::var("CLICKHOUSE_URL").map_err(|_| ConfigError::MissingEnv { + var: "CLICKHOUSE_URL", + })?; + + let url = Url::parse(&url_str).map_err(|e| ConfigError::UrlParse { + url: url_str, + message: e.to_string(), + })?; + + let database = + std::env::var("CLICKHOUSE_DATABASE").map_err(|_| ConfigError::MissingEnv { + var: "CLICKHOUSE_DATABASE", + })?; + + let user = std::env::var("CLICKHOUSE_USER").map_err(|_| ConfigError::MissingEnv { + var: "CLICKHOUSE_USER", + })?; + + let password = + std::env::var("CLICKHOUSE_PASSWORD").map_err(|_| ConfigError::MissingEnv { + var: "CLICKHOUSE_PASSWORD", + })?; -impl Default for JetstreamConfig { - fn default() -> Self { - Self { - endpoint: "wss://jetstream1.us-east.bsky.network/subscribe".into(), - } + Ok(Self { + url, + database, + user, + password, + }) } } -#[derive(Deserialize, Serialize, Clone)] -pub struct CoreConfig { - pub db_path: String, - pub listen_addr: String, - pub appview_host: String, - pub cookie_secret: String, +/// Firehose relay configuration +#[derive(Debug, Clone)] +pub struct FirehoseConfig { + pub relay_url: Url, + pub cursor: Option, } -impl Default for CoreConfig { - fn default() -> Self { - Self { - db_path: "postgres://postgres:@localhost/weaver_appview".into(), - listen_addr: "0.0.0.0:4000".into(), - appview_host: "https://appview.weaver.sh".into(), - cookie_secret: "00000000000000000000000000000000".into(), - } +impl FirehoseConfig { + /// Default relay URL (Bluesky network) + pub const DEFAULT_RELAY: &'static str = "wss://bsky.network"; + + /// Load configuration from environment variables. + /// + /// Optional env vars: + /// - `FIREHOSE_RELAY_URL`: Relay WebSocket URL (default: wss://bsky.network) + /// - `FIREHOSE_CURSOR`: Starting cursor position (default: none, starts from live) + pub fn from_env() -> Result { + let relay_str = + std::env::var("FIREHOSE_RELAY_URL").unwrap_or_else(|_| Self::DEFAULT_RELAY.to_string()); + + let relay_url = Url::parse(&relay_str).map_err(|e| ConfigError::UrlParse { + url: relay_str, + message: e.to_string(), + })?; + + let cursor = std::env::var("FIREHOSE_CURSOR") + .ok() + .and_then(|s| s.parse().ok()); + + Ok(Self { relay_url, cursor }) } } -#[derive(Deserialize, Serialize, Clone)] +/// Combined configuration for the indexer +#[derive(Debug, Clone)] pub struct Config { - pub oauth: OauthConfig, - pub jetstream: JetstreamConfig, - pub core: CoreConfig, + pub clickhouse: ClickHouseConfig, + pub firehose: FirehoseConfig, } impl Config { - pub fn load(config_file: &str) -> miette::Result { - let mut config_string = fs::read_to_string(config_file) - .map_err(|e| miette!("error reading config file {}", e))?; - // substitute environment variables in config file - for (k, v) in env::vars() { - config_string = config_string.replace(&format!("${}", k), &v); - } - - Ok(toml::from_str(&config_string) - .map_err(|e| miette!("error parsing config file {}", e))?) + /// Load all configuration from environment variables. + pub fn from_env() -> Result { + Ok(Self { + clickhouse: ClickHouseConfig::from_env()?, + firehose: FirehoseConfig::from_env()?, + }) } } diff --git a/crates/weaver-index/src/db.rs b/crates/weaver-index/src/db.rs deleted file mode 100644 index b8b3031..0000000 --- a/crates/weaver-index/src/db.rs +++ /dev/null @@ -1,63 +0,0 @@ -use diesel::prelude::*; -use diesel_migrations::{EmbeddedMigrations, MigrationHarness, embed_migrations}; -pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!(); -use diesel_async::RunQueryDsl; -use diesel_async::pooled_connection::AsyncDieselConnectionManager; -use diesel_async::pooled_connection::deadpool::Pool; -use diesel_async::sync_connection_wrapper::SyncConnectionWrapper; - -#[derive(Clone)] -pub struct Db { - pub pool: Pool>, -} - -impl Db { - /// Yes, this fuction can and WILL panic if it can't create the connection pool - /// for some reason. We just want to bail because the appview - /// does not work without a database. - pub async fn new(db_path: Option) -> Self { - let database_url = if let Some(db_path) = db_path { - db_path - } else { - std::env::var("DATABASE_URL").expect("DATABASE_URL must be set") - }; - let config = AsyncDieselConnectionManager::>::new( - database_url, - ); - let pool = Pool::builder(config) - .build() - .expect("Failed to create pool"); - Self { pool } - } -} - -pub fn run_migrations( - db_path: Option, -) -> Result<(), Box> { - let database_url = if let Some(db_path) = db_path { - db_path - } else { - std::env::var("DATABASE_URL").expect("DATABASE_URL must be set") - }; - let mut connection = SqliteConnection::establish(&database_url) - .unwrap_or_else(|_| panic!("Error connecting to {}", database_url)); - // This will run the necessary migrations. - // - // See the documentation for `MigrationHarness` for - // all available methods. - println!("Attempting migrations..."); - let result = connection.run_pending_migrations(MIGRATIONS); - println!("{:?}", result); - if result.is_err() { - println!("Failed to run migrations"); - return result.map(|_| ()); - } - println!("Migrations Applied:"); - let applied_migrations = connection.applied_migrations()?; - for migration in applied_migrations { - println!(" * {}", migration); - } - Ok(()) -} - -pub struct Runtime; diff --git a/crates/weaver-index/src/error.rs b/crates/weaver-index/src/error.rs new file mode 100644 index 0000000..fba3309 --- /dev/null +++ b/crates/weaver-index/src/error.rs @@ -0,0 +1,118 @@ +use miette::Diagnostic; +use thiserror::Error; + +/// Top-level error type for weaver-index operations +#[derive(Debug, Error, Diagnostic)] +pub enum IndexError { + #[error(transparent)] + #[diagnostic(transparent)] + ClickHouse(#[from] ClickHouseError), + + #[error(transparent)] + #[diagnostic(transparent)] + Firehose(#[from] FirehoseError), + + #[error(transparent)] + #[diagnostic(transparent)] + Car(#[from] CarError), + + #[error(transparent)] + #[diagnostic(transparent)] + Config(#[from] ConfigError), +} + +/// ClickHouse database errors +#[derive(Debug, Error, Diagnostic)] +pub enum ClickHouseError { + #[error("failed to connect to ClickHouse: {message}")] + #[diagnostic(code(clickhouse::connection))] + Connection { + message: String, + #[source] + source: clickhouse::error::Error, + }, + + #[error("ClickHouse query failed: {message}")] + #[diagnostic(code(clickhouse::query))] + Query { + message: String, + #[source] + source: clickhouse::error::Error, + }, + + #[error("failed to insert batch: {message}")] + #[diagnostic(code(clickhouse::insert))] + Insert { + message: String, + #[source] + source: clickhouse::error::Error, + }, + + #[error("schema migration failed: {message}")] + #[diagnostic(code(clickhouse::schema))] + Schema { message: String }, +} + +/// Firehose/subscription stream errors +#[derive(Debug, Error, Diagnostic)] +pub enum FirehoseError { + #[error("failed to connect to relay at {url}")] + #[diagnostic(code(firehose::connection))] + Connection { url: String, message: String }, + + #[error("websocket stream error: {message}")] + #[diagnostic(code(firehose::stream))] + Stream { message: String }, + + #[error("failed to parse event header")] + #[diagnostic(code(firehose::header))] + HeaderParse { + #[source] + source: ciborium::de::Error, + }, + + #[error("failed to decode event body: {event_type}")] + #[diagnostic(code(firehose::decode))] + BodyDecode { event_type: String, message: String }, + + #[error("unknown event type: {event_type}")] + #[diagnostic(code(firehose::unknown_event))] + UnknownEvent { event_type: String }, +} + +/// CAR file parsing errors +#[derive(Debug, Error, Diagnostic)] +pub enum CarError { + #[error("failed to parse CAR data")] + #[diagnostic(code(car::parse))] + Parse { message: String }, + + #[error("block not found for CID: {cid}")] + #[diagnostic(code(car::block_not_found))] + BlockNotFound { cid: String }, + + #[error("failed to decode record from block: {message}")] + #[diagnostic(code(car::record_decode))] + RecordDecode { message: String }, +} + +/// Configuration errors +#[derive(Debug, Error, Diagnostic)] +pub enum ConfigError { + #[error("missing required environment variable: {var}")] + #[diagnostic( + code(config::missing_env), + help("Set the {var} environment variable or add it to your .env file") + )] + MissingEnv { var: &'static str }, + + #[error("invalid configuration value for {field}: {message}")] + #[diagnostic(code(config::invalid))] + Invalid { field: &'static str, message: String }, + + #[error("failed to parse URL: {url}")] + #[diagnostic(code(config::url_parse))] + UrlParse { url: String, message: String }, +} + +pub type Result = std::result::Result; diff --git a/crates/weaver-index/src/firehose.rs b/crates/weaver-index/src/firehose.rs new file mode 100644 index 0000000..3a1c7ba --- /dev/null +++ b/crates/weaver-index/src/firehose.rs @@ -0,0 +1,7 @@ +mod consumer; +mod records; + +pub use consumer::{ + FirehoseConsumer, MessageStream, SubscribeReposMessage, Commit, Identity, Account, Sync, +}; +pub use records::{extract_records, ExtractedRecord}; diff --git a/crates/weaver-index/src/firehose/consumer.rs b/crates/weaver-index/src/firehose/consumer.rs new file mode 100644 index 0000000..772398f --- /dev/null +++ b/crates/weaver-index/src/firehose/consumer.rs @@ -0,0 +1,48 @@ +use crate::config::FirehoseConfig; +use crate::error::{FirehoseError, IndexError}; +use jacquard_common::stream::StreamError; +use jacquard_common::xrpc::subscription::{SubscriptionClient, TungsteniteSubscriptionClient}; +use n0_future::stream::Boxed; + +// Re-export the message types from weaver_api for convenience +pub use weaver_api::com_atproto::sync::subscribe_repos::{ + Account, Commit, Identity, SubscribeRepos, SubscribeReposMessage, Sync, +}; + +/// Typed firehose message stream +pub type MessageStream = Boxed, StreamError>>; + +/// Firehose consumer that connects to a relay and yields typed events +pub struct FirehoseConsumer { + config: FirehoseConfig, +} + +impl FirehoseConsumer { + pub fn new(config: FirehoseConfig) -> Self { + Self { config } + } + + /// Connect to the firehose and return a typed message stream + /// + /// Messages are automatically decoded and converted to owned ('static) types. + pub async fn connect(&self) -> Result { + let client = TungsteniteSubscriptionClient::from_base_uri(self.config.relay_url.clone()); + + let mut params = SubscribeRepos::new(); + if let Some(cursor) = self.config.cursor { + params = params.cursor(cursor); + } + let params = params.build(); + + let stream = client + .subscribe(¶ms) + .await + .map_err(|e| FirehoseError::Connection { + url: self.config.relay_url.to_string(), + message: e.to_string(), + })?; + + let (_sink, messages) = stream.into_stream(); + Ok(messages) + } +} diff --git a/crates/weaver-index/src/firehose/records.rs b/crates/weaver-index/src/firehose/records.rs new file mode 100644 index 0000000..fa64762 --- /dev/null +++ b/crates/weaver-index/src/firehose/records.rs @@ -0,0 +1,108 @@ +use crate::error::{CarError, IndexError}; +use bytes::Bytes; +use jacquard_repo::car::reader::parse_car_bytes; +use smol_str::{SmolStr, ToSmolStr}; + +use super::consumer::Commit; + +/// An extracted record from a firehose commit +#[derive(Debug, Clone)] +pub struct ExtractedRecord { + /// DID of the repo owner + pub did: SmolStr, + /// Collection NSID (e.g., "app.bsky.feed.post") + pub collection: SmolStr, + /// Record key within the collection + pub rkey: SmolStr, + /// Content identifier + pub cid: String, + /// Operation type: "create", "update", or "delete" + pub operation: SmolStr, + /// Raw DAG-CBOR bytes of the record (None for deletes) + pub cbor_bytes: Option, + /// Sequence number from the firehose event + pub seq: i64, + /// Event timestamp (milliseconds since epoch) + pub event_time_ms: i64, +} + +impl ExtractedRecord { + /// Decode the CBOR bytes to JSON string + /// + /// Uses jacquard's RawData type which properly handles CID links + /// and other AT Protocol specific types. + pub fn to_json(&self) -> Result, IndexError> { + use jacquard_common::types::value::{RawData, from_cbor}; + + match &self.cbor_bytes { + Some(bytes) => { + // RawData handles CID links and other IPLD types correctly + let value: RawData<'static> = + from_cbor::(bytes).map_err(|e| CarError::RecordDecode { + message: format!("failed to decode DAG-CBOR: {}", e), + })?; + let json = serde_json::to_string(&value).map_err(|e| CarError::RecordDecode { + message: format!("failed to encode JSON: {}", e), + })?; + Ok(Some(json)) + } + None => Ok(None), + } + } +} + +/// Extract records from a firehose commit +/// +/// Parses the CAR data and extracts each record referenced by the operations. +pub async fn extract_records(commit: &Commit<'_>) -> Result, IndexError> { + let parsed_car = parse_car_bytes(&commit.blocks) + .await + .map_err(|e| CarError::Parse { + message: e.to_string(), + })?; + + let event_time_ms = commit.time.as_ref().timestamp_millis(); + let mut records = Vec::with_capacity(commit.ops.len()); + + for op in &commit.ops { + let path: &str = op.path.as_ref(); + + // Path format: "collection/rkey" + let (collection, rkey) = match path.split_once('/') { + Some((c, r)) => (c.to_smolstr(), r.to_smolstr()), + None => { + tracing::warn!(path = %path, "invalid op path format, skipping"); + continue; + } + }; + + let operation = op.action.to_smolstr(); + let cid_str = op.cid.as_ref().map(|c| c.to_string()).unwrap_or_default(); + + // For creates/updates, look up the record in the CAR blocks + let cbor_bytes = if let Some(cid_link) = &op.cid { + match cid_link.0.to_ipld() { + Ok(ipld_cid) => parsed_car.blocks.get(&ipld_cid).cloned(), + Err(_) => { + tracing::warn!(cid = %cid_str, "failed to convert CID to IPLD format"); + None + } + } + } else { + None + }; + + records.push(ExtractedRecord { + did: commit.repo.to_smolstr(), + collection, + rkey, + cid: cid_str, + operation, + cbor_bytes, + seq: commit.seq, + event_time_ms, + }); + } + + Ok(records) +} diff --git a/crates/weaver-index/src/lib.rs b/crates/weaver-index/src/lib.rs new file mode 100644 index 0000000..f175934 --- /dev/null +++ b/crates/weaver-index/src/lib.rs @@ -0,0 +1,7 @@ +pub mod clickhouse; +pub mod config; +pub mod error; +pub mod firehose; + +pub use config::Config; +pub use error::{IndexError, Result}; diff --git a/crates/weaver-index/src/main.rs b/crates/weaver-index/src/main.rs deleted file mode 100644 index 5f6118a..0000000 --- a/crates/weaver-index/src/main.rs +++ /dev/null @@ -1,134 +0,0 @@ -pub mod api_error; - -pub mod config; -pub mod db; -pub mod middleware; -pub mod models; -pub mod oauth; -pub mod routes; -pub mod schema; -pub mod state; -pub mod telemetry; - -use axum::Router; -use clap::Parser; -use config::*; -use db::*; -use dotenvy::dotenv; -use miette::IntoDiagnostic; -use miette::miette; -use state::*; -use std::env; - -use tokio::net::TcpListener; -use tracing::{debug, error, info}; - -#[derive(Parser)] -#[command(author, version, about, long_about = None)] -struct Cli { - #[arg( - short, - long, - value_name = "FILE", - default_value = "appview-config.toml" - )] - config: String, -} - -#[tokio::main] -async fn main() -> miette::Result<()> { - let config = initialize()?; - // Run any migrations before we do anything else. - let db_path = config.core.db_path.clone(); - let _ = tokio::task::spawn_blocking(|| db::run_migrations(Some(db_path))) - .await - .into_diagnostic()?; - let db = Db::new(Some(config.core.db_path.clone())).await; - debug!("Connected to database"); - // Spin up our server. - info!("Starting server on {}", config.core.listen_addr); - let listener = TcpListener::bind(&config.core.listen_addr) - .await - .expect("Failed to bind address"); - let router = router(config, db); - axum::serve(listener, router) - .await - .expect("Failed to start server"); - Ok(()) -} - -pub fn router(cfg: Config, db: Db) -> Router { - let app_state = AppState::new(cfg, db); - - // Middleware that adds high level tracing to a Service. - // Trace comes with good defaults but also supports customizing many aspects of the output: - // https://docs.rs/tower-http/latest/tower_http/trace/index.html - let trace_layer = telemetry::trace_layer(); - - // Sets 'x-request-id' header with randomly generated uuid v7. - let request_id_layer = middleware::request_id_layer(); - - // Propagates 'x-request-id' header from the request to the response. - let propagate_request_id_layer = middleware::propagate_request_id_layer(); - - // Layer that applies the Cors middleware which adds headers for CORS. - let cors_layer = middleware::cors_layer(); - - // Layer that applies the Timeout middleware, which sets a timeout for requests. - // The default value is 15 seconds. - let timeout_layer = middleware::timeout_layer(); - - // Any trailing slashes from request paths will be removed. For example, a request with `/foo/` - // will be changed to `/foo` before reaching the internal service. - let normalize_path_layer = middleware::normalize_path_layer(); - - // Create the router with the routes. - let router = routes::router(); - - // Combine all the routes and apply the middleware layers. - // The order of the layers is important. The first layer is the outermost layer. - Router::new() - .merge(router) - .layer(normalize_path_layer) - .layer(cors_layer) - .layer(timeout_layer) - .layer(propagate_request_id_layer) - .layer(trace_layer) - .layer(request_id_layer) - .with_state(app_state) -} - -pub fn initialize() -> miette::Result { - miette::set_hook(Box::new(|_| { - Box::new( - miette::MietteHandlerOpts::new() - .terminal_links(true) - //.rgb_colors(miette::RgbColors::) - .with_cause_chain() - .with_syntax_highlighting(miette::highlighters::SyntectHighlighter::default()) - .color(true) - .context_lines(5) - .tab_width(2) - .break_words(true) - .build(), - ) - })) - .map_err(|e| miette!("Failed to set miette hook: {}", e))?; - miette::set_panic_hook(); - dotenv().ok(); - let cli = Cli::parse(); - let config = config::Config::load(&cli.config); - let config = if let Err(e) = config { - error!("{}", e); - config::Config::load( - &env::var("APPVIEW_CONFIG").expect("Either set APPVIEW_CONFIG to the path to your config file, pass --config FILE to specify the path, or create a file called appview-config.toml in the directory where you are running the binary from."), - ) - .map_err(|e| miette!(e)) - } else { - config - }?; - let log_dir = env::var("LOG_DIR").unwrap_or_else(|_| "/tmp/appview".to_string()); - std::fs::create_dir_all(&log_dir).unwrap(); - let _guard = telemetry::setup_tracing(&log_dir); - Ok(config) -} diff --git a/crates/weaver-index/src/middleware.rs b/crates/weaver-index/src/middleware.rs deleted file mode 100644 index dde3ec3..0000000 --- a/crates/weaver-index/src/middleware.rs +++ /dev/null @@ -1,61 +0,0 @@ -use std::time::Duration; - -use axum::http::HeaderName; -use hyper::Request; -use tower_http::{ - cors::{AllowHeaders, Any, CorsLayer}, - normalize_path::NormalizePathLayer, - request_id::{MakeRequestId, PropagateRequestIdLayer, RequestId, SetRequestIdLayer}, - timeout::TimeoutLayer, -}; - -#[derive(Clone, Default)] -pub struct Id; - -impl MakeRequestId for Id { - fn make_request_id(&mut self, _: &Request) -> Option { - let id = uuid::Uuid::now_v7().to_string().parse().unwrap(); - Some(RequestId::new(id)) - } -} - -/// Sets the 'x-request-id' header with a randomly generated UUID v7. -/// -/// SetRequestId will not override request IDs if they are already present -/// on requests or responses. -pub fn request_id_layer() -> SetRequestIdLayer { - let x_request_id = HeaderName::from_static("x-request-id"); - SetRequestIdLayer::new(x_request_id.clone(), Id) -} - -// Propagates 'x-request-id' header from the request to the response. -/// -/// PropagateRequestId wont override request ids if its already -/// present on requests or responses. -pub fn propagate_request_id_layer() -> PropagateRequestIdLayer { - let x_request_id = HeaderName::from_static("x-request-id"); - PropagateRequestIdLayer::new(x_request_id) -} - -/// Layer that applies the Cors middleware which adds headers for CORS. -pub fn cors_layer() -> CorsLayer { - CorsLayer::new() - .allow_origin(Any) - .allow_methods(Any) - .allow_headers(AllowHeaders::mirror_request()) - .max_age(Duration::from_secs(600)) -} - -/// Layer that applies the Timeout middleware which apply a timeout to requests. -/// The default timeout value is set to 15 seconds. -pub fn timeout_layer() -> TimeoutLayer { - TimeoutLayer::new(Duration::from_secs(15)) -} - -/// Middleware that normalizes paths. -/// -/// Any trailing slashes from request paths will be removed. For example, a request with `/foo/` -/// will be changed to `/foo` before reaching the inner service. -pub fn normalize_path_layer() -> NormalizePathLayer { - NormalizePathLayer::trim_trailing_slash() -} diff --git a/crates/weaver-index/src/models.rs b/crates/weaver-index/src/models.rs deleted file mode 100644 index 0e193c1..0000000 --- a/crates/weaver-index/src/models.rs +++ /dev/null @@ -1,124 +0,0 @@ -use chrono::NaiveDateTime; -use diesel::prelude::*; -use jacquard::CowStr; - -#[derive(Queryable, Selectable)] -#[diesel(table_name = crate::schema::profile)] -pub struct Profile { - pub id: i32, - pub did: String, - pub avatar: Option, - pub description: String, - pub include_bluesky: bool, - pub include_tangled: bool, - pub location: Option, - pub pinned_post: Option, - pub created_at: Option, -} - -#[derive(Queryable, Selectable)] -#[diesel(table_name = crate::schema::registrations)] -pub struct Registration { - pub id: i32, - pub domain: String, - pub did: String, - pub secret: String, - pub created: NaiveDateTime, - pub registered: Option, -} - -#[derive(Queryable, Selectable)] -#[diesel(table_name = crate::schema::public_keys)] -pub struct PublicKey<'a> { - pub id: i32, - pub did: CowStr<'a>, - pub name: String, - pub key_contents: String, - pub rkey: String, - pub created: NaiveDateTime, -} - -#[derive(Queryable, Selectable)] -#[diesel(table_name = crate::schema::follows)] -pub struct Follow { - pub user_did: String, - pub subject_did: String, - pub rkey: String, - pub followed_at: NaiveDateTime, -} - -#[derive(Queryable, Selectable)] -#[diesel(table_name = crate::schema::_jetstream)] -pub struct Jetstream { - pub id: i32, - pub last_time_us: i32, -} - -#[derive(Queryable, Selectable)] -#[diesel(table_name = crate::schema::emails)] -pub struct Email { - pub id: i32, - pub did: String, - pub email: String, - pub verified: bool, - pub verification_code: String, - pub last_sent: NaiveDateTime, - pub is_primary: bool, - pub created: NaiveDateTime, -} - -#[derive(Queryable, Selectable)] -#[diesel(table_name = crate::schema::profile_links)] -pub struct ProfileLink { - pub id: i32, - pub did: String, - pub link: String, -} - -#[derive(Queryable, Selectable)] -#[diesel(table_name = crate::schema::profile_pronouns)] -pub struct ProfilePronoun { - pub id: i32, - pub did: String, - pub pronoun: String, -} - -// Jacquard OAuth models - -#[derive(Queryable, Selectable, QueryableByName)] -#[diesel(table_name = crate::schema::oauth_sessions, check_for_backend(diesel::sqlite::Sqlite))] -pub struct OauthSession { - pub id: i32, - pub did: String, - pub session_id: String, - pub session_data: serde_json::Value, - pub created_at: NaiveDateTime, - pub updated_at: NaiveDateTime, -} - -#[derive(Insertable)] -#[diesel(table_name = crate::schema::oauth_sessions)] -pub struct NewOauthSession { - pub did: String, - pub session_id: String, - pub session_data: serde_json::Value, -} - -#[derive(Queryable, Selectable, QueryableByName)] -#[diesel(table_name = crate::schema::oauth_auth_requests)] -pub struct OauthAuthRequest { - pub id: i32, - pub state: String, - pub account_did: Option, - pub auth_req_data: serde_json::Value, - pub created_at: NaiveDateTime, - pub expires_at: NaiveDateTime, -} - -#[derive(Insertable)] -#[diesel(table_name = crate::schema::oauth_auth_requests)] -pub struct NewOauthAuthRequest { - pub state: String, - pub account_did: Option, - pub auth_req_data: serde_json::Value, -} diff --git a/crates/weaver-index/src/oauth.rs b/crates/weaver-index/src/oauth.rs deleted file mode 100644 index ce7a4a3..0000000 --- a/crates/weaver-index/src/oauth.rs +++ /dev/null @@ -1,200 +0,0 @@ -use diesel::prelude::*; -use diesel_async::RunQueryDsl; -use jacquard::oauth::authstore::ClientAuthStore; -use jacquard::oauth::session::{AuthRequestData, ClientSessionData}; -use jacquard::session::SessionStoreError; -use jacquard::types::string::Did; -use miette::IntoDiagnostic; - -use crate::models::{NewOauthAuthRequest, NewOauthSession, OauthAuthRequest, OauthSession}; - -/// Database-backed implementation of ClientAuthStore for jacquard OAuth -pub struct DBAuthStore { - pub db: crate::db::Db, -} - -impl DBAuthStore { - pub fn new(db: &crate::db::Db) -> Self { - Self { db: db.clone() } - } -} - -impl ClientAuthStore for DBAuthStore { - async fn get_session( - &self, - did_param: &Did<'_>, - session_id_param: &str, - ) -> Result>, SessionStoreError> { - use crate::schema::oauth_sessions::dsl::*; - - let mut conn = self - .db - .pool - .get() - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - let result: Vec = oauth_sessions - .filter(did.eq(did_param.as_str())) - .filter(session_id.eq(session_id_param)) - .limit(1) - .select(OauthSession::as_select()) - .load(&mut conn) - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - if let Some(sess) = result.get(0) { - let data = jacquard::from_json_value::(sess.session_data.clone()) - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - Ok(Some(data)) - } else { - Ok(None) - } - } - - async fn upsert_session( - &self, - session: ClientSessionData<'_>, - ) -> Result<(), SessionStoreError> { - use crate::schema::oauth_sessions::dsl::*; - - let mut conn = self - .db - .pool - .get() - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - let session_json = - serde_json::to_value(&session).map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - let new_session = NewOauthSession { - did: session.account_did.as_str().to_string(), - session_id: session.session_id.as_ref().to_string(), - session_data: session_json, - }; - - // Try insert, on conflict update - diesel::insert_into(oauth_sessions) - .values(&new_session) - .on_conflict((did, session_id)) - .do_update() - .set(( - session_data.eq(&new_session.session_data), - updated_at.eq(diesel::dsl::now), - )) - .execute(&mut conn) - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - Ok(()) - } - - async fn delete_session( - &self, - did_param: &Did<'_>, - session_id_param: &str, - ) -> Result<(), SessionStoreError> { - use crate::schema::oauth_sessions::dsl::*; - - let mut conn = self - .db - .pool - .get() - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - diesel::delete( - oauth_sessions - .filter(did.eq(did_param.as_str())) - .filter(session_id.eq(session_id_param)), - ) - .execute(&mut conn) - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - Ok(()) - } - - async fn get_auth_req_info( - &self, - state_param: &str, - ) -> Result>, SessionStoreError> { - use crate::schema::oauth_auth_requests::dsl::*; - - let mut conn = self - .db - .pool - .get() - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - let result: Vec = oauth_auth_requests - .filter(state.eq(state_param)) - .limit(1) - .select(OauthAuthRequest::as_select()) - .load(&mut conn) - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - if let Some(req) = result.get(0) { - let auth_data = jacquard::from_json_value::(req.auth_req_data.clone()) - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - Ok(Some(auth_data)) - } else { - Ok(None) - } - } - - async fn save_auth_req_info( - &self, - auth_req_info: &AuthRequestData<'_>, - ) -> Result<(), SessionStoreError> { - use crate::schema::oauth_auth_requests::dsl::*; - - let mut conn = self - .db - .pool - .get() - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - let auth_json = serde_json::to_value(auth_req_info) - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - let new_auth_req = NewOauthAuthRequest { - state: auth_req_info.state.as_ref().to_string(), - account_did: auth_req_info - .account_did - .as_ref() - .map(|d| d.as_str().to_string()), - auth_req_data: auth_json, - }; - - diesel::insert_into(oauth_auth_requests) - .values(&new_auth_req) - .execute(&mut conn) - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - Ok(()) - } - - async fn delete_auth_req_info(&self, state_param: &str) -> Result<(), SessionStoreError> { - use crate::schema::oauth_auth_requests::dsl::*; - - let mut conn = self - .db - .pool - .get() - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - diesel::delete(oauth_auth_requests.filter(state.eq(state_param))) - .execute(&mut conn) - .await - .map_err(|e| SessionStoreError::Other(Box::new(e)))?; - - Ok(()) - } -} diff --git a/crates/weaver-index/src/routes/health_check.rs b/crates/weaver-index/src/routes/health_check.rs deleted file mode 100644 index dd319a6..0000000 --- a/crates/weaver-index/src/routes/health_check.rs +++ /dev/null @@ -1,8 +0,0 @@ -use axum::Json; -use serde_json::{Value, json}; - -use crate::api_error::ApiError; - -pub async fn health_check() -> Result, ApiError> { - Ok(Json(json!({ "status": "ok" }))) -} diff --git a/crates/weaver-index/src/routes/login.rs b/crates/weaver-index/src/routes/login.rs deleted file mode 100644 index 6459a50..0000000 --- a/crates/weaver-index/src/routes/login.rs +++ /dev/null @@ -1,4 +0,0 @@ -use axum::Json; -use serde_json::{Value, json}; - -use crate::api_error::ApiError; diff --git a/crates/weaver-index/src/routes/logout.rs b/crates/weaver-index/src/routes/logout.rs deleted file mode 100644 index 6459a50..0000000 --- a/crates/weaver-index/src/routes/logout.rs +++ /dev/null @@ -1,4 +0,0 @@ -use axum::Json; -use serde_json::{Value, json}; - -use crate::api_error::ApiError; diff --git a/crates/weaver-index/src/routes/mod.rs b/crates/weaver-index/src/routes/mod.rs deleted file mode 100644 index 0fb6db1..0000000 --- a/crates/weaver-index/src/routes/mod.rs +++ /dev/null @@ -1,10 +0,0 @@ -use axum::{Router, routing::get}; - -pub mod health_check; -pub mod oauth; - -use crate::AppState; - -pub fn router() -> Router { - Router::new().route("/health_check", get(health_check::health_check)) -} diff --git a/crates/weaver-index/src/routes/oauth.rs b/crates/weaver-index/src/routes/oauth.rs deleted file mode 100644 index f45227f..0000000 --- a/crates/weaver-index/src/routes/oauth.rs +++ /dev/null @@ -1,104 +0,0 @@ -use axum::{ - Form, Json, - extract::{Query, State}, -}; -use jacquard::{ - IntoStatic, - oauth::{ - atproto::atproto_client_metadata, - types::{AuthorizeOptions, CallbackParams}, - }, -}; -use miette::{IntoDiagnostic, Result, miette}; -use serde::{Deserialize, Serialize}; -use serde_json::{Value, json}; - -use crate::{api_error::ApiError, state::AppState}; - -/// Passthrough callback for native endpoint -pub async fn callback_native( - Query(params): Query>, -) -> Result, ApiError> { - Ok(Json(json!({ - "code": params.code, - "iss": params.iss, - "state": params.state, - }))) -} - -/// OAuth callback handler -pub async fn callback( - State(state): State, - Query(params): Query>, -) -> Result, ApiError> { - let oauth_client = state.oauth_client(); - let active_sessions = state.active_sessions(); - - let session = oauth_client - .callback(params) - .await - .map_err(|e| ApiError::InternalError(miette!("oauth callback failed: {e}")))?; - - let (did, session_id) = session.session_info().await; - let did = did.to_string(); - let session_id = session_id.clone().into_static(); - - active_sessions.insert(session_id.to_string(), session); - - Ok(Json(json!({ - "status": "authenticated", - "did": did, - "session_id": session_id, - }))) -} - -/// Get OAuth client metadata -pub async fn get_client_metadata(State(state): State) -> Result, ApiError> { - let metadata = atproto_client_metadata( - state.oauth_client().registry.client_data.config.clone(), - &state.oauth_client().registry.client_data.keyset, - ) - .map_err(|e| ApiError::InternalError(miette!("couldn't get oauth metadata: {e}")))?; - - Ok(Json(serde_json::to_value(metadata).map_err(|e| { - ApiError::InternalError(miette!("json serialization error: {e}")) - })?)) -} - -/// Get JWKS (public keys) -pub async fn get_jwks(State(state): State) -> Result, ApiError> { - let jwks = state.oauth_client().jwks(); - - Ok(Json(serde_json::to_value(jwks).map_err(|e| { - ApiError::InternalError(miette!("json serialization error: {e}")) - })?)) -} - -/// Login stub -pub async fn login(State(_state): State) -> Result, ApiError> { - Ok(Json(json!({ "status": "ok" }))) -} - -#[derive(Deserialize)] -pub struct AuthorizeParams { - pub handle: String, -} - -/// Start OAuth authorization flow -pub async fn authorize( - State(state): State, - Form(params): Form, -) -> Result, ApiError> { - let url = state - .oauth_client() - .start_auth(params.handle, AuthorizeOptions::default()) - .await - .map_err(|e| ApiError::InternalError(miette!("oauth authorize error: {e}")))?; - - Ok(Json(json!({ "url": url }))) -} - -/// Logout stub -pub async fn logout(State(_state): State) -> Result, ApiError> { - Ok(Json(json!({ "status": "ok" }))) -} diff --git a/crates/weaver-index/src/schema.patch b/crates/weaver-index/src/schema.patch deleted file mode 100644 index d49ecfe..0000000 --- a/crates/weaver-index/src/schema.patch +++ /dev/null @@ -1,29 +0,0 @@ ---- src/schema.rs 2025-11-07 17:26:09.742146790 -0500 -+++ src/schema.rs 2025-11-07 17:25:42.871390425 -0500 -@@ -31,24 +31,24 @@ - - diesel::table! { - oauth_auth_requests (id) { - id -> Integer, - state -> Text, - account_did -> Nullable, -- auth_req_data -> Binary, -+ auth_req_data -> Jsonb, - created_at -> Timestamp, - expires_at -> Timestamp, - } - } - - diesel::table! { - oauth_sessions (id) { - id -> Integer, - did -> Text, - session_id -> Text, -- session_data -> Binary, -+ session_data -> Jsonb, - created_at -> Timestamp, - updated_at -> Timestamp, - } - } - - diesel::table! { diff --git a/crates/weaver-index/src/schema.rs b/crates/weaver-index/src/schema.rs deleted file mode 100644 index 87cb3f6..0000000 --- a/crates/weaver-index/src/schema.rs +++ /dev/null @@ -1,117 +0,0 @@ -// @generated automatically by Diesel CLI. - -diesel::table! { - _jetstream (id) { - id -> Integer, - last_time_us -> Integer, - } -} - -diesel::table! { - emails (id) { - id -> Integer, - did -> Text, - email -> Text, - verified -> Bool, - verification_code -> Text, - last_sent -> Timestamp, - is_primary -> Bool, - created -> Timestamp, - } -} - -diesel::table! { - follows (user_did, subject_did) { - user_did -> Text, - subject_did -> Text, - rkey -> Text, - followed_at -> Timestamp, - } -} - -diesel::table! { - oauth_auth_requests (id) { - id -> Integer, - state -> Text, - account_did -> Nullable, - auth_req_data -> Jsonb, - created_at -> Timestamp, - expires_at -> Timestamp, - } -} - -diesel::table! { - oauth_sessions (id) { - id -> Integer, - did -> Text, - session_id -> Text, - session_data -> Jsonb, - created_at -> Timestamp, - updated_at -> Timestamp, - } -} - -diesel::table! { - profile (id) { - id -> Integer, - did -> Text, - avatar -> Nullable, - description -> Text, - include_bluesky -> Bool, - include_tangled -> Bool, - location -> Nullable, - pinned_post -> Nullable, - created_at -> Nullable, - } -} - -diesel::table! { - profile_links (id) { - id -> Integer, - did -> Text, - link -> Text, - } -} - -diesel::table! { - profile_pronouns (id) { - id -> Integer, - did -> Text, - pronoun -> Text, - } -} - -diesel::table! { - public_keys (id) { - id -> Integer, - did -> Text, - name -> Text, - key_contents -> Text, - rkey -> Text, - created -> Timestamp, - } -} - -diesel::table! { - registrations (id) { - id -> Integer, - domain -> Text, - did -> Text, - secret -> Text, - created -> Timestamp, - registered -> Nullable, - } -} - -diesel::allow_tables_to_appear_in_same_query!( - _jetstream, - emails, - follows, - oauth_auth_requests, - oauth_sessions, - profile, - profile_links, - profile_pronouns, - public_keys, - registrations, -); diff --git a/crates/weaver-index/src/state.rs b/crates/weaver-index/src/state.rs deleted file mode 100644 index 67a2cec..0000000 --- a/crates/weaver-index/src/state.rs +++ /dev/null @@ -1,85 +0,0 @@ -use dashmap::DashMap; -use jacquard::identity::JacquardResolver; -use jacquard::oauth::atproto::{AtprotoClientMetadata, GrantType}; -use jacquard::oauth::client::OAuthClient; -use jacquard::oauth::keyset::Keyset; -use jacquard::oauth::scopes::Scope; -use jacquard::oauth::session::ClientData; -use std::sync::Arc; -use url::Url; - -use crate::config::Config; -use crate::db::Db; -use crate::oauth::DBAuthStore; - -pub type AppviewOAuthClient = OAuthClient; -pub type AppviewOAuthSession = jacquard::oauth::client::OAuthSession; - -pub struct AppStateInner { - pub cfg: Config, - pub oauth_client: AppviewOAuthClient, - pub active_sessions: DashMap, -} - -#[derive(Clone)] -pub struct AppState { - pub db: Db, - inner: Arc, -} - -impl AppState { - pub fn new(cfg: Config, db: Db) -> Self { - let store = DBAuthStore::new(&db); - - // Build keyset from config JWKs - let keyset = Some( - Keyset::try_from(cfg.oauth.jwks.clone()).expect("failed to create keyset from JWKs"), - ); - - // Build AT Protocol client metadata - let client_id = - Url::parse(&cfg.core.appview_host).expect("appview_host must be a valid URL"); - - let redirect_uris = vec![ - Url::parse(&format!("{}/oauth/callback", cfg.core.appview_host)) - .expect("failed to build redirect URI"), - ]; - - let scopes = - Scope::parse_multiple("atproto transition:generic").expect("failed to parse scopes"); - - let config = AtprotoClientMetadata::new( - client_id.clone(), - Some(client_id), - redirect_uris, - vec![GrantType::AuthorizationCode, GrantType::RefreshToken], - scopes, - None, // jwks_uri - None means jwks will be embedded - ); - - let client_data = ClientData { keyset, config }; - - let oauth_client = OAuthClient::new(store, client_data); - - Self { - db, - inner: Arc::new(AppStateInner { - cfg, - oauth_client, - active_sessions: DashMap::new(), - }), - } - } - - pub fn cfg(&self) -> &Config { - &self.inner.as_ref().cfg - } - - pub fn oauth_client(&self) -> &AppviewOAuthClient { - &self.inner.as_ref().oauth_client - } - - pub fn active_sessions(&self) -> &DashMap { - &self.inner.as_ref().active_sessions - } -} diff --git a/crates/weaver-index/src/telemetry.rs b/crates/weaver-index/src/telemetry.rs deleted file mode 100644 index c491905..0000000 --- a/crates/weaver-index/src/telemetry.rs +++ /dev/null @@ -1,48 +0,0 @@ -use tower_http::{ - classify::{ServerErrorsAsFailures, SharedClassifier}, - trace::{DefaultMakeSpan, DefaultOnRequest, DefaultOnResponse, TraceLayer}, -}; -use tracing::Level; -use tracing_appender::{self, non_blocking, non_blocking::WorkerGuard, rolling::daily}; -use tracing_subscriber::{ - EnvFilter, - fmt::{self, layer, writer::MakeWriterExt}, - layer::SubscriberExt, - registry, - util::SubscriberInitExt, -}; -/// The `EnvFilter` type is used to filter log events based on the value of an environment variable. -/// In this case, we are using the `try_from_default_env` method to attempt to read the `RUST_LOG` environment variable, -/// which is used to set the log level for the application. -/// If the environment variable is not set, we default to the log level of `debug`. -/// The `RUST_LOG` environment variable is set in the Dockerfile and .env files. -pub fn setup_tracing>(logdir: S) -> WorkerGuard { - let (non_blocking_appender, guard) = non_blocking(daily(logdir.as_ref(), "general.log")); - let env_filter_layer = EnvFilter::try_from_default_env().unwrap_or_else(|_| { - format!( - "debug,{}=debug,tower_http=debug,axum=debug,hyper=debug,axum::rejection=trace,markdown=info", - env!("CARGO_PKG_NAME"), - ).into() - }); - let formatting_layer = fmt::layer().json(); - tracing_subscriber::registry() - .with(env_filter_layer) - .with(formatting_layer) - .with( - layer() - .with_writer(std::io::stdout.with_max_level(Level::DEBUG)) - .event_format(tracing_subscriber::fmt::format().pretty()), - ) - .with(layer().with_writer(non_blocking_appender.with_max_level(Level::INFO))) - .init(); - guard -} - -/// Returns a `TraceLayer` for HTTP requests and responses. -/// The `TraceLayer` is used to trace requests and responses in the application. -pub fn trace_layer() -> TraceLayer> { - TraceLayer::new_for_http() - .make_span_with(DefaultMakeSpan::new().level(Level::INFO)) - .on_request(DefaultOnRequest::new().level(Level::INFO)) - .on_response(DefaultOnResponse::new().level(Level::INFO)) -}