json vs cbor clickhouse benchmark, indexer skeleton

Cargo.lock

···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "rustversion",
  "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"
···
  "multihash",
  "serde",
  "serde_bytes",
- "unsigned-varint",
+ "unsigned-varint 0.8.0",
 ]
 
 [[package]]
···
  "inout",
  "zeroize",
 ]
+
+[[package]]
+name = "cityhash-rs"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "93a719913643003b84bd13022b4b7e703c09342cd03b679c4641c7d2e50dc34d"
 
 [[package]]
 name = "clap"
···
 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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "crypto_secretbox",
  "curve25519-dalek 5.0.0-pre.1",
  "salsa20",
- "serdect 0.4.1",
+ "serdect",
  "subtle",
  "zeroize",
 ]
···
 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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 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"
···
 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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53"
 dependencies = [
+ "pkcs8 0.10.2",
  "signature 2.2.0",
 ]
 
···
  "curve25519-dalek 4.1.3",
  "ed25519 2.2.3",
  "rand_core 0.6.4",
+ "serde",
  "sha2 0.10.9",
  "subtle",
  "zeroize",
···
 checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47"
 dependencies = [
  "base16ct 0.2.0",
- "base64ct",
  "crypto-bigint",
  "digest 0.10.7",
  "ff",
···
  "pkcs8 0.10.2",
  "rand_core 0.6.4",
  "sec1",
- "serde_json",
- "serdect 0.2.0",
  "subtle",
  "zeroize",
 ]
···
 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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2"
 dependencies = [
- "serde",
  "typenum",
  "version_check",
  "zeroize",
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
 dependencies = [
- "allocator-api2",
- "equivalent",
  "foldhash 0.1.5",
- "serde",
 ]
 
 [[package]]
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "chrono",
  "ciborium",
  "cid",
+ "ed25519-dalek 2.2.0",
  "futures",
  "getrandom 0.2.16",
  "getrandom 0.3.4",
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "cfg-if",
  "ecdsa",
  "elliptic-curve",
+ "once_cell",
  "sha2 0.10.9",
+ "signature 2.2.0",
 ]
 
 [[package]]
···
  "bitflags 2.10.0",
  "libc",
  "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]]
···
 ]
 
 [[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"
 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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "supports-color",
  "supports-hyperlinks",
  "supports-unicode",
- "syntect",
  "terminal_size",
  "textwrap",
  "unicode-width 0.1.14",
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 dependencies = [
  "core2",
  "serde",
- "unsigned-varint",
+ "unsigned-varint 0.8.0",
 ]
 
 [[package]]
···
 ]
 
 [[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"
 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"
 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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fe42f1670a52a47d448f14b6a5c61dd78fce51856e68edaa38f7ae3a46b8d6b6"
 dependencies = [
- "ecdsa",
  "elliptic-curve",
  "primeorder",
- "sha2 0.10.9",
 ]
 
 [[package]]
···
 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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 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"
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "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 = "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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d3ef0e35b322ddfaecbc60f34ab448e157e48531288ee49fafbb053696b8ffe2"
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "flate2",
  "fnv",
  "once_cell",
- "onig",
  "plist",
  "regex-syntax",
  "serde",
···
  "cfg-expr",
  "heck 0.5.0",
  "pkg-config",
- "toml 0.8.23",
+ "toml",
  "version-compare",
 ]
 
···
 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]]
···
 dependencies = [
  "indexmap 2.12.1",
  "serde",
- "serde_spanned 0.6.9",
+ "serde_spanned",
  "toml_datetime 0.6.11",
- "toml_write",
  "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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "tower-layer",
  "tower-service",
  "tracing",
- "uuid",
 ]
 
 [[package]]
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "nu-ansi-term",
  "once_cell",
  "regex-automata",
- "serde",
- "serde_json",
  "sharded-slab",
  "smallvec",
  "thread_local",
  "tracing",
  "tracing-core",
  "tracing-log",
- "tracing-serde",
 ]
 
 [[package]]
···
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eb066959b24b5196ae73cb057f45598450d2c5f71460e98c49b738086eff9c06"
···
 checksum = "673a33c33048a5ade91a6b139580fa174e19fb0d23f396dca9fa15f2e1e49b35"
 dependencies = [
  "cc",
- "downcast-rs 1.2.1",
+ "downcast-rs",
  "rustix",
  "scoped-tls",
  "smallvec",
···
 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]]

Cargo.toml

···
 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"] }

crates/weaver-index/Cargo.toml

···
 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"
-
-minijinja = { version = "2.9.0" }
-minijinja-contrib = { version = "2.9.0" }
-dashmap = "6.1.0"
+chrono = { version = "0.4", features = ["serde"] }
+smol_str = "0.3"
 
+# 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"

crates/weaver-index/build.rs

···
-fn main() {
-    println!("cargo:rerun-if-changed=migrations");
-}

crates/weaver-index/diesel.toml

···
-# 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"

crates/weaver-index/migrations/2025-05-15-230036_create_appviewdb/down.sql

···
--- 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;

crates/weaver-index/migrations/2025-05-15-230036_create_appviewdb/up.sql

···
-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;

crates/weaver-index/migrations/clickhouse/000_migrations.sql

···
+-- 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);

crates/weaver-index/migrations/clickhouse/001_raw_records.sql

···
+-- 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);

crates/weaver-index/migrations/clickhouse/002_identity_events.sql

···
+-- 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);

crates/weaver-index/migrations/clickhouse/003_account_events.sql

···
+-- 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);

crates/weaver-index/migrations/clickhouse/004_events_dlq.sql

···
+-- 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);

crates/weaver-index/migrations/clickhouse/005_firehose_cursor.sql

···
+-- 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);

crates/weaver-index/src/api_error.rs

···
-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<miette::Report> 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()
-    }
-}

crates/weaver-index/src/bin/storage_benchmark.rs

···
+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<Utc>,
+}
+
+/// 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<Utc>,
+}
+
+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<i64>,
+    max_ms: Option<i64>,
+    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::<RawRecordJson>(TABLE_JSON);
+    let mut cbor_inserter = client.inserter::<RawRecordCbor>(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);
+        }
+    }
+}

crates/weaver-index/src/bin/weaver_indexer.rs

···
+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(())
+}

crates/weaver-index/src/clickhouse.rs

···
+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,
+};

crates/weaver-index/src/clickhouse/client.rs

···
+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<Self, IndexError> {
+        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<T: Row>(&self, table: &str) -> Inserter<T> {
+        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<Vec<TableSize>, IndexError> {
+        let table_list = tables
+            .iter()
+            .map(|t| format!("'{}'", t))
+            .collect::<Vec<_>>()
+            .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::<TableSize>()
+            .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
+        }
+    }
+}

crates/weaver-index/src/clickhouse/migrations.rs

···
+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<MigrationResult, IndexError> {
+        // 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<Vec<String>, 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<String> = 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<Vec<String>, IndexError> {
+        let rows: Vec<MigrationRow> = 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
+        )
+    }
+}

crates/weaver-index/src/clickhouse/schema.rs

···
+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<Utc>,
+}
+
+/// 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<Utc>,
+}
+
+/// 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<Utc>,
+}
+
+/// 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<Utc>,
+}

crates/weaver-index/src/config.rs

···
-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<jose_jwk::Jwk>,
+/// 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<Self, IndexError> {
+        let url_str = std::env::var("CLICKHOUSE_URL").map_err(|_| ConfigError::MissingEnv {
+            var: "CLICKHOUSE_URL",
+        })?;
 
-impl Default for JetstreamConfig {
-    fn default() -> Self {
-        Self {
-            endpoint: "wss://jetstream1.us-east.bsky.network/subscribe".into(),
-        }
+        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",
+            })?;
+
+        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<i64>,
 }
 
-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<Self, IndexError> {
+        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<Config> {
-        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<Self, IndexError> {
+        Ok(Self {
+            clickhouse: ClickHouseConfig::from_env()?,
+            firehose: FirehoseConfig::from_env()?,
+        })
     }
 }

crates/weaver-index/src/db.rs

···
-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<SyncConnectionWrapper<SqliteConnection>>,
-}
-
-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<String>) -> 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::<SyncConnectionWrapper<SqliteConnection>>::new(
-            database_url,
-        );
-        let pool = Pool::builder(config)
-            .build()
-            .expect("Failed to create pool");
-        Self { pool }
-    }
-}
-
-pub fn run_migrations(
-    db_path: Option<String>,
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    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;

crates/weaver-index/src/error.rs

···
+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<std::io::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<T> = std::result::Result<T, IndexError>;

crates/weaver-index/src/firehose.rs

···
+mod consumer;
+mod records;
+
+pub use consumer::{
+    FirehoseConsumer, MessageStream, SubscribeReposMessage, Commit, Identity, Account, Sync,
+};
+pub use records::{extract_records, ExtractedRecord};

crates/weaver-index/src/firehose/consumer.rs

···
+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<Result<SubscribeReposMessage<'static>, 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<MessageStream, IndexError> {
+        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(&params)
+            .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)
+    }
+}

crates/weaver-index/src/firehose/records.rs

···
+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<Bytes>,
+    /// 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<Option<String>, 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::<RawData>(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<Vec<ExtractedRecord>, 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)
+}

crates/weaver-index/src/lib.rs

···
+pub mod clickhouse;
+pub mod config;
+pub mod error;
+pub mod firehose;
+
+pub use config::Config;
+pub use error::{IndexError, Result};

crates/weaver-index/src/main.rs

···
-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<Config> {
-    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)
-}

crates/weaver-index/src/middleware.rs

···
-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<B>(&mut self, _: &Request<B>) -> Option<RequestId> {
-        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<Id> {
-    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()
-}

crates/weaver-index/src/models.rs

···
-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<String>,
-    pub description: String,
-    pub include_bluesky: bool,
-    pub include_tangled: bool,
-    pub location: Option<String>,
-    pub pinned_post: Option<serde_json::Value>,
-    pub created_at: Option<NaiveDateTime>,
-}
-
-#[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<String>,
-}
-
-#[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<String>,
-    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<String>,
-    pub auth_req_data: serde_json::Value,
-}

crates/weaver-index/src/oauth.rs

···
-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<Option<ClientSessionData<'_>>, 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<OauthSession> = 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::<ClientSessionData>(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<Option<AuthRequestData<'_>>, 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<OauthAuthRequest> = 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::<AuthRequestData>(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(())
-    }
-}

crates/weaver-index/src/routes/health_check.rs

···
-use axum::Json;
-use serde_json::{Value, json};
-
-use crate::api_error::ApiError;
-
-pub async fn health_check() -> Result<Json<Value>, ApiError> {
-    Ok(Json(json!({ "status": "ok" })))
-}

crates/weaver-index/src/routes/login.rs

···
-use axum::Json;
-use serde_json::{Value, json};
-
-use crate::api_error::ApiError;

crates/weaver-index/src/routes/logout.rs

···
-use axum::Json;
-use serde_json::{Value, json};
-
-use crate::api_error::ApiError;

crates/weaver-index/src/routes/mod.rs

···
-use axum::{Router, routing::get};
-
-pub mod health_check;
-pub mod oauth;
-
-use crate::AppState;
-
-pub fn router() -> Router<AppState> {
-    Router::new().route("/health_check", get(health_check::health_check))
-}

crates/weaver-index/src/routes/oauth.rs

···
-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<CallbackParams<'_>>,
-) -> Result<Json<Value>, ApiError> {
-    Ok(Json(json!({
-        "code": params.code,
-        "iss": params.iss,
-        "state": params.state,
-    })))
-}
-
-/// OAuth callback handler
-pub async fn callback(
-    State(state): State<AppState>,
-    Query(params): Query<CallbackParams<'_>>,
-) -> Result<Json<Value>, 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<AppState>) -> Result<Json<Value>, 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<AppState>) -> Result<Json<Value>, 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<AppState>) -> Result<Json<Value>, 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<AppState>,
-    Form(params): Form<AuthorizeParams>,
-) -> Result<Json<Value>, 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<AppState>) -> Result<Json<Value>, ApiError> {
-    Ok(Json(json!({ "status": "ok" })))
-}

crates/weaver-index/src/schema.patch

···
---- 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<Text>,
--        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! {

crates/weaver-index/src/schema.rs

···
-// @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<Text>,
-        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<Text>,
-        description -> Text,
-        include_bluesky -> Bool,
-        include_tangled -> Bool,
-        location -> Nullable<Text>,
-        pinned_post -> Nullable<Text>,
-        created_at -> Nullable<Timestamp>,
-    }
-}
-
-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<Text>,
-    }
-}
-
-diesel::allow_tables_to_appear_in_same_query!(
-    _jetstream,
-    emails,
-    follows,
-    oauth_auth_requests,
-    oauth_sessions,
-    profile,
-    profile_links,
-    profile_pronouns,
-    public_keys,
-    registrations,
-);

crates/weaver-index/src/state.rs

···
-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<JacquardResolver, DBAuthStore>;
-pub type AppviewOAuthSession = jacquard::oauth::client::OAuthSession<JacquardResolver, DBAuthStore>;
-
-pub struct AppStateInner {
-    pub cfg: Config,
-    pub oauth_client: AppviewOAuthClient,
-    pub active_sessions: DashMap<String, AppviewOAuthSession>,
-}
-
-#[derive(Clone)]
-pub struct AppState {
-    pub db: Db,
-    inner: Arc<AppStateInner>,
-}
-
-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<String, AppviewOAuthSession> {
-        &self.inner.as_ref().active_sessions
-    }
-}

crates/weaver-index/src/telemetry.rs

···
-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<S: AsRef<str>>(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<SharedClassifier<ServerErrorsAsFailures>> {
-    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))
-}