backfill works, sqlite infra + observability set up, and now works as a nice sub in for record fetches, and acts as a cache.

Cargo.lock

···
 
 [[package]]
 name = "async-compression"
-version = "0.4.35"
+version = "0.4.36"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "07a926debf178f2d355197f9caddb08e54a9329d44748034bba349c5848cb519"
+checksum = "98ec5f6c2f8bc326c994cb9e241cc257ddaba9afa8555a43cffbb5dd86efaa37"
 dependencies = [
  "compression-codecs",
  "compression-core",
···
 version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
+
+[[package]]
+name = "aws-lc-rs"
+version = "1.15.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6b5ce75405893cd713f9ab8e297d8e438f624dde7d706108285f7e17a25a180f"
+dependencies = [
+ "aws-lc-sys",
+ "zeroize",
+]
+
+[[package]]
+name = "aws-lc-sys"
+version = "0.34.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "179c3777a8b5e70e90ea426114ffc565b2c1a9f82f6c4a0c5a34aa6ef5e781b6"
+dependencies = [
+ "cc",
+ "cmake",
+ "dunce",
+ "fs_extra",
+]
 
 [[package]]
 name = "axum"
···
 ]
 
 [[package]]
+name = "cmake"
+version = "0.1.54"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e7caa3f9de89ddbe2c607f4101924c5abec803763ae9534e4f4d7d8f84aa81f0"
+dependencies = [
+ "cc",
+]
+
+[[package]]
 name = "cobs"
 version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 
 [[package]]
 name = "compression-codecs"
-version = "0.4.34"
+version = "0.4.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "34a3cbbb8b6eca96f3a5c4bf6938d5b27ced3675d69f95bb51948722870bc323"
+checksum = "b0f7ac3e5b97fdce45e8922fb05cae2c37f7bbd63d30dd94821dacfd8f3f2bf2"
 dependencies = [
  "compression-core",
  "flate2",
···
 [[package]]
 name = "dioxus-primitives"
 version = "0.0.1"
-source = "git+https://github.com/DioxusLabs/components#f165b24277a8f04a2d78a6303effb9ba3a86c285"
+source = "git+https://github.com/DioxusLabs/components#7e5862e574aeceb3a3a021d042c165a839f1860b"
 dependencies = [
  "dioxus 0.7.2",
- "dioxus-time",
+ "dioxus-sdk-time",
  "lazy-js-bundle 0.6.2",
  "num-integer",
  "time",
···
 ]
 
 [[package]]
-name = "dioxus-time"
-version = "0.7.0"
-source = "git+https://github.com/ealmloff/dioxus-std?branch=0.7#8c868ac1d60e3232e3f16f6195d6deb3c016de17"
-dependencies = [
- "dioxus 0.7.2",
- "futures",
- "gloo-timers",
- "tokio",
-]
-
-[[package]]
 name = "dioxus-use-js-macro"
 version = "0.1.0"
 dependencies = [
···
 ]
 
 [[package]]
+name = "fallible-iterator"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
+
+[[package]]
+name = "fallible-streaming-iterator"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
+
+[[package]]
 name = "fancy-regex"
 version = "0.16.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "swc_macros_common",
  "syn 2.0.111",
 ]
+
+[[package]]
+name = "fs_extra"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
 [[package]]
 name = "futf"
···
 checksum = "0bb0228f477c0900c880fd78c8759b95c7636dbd7842707f49e132378aa2acdc"
 dependencies = [
  "heck 0.4.1",
- "proc-macro-crate 2.0.0",
+ "proc-macro-crate 2.0.2",
  "proc-macro-error",
  "proc-macro2",
  "quote",
···
  "hyper",
  "hyper-util",
  "rustls",
+ "rustls-native-certs",
  "rustls-pki-types",
  "tokio",
  "tokio-rustls",
···
 ]
 
 [[package]]
+name = "hyper-tls"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "70206fc6890eaca9fde8a0bf71caa2ddfc9fe045ac9e5c70df101a7dbde866e0"
+dependencies = [
+ "bytes",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
+ "native-tls",
+ "tokio",
+ "tokio-native-tls",
+ "tower-service",
+]
+
+[[package]]
 name = "hyper-util"
 version = "0.1.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 
 [[package]]
 name = "icu_properties"
-version = "2.1.1"
+version = "2.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e93fcd3157766c0c8da2f8cff6ce651a31f0810eaa1c51ec363ef790bbb5fb99"
+checksum = "020bfc02fe870ec3a66d93e677ccca0562506e5872c650f893269e08615d74ec"
 dependencies = [
  "icu_collections",
  "icu_locale_core",
···
 
 [[package]]
 name = "icu_properties_data"
-version = "2.1.1"
+version = "2.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "02845b3647bb045f1100ecd6480ff52f34c35f82d9880e029d329c21d1054899"
+checksum = "616c294cf8d725c6afcd8f55abc17c56464ef6211f9ed59cccffe534129c77af"
 
 [[package]]
 name = "icu_provider"
···
 dependencies = [
  "bitmaps",
  "rand_core 0.6.4",
- "rand_xoshiro",
+ "rand_xoshiro 0.6.0",
  "serde",
  "sized-chunks",
  "typenum",
···
 ]
 
 [[package]]
+name = "itertools"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285"
+dependencies = [
+ "either",
+]
+
+[[package]]
 name = "itoa"
 version = "1.0.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 [[package]]
 name = "jacquard"
 version = "0.9.4"
+source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4"
 dependencies = [
  "bytes",
  "getrandom 0.2.16",
···
 [[package]]
 name = "jacquard-api"
 version = "0.9.2"
+source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4"
 dependencies = [
  "bon",
  "bytes",
···
 [[package]]
 name = "jacquard-axum"
 version = "0.9.2"
+source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4"
 dependencies = [
  "axum",
  "bytes",
···
 [[package]]
 name = "jacquard-common"
 version = "0.9.2"
+source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4"
 dependencies = [
  "base64 0.22.1",
  "bon",
···
 [[package]]
 name = "jacquard-derive"
 version = "0.9.4"
+source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4"
 dependencies = [
  "heck 0.5.0",
  "jacquard-lexicon",
···
 [[package]]
 name = "jacquard-identity"
 version = "0.9.2"
+source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4"
 dependencies = [
  "bon",
  "bytes",
···
 [[package]]
 name = "jacquard-lexicon"
 version = "0.9.2"
+source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4"
 dependencies = [
  "cid",
  "dashmap 6.1.0",
···
 [[package]]
 name = "jacquard-oauth"
 version = "0.9.2"
+source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4"
 dependencies = [
  "base64 0.22.1",
  "bytes",
···
 [[package]]
 name = "jacquard-repo"
 version = "0.9.4"
+source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4"
 dependencies = [
  "bytes",
  "cid",
···
 ]
 
 [[package]]
+name = "libsqlite3-sys"
+version = "0.35.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "133c182a6a2c87864fe97778797e46c7e999672690dc9fa3ee8e241aa4a9c13f"
+dependencies = [
+ "cc",
+ "pkg-config",
+ "vcpkg",
+]
+
+[[package]]
 name = "libxdo"
 version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 version = "0.4.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
+
+[[package]]
+name = "loki-api"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bdc38a304f59a03e6efa3876766a48c70a766a93f88341c3fff4212834b8e327"
+dependencies = [
+ "prost",
+ "prost-types",
+]
 
 [[package]]
 name = "lol_alloc"
···
 
 [[package]]
 name = "loro"
-version = "1.10.0"
+version = "1.10.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "88dccd9df337cf38accfa64bd2267edfd2eeb17a459d38ab0f7ac80eb878cdc5"
+checksum = "d75216d8f99725531a30f7b00901ee154a4f8a9b7f125bfe032e197d4c7ffb8c"
 dependencies = [
  "enum-as-inner 0.6.1",
  "generic-btree",
···
 
 [[package]]
 name = "loro-internal"
-version = "1.10.0"
+version = "1.10.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9e5c014162198a558f119e67287f042c37feb55e82b50d97d6ac2a90333995dd"
+checksum = "f447044ec3d3ba572623859add3334bd87b84340ee5fdf00315bfee0e3ad3e3f"
 dependencies = [
  "append-only-bytes",
  "arref",
···
 [[package]]
 name = "markdown-weaver"
 version = "0.13.0"
+source = "git+https://github.com/rsform/markdown-weaver#52075e20a194375f1bd4a0c78201ce3b3a52c82d"
 dependencies = [
  "bitflags 2.10.0",
  "getopts",
···
 [[package]]
 name = "markdown-weaver-escape"
 version = "0.11.0"
+source = "git+https://github.com/rsform/markdown-weaver#52075e20a194375f1bd4a0c78201ce3b3a52c82d"
 
 [[package]]
 name = "markup5ever"
···
 ]
 
 [[package]]
+name = "metrics"
+version = "0.24.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5d5312e9ba3771cfa961b585728215e3d972c950a3eed9252aa093d6301277e8"
+dependencies = [
+ "ahash",
+ "portable-atomic",
+]
+
+[[package]]
+name = "metrics-exporter-prometheus"
+version = "0.17.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b166dea96003ee2531cf14833efedced545751d800f03535801d833313f8c15"
+dependencies = [
+ "base64 0.22.1",
+ "http-body-util",
+ "hyper",
+ "hyper-rustls",
+ "hyper-util",
+ "indexmap 2.12.1",
+ "ipnet",
+ "metrics",
+ "metrics-util",
+ "quanta",
+ "thiserror 2.0.17",
+ "tokio",
+ "tracing",
+]
+
+[[package]]
+name = "metrics-util"
+version = "0.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cdfb1365fea27e6dd9dc1dbc19f570198bc86914533ad639dae939635f096be4"
+dependencies = [
+ "crossbeam-epoch",
+ "crossbeam-utils",
+ "hashbrown 0.16.1",
+ "metrics",
+ "quanta",
+ "rand 0.9.2",
+ "rand_xoshiro 0.7.0",
+ "sketches-ddsketch",
+]
+
+[[package]]
 name = "miette"
 version = "5.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 [[package]]
 name = "mini-moka"
 version = "0.10.99"
+source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4"
 dependencies = [
  "crossbeam-channel",
  "crossbeam-utils",
···
 
 [[package]]
 name = "proc-macro-crate"
-version = "2.0.0"
+version = "2.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e8366a6159044a37876a2b9817124296703c586a5c92e2c53751fa06d8d43e8"
+checksum = "b00f26d3400549137f92511a46ac1cd8ce37cb5598a96d382381458b992a5d24"
 dependencies = [
- "toml_edit 0.20.7",
+ "toml_datetime 0.6.3",
+ "toml_edit 0.20.2",
 ]
 
 [[package]]
···
 ]
 
 [[package]]
+name = "prost"
+version = "0.13.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2796faa41db3ec313a31f7624d9286acf277b52de526150b7e69f3debf891ee5"
+dependencies = [
+ "bytes",
+ "prost-derive",
+]
+
+[[package]]
+name = "prost-derive"
+version = "0.13.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
+dependencies = [
+ "anyhow",
+ "itertools 0.14.0",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.111",
+]
+
+[[package]]
+name = "prost-types"
+version = "0.13.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "52c2c1bf36ddb1a1c396b3601a3cec27c2462e45f07c386894ec3ccf5332bd16"
+dependencies = [
+ "prost",
+]
+
+[[package]]
 name = "psl-types"
 version = "2.0.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "6f97cdb2a36ed4183de61b2f824cc45c9f1037f28afe0a322e9fff4c108b5aaa"
 dependencies = [
  "rand_core 0.6.4",
+]
+
+[[package]]
+name = "rand_xoshiro"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f703f4665700daf5512dcca5f43afa6af89f09db47fb56be587f80636bda2d41"
+dependencies = [
+ "rand_core 0.9.3",
 ]
 
 [[package]]
···
 
 [[package]]
 name = "reqwest"
-version = "0.12.24"
+version = "0.12.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9d0946410b9f7b082a427e4ef5c8ff541a88b357bc6c637c40db3a68ac70a36f"
+checksum = "b6eff9328d40131d43bd911d42d79eb6a47312002a4daefc9e37f17e74a7701a"
 dependencies = [
- "async-compression",
  "base64 0.22.1",
  "bytes",
  "cookie",
···
  "http-body-util",
  "hyper",
  "hyper-rustls",
+ "hyper-tls",
  "hyper-util",
  "js-sys",
  "log",
  "mime",
  "mime_guess",
+ "native-tls",
  "percent-encoding",
  "pin-project-lite",
  "quinn",
···
  "serde_urlencoded",
  "sync_wrapper",
  "tokio",
+ "tokio-native-tls",
  "tokio-rustls",
  "tokio-util",
  "tower",
···
 ]
 
 [[package]]
+name = "rusqlite"
+version = "0.37.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "165ca6e57b20e1351573e3729b958bc62f0e48025386970b6e4d29e7a7e71f3f"
+dependencies = [
+ "bitflags 2.10.0",
+ "fallible-iterator",
+ "fallible-streaming-iterator",
+ "hashlink",
+ "libsqlite3-sys",
+ "smallvec",
+]
+
+[[package]]
+name = "rusqlite_migration"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3fc9767ae49274bafd3e55be9d30405a033b7a59548327d87fd4971fbb58e264"
+dependencies = [
+ "include_dir",
+ "log",
+ "rusqlite",
+]
+
+[[package]]
 name = "rustc-demangle"
 version = "0.1.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "533f54bc6a7d4f647e46ad909549eda97bf5afc1585190ef692b4286b198bd8f"
 dependencies = [
+ "aws-lc-rs",
  "log",
  "once_cell",
  "ring",
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2ffdfa2f5286e2247234e03f680868ac2815974dc39e00ea15adc445d0aafe52"
 dependencies = [
+ "aws-lc-rs",
  "ring",
  "rustls-pki-types",
  "untrusted",
···
 ]
 
 [[package]]
+name = "sketches-ddsketch"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c1e9a774a6c28142ac54bb25d25562e6bcf957493a184f15ad4eebccb23e410a"
+
+[[package]]
 name = "slab"
 version = "0.4.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "borsh",
  "serde_core",
 ]
+
+[[package]]
+name = "snap"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1b6b67fb9a61334225b5b790716f609cd58395f895b3fe8b328786812a40bc3b"
 
 [[package]]
 name = "socket2"
···
 ]
 
 [[package]]
+name = "tokio-native-tls"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2"
+dependencies = [
+ "native-tls",
+ "tokio",
+]
+
+[[package]]
 name = "tokio-rustls"
 version = "0.26.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 
 [[package]]
 name = "tokio-tungstenite"
+version = "0.26.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a9daff607c6d2bf6c16fd681ccb7eecc83e4e2cdc1ca067ffaadfca5de7f084"
+dependencies = [
+ "futures-util",
+ "log",
+ "native-tls",
+ "tokio",
+ "tokio-native-tls",
+ "tungstenite 0.26.2",
+]
+
+[[package]]
+name = "tokio-tungstenite"
 version = "0.27.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "489a59b6730eda1b0171fcfda8b121f4bee2b35cba8645ca35c5f7ba3eb736c1"
···
 
 [[package]]
 name = "toml"
-version = "0.8.23"
+version = "0.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362"
+checksum = "185d8ab0dfbb35cf1399a6344d8484209c088f75f8f68230da55d48d95d43e3d"
 dependencies = [
  "serde",
  "serde_spanned",
- "toml_datetime 0.6.11",
- "toml_edit 0.22.27",
+ "toml_datetime 0.6.3",
+ "toml_edit 0.20.2",
 ]
 
 [[package]]
 name = "toml_datetime"
-version = "0.6.11"
+version = "0.6.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c"
+checksum = "7cda73e2f1397b1262d6dfdcef8aafae14d1de7748d66822d3bfeeb6d03e5e4b"
 dependencies = [
  "serde",
 ]
···
 checksum = "1b5bb770da30e5cbfde35a2d7b9b8a2c4b8ef89548a7a6aeab5c9a576e3e7421"
 dependencies = [
  "indexmap 2.12.1",
- "toml_datetime 0.6.11",
+ "toml_datetime 0.6.3",
  "winnow 0.5.40",
 ]
 
 [[package]]
 name = "toml_edit"
-version = "0.20.7"
+version = "0.20.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70f427fce4d84c72b5b732388bf4a9f4531b53f74e2887e3ecb2481f68f66d81"
-dependencies = [
- "indexmap 2.12.1",
- "toml_datetime 0.6.11",
- "winnow 0.5.40",
-]
-
-[[package]]
-name = "toml_edit"
-version = "0.22.27"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a"
+checksum = "396e4d48bbb2b7554c944bde63101b5ae446cff6ec4a24227428f15eb72ef338"
 dependencies = [
  "indexmap 2.12.1",
  "serde",
  "serde_spanned",
- "toml_datetime 0.6.11",
- "winnow 0.7.14",
+ "toml_datetime 0.6.3",
+ "winnow 0.5.40",
 ]
 
 [[package]]
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8"
 dependencies = [
+ "async-compression",
  "bitflags 2.10.0",
  "bytes",
  "futures-core",
···
 ]
 
 [[package]]
+name = "tracing-loki"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba3beec919fbdf99d719de8eda6adae3281f8a5b71ae40431f44dc7423053d34"
+dependencies = [
+ "loki-api",
+ "reqwest",
+ "serde",
+ "serde_json",
+ "snap",
+ "tokio",
+ "tokio-stream",
+ "tracing",
+ "tracing-core",
+ "tracing-log",
+ "tracing-serde",
+ "tracing-subscriber",
+ "url",
+]
+
+[[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"
···
  "rustls-pki-types",
  "sha1 0.10.6",
  "thiserror 1.0.69",
+ "utf-8",
+]
+
+[[package]]
+name = "tungstenite"
+version = "0.26.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4793cb5e56680ecbb1d843515b23b6de9a75eb04b66643e256a396d43be33c13"
+dependencies = [
+ "bytes",
+ "data-encoding",
+ "http",
+ "httparse",
+ "log",
+ "native-tls",
+ "rand 0.9.2",
+ "sha1 0.10.6",
+ "thiserror 2.0.17",
  "utf-8",
 ]
 
···
  "js-sys",
  "markdown-weaver",
  "markdown-weaver-escape",
+ "metrics",
+ "metrics-exporter-prometheus",
  "miette 7.6.0",
  "mime-sniffer",
  "n0-future 0.1.3",
···
  "thiserror 2.0.17",
  "tokio",
  "tracing",
+ "tracing-loki",
+ "tracing-subscriber",
  "trait-variant",
  "wasm-bindgen",
  "wasm-bindgen-futures",
···
 name = "weaver-index"
 version = "0.1.0"
 dependencies = [
+ "axum",
  "base64 0.22.1",
  "bytes",
  "chrono",
···
  "clickhouse",
  "dashmap 6.1.0",
  "dotenvy",
+ "futures-util",
  "humansize",
  "include_dir",
  "jacquard",
+ "jacquard-axum",
  "jacquard-common",
  "jacquard-repo",
  "miette 7.6.0",
  "n0-future 0.1.3",
+ "rusqlite",
+ "rusqlite_migration",
  "serde",
  "serde_ipld_dagcbor",
  "serde_json",
  "smol_str",
  "thiserror 2.0.17",
  "tokio",
+ "tokio-tungstenite 0.26.2",
+ "tower",
+ "tower-http",
  "tracing",
  "tracing-subscriber",
  "url",
  "weaver-api",
+ "weaver-common",
 ]
 
 [[package]]

Cargo.toml

···
 syntect = { version = "5.2.0", default-features = false }
 n0-future = "=0.1.3"
 tracing = { version = "0.1.41", default-features = false, features = ["std"] }
-# markdown-weaver = { git = "https://github.com/rsform/markdown-weaver" }
-# markdown-weaver-escape = { git = "https://github.com/rsform/markdown-weaver" }
-markdown-weaver = { path = "../markdown-weaver/markdown-weaver" }
-markdown-weaver-escape = { path = "../markdown-weaver/markdown-weaver-escape" }
+markdown-weaver = { git = "https://github.com/rsform/markdown-weaver" }
+markdown-weaver-escape = { git = "https://github.com/rsform/markdown-weaver" }
+# markdown-weaver = { path = "../markdown-weaver/markdown-weaver" }
+# markdown-weaver-escape = { path = "../markdown-weaver/markdown-weaver-escape" }
 
-# jacquard = { git = "https://tangled.org/@nonbinary.computer/jacquard", default-features = false, features = ["derive", "api_bluesky", "tracing"] }
-# jacquard-identity = { git = "https://tangled.org/@nonbinary.computer/jacquard", features = ["cache"] }
-# jacquard-common = { git = "https://tangled.org/@nonbinary.computer/jacquard" }
-# jacquard-axum = { git = "https://tangled.org/@nonbinary.computer/jacquard" }
-# jacquard-derive = { git = "https://tangled.org/@nonbinary.computer/jacquard" }
-# jacquard-lexicon = { git = "https://tangled.org/@nonbinary.computer/jacquard", default-features = false }
+jacquard = { git = "https://tangled.org/@nonbinary.computer/jacquard", default-features = false, features = ["derive", "api_bluesky", "tracing"] }
+jacquard-identity = { git = "https://tangled.org/@nonbinary.computer/jacquard", features = ["cache"] }
+jacquard-common = { git = "https://tangled.org/@nonbinary.computer/jacquard" }
+jacquard-axum = { git = "https://tangled.org/@nonbinary.computer/jacquard" }
+jacquard-derive = { git = "https://tangled.org/@nonbinary.computer/jacquard" }
+jacquard-lexicon = { git = "https://tangled.org/@nonbinary.computer/jacquard", default-features = false }
+jacquard-repo = { git = "https://tangled.org/@nonbinary.computer/jacquard" }
 
-jacquard = { path = "../jacquard/crates/jacquard", default-features = false, features = ["derive", "api_bluesky", "tracing", "cache"] }
-jacquard-identity = { path = "../jacquard/crates/jacquard-identity", features = ["cache"] }
-jacquard-api = { path = "../jacquard/crates/jacquard-api" }
-jacquard-common = { path = "../jacquard/crates/jacquard-common" }
-jacquard-axum = {path = "../jacquard/crates/jacquard-axum" }
-jacquard-derive = { path = "../jacquard/crates/jacquard-derive" }
-jacquard-lexicon = { path = "../jacquard/crates/jacquard-lexicon", default-features = false }
-jacquard-repo = { path = "../jacquard/crates/jacquard-repo" }
+# jacquard = { path = "../jacquard/crates/jacquard", default-features = false, features = ["derive", "api_bluesky", "tracing", "cache"] }
+# jacquard-identity = { path = "../jacquard/crates/jacquard-identity", features = ["cache"] }
+# jacquard-api = { path = "../jacquard/crates/jacquard-api" }
+# jacquard-common = { path = "../jacquard/crates/jacquard-common" }
+# jacquard-axum = {path = "../jacquard/crates/jacquard-axum" }
+# jacquard-derive = { path = "../jacquard/crates/jacquard-derive" }
+# jacquard-lexicon = { path = "../jacquard/crates/jacquard-lexicon", default-features = false }
+# jacquard-repo = { path = "../jacquard/crates/jacquard-repo" }
 
 # jacquard = { path = "../jacquard-facet/crates/jacquard", default-features = false, features = ["derive", "api_bluesky", "tracing", "serde"] }
 # jacquard-identity = { path = "../jacquard-facet/crates/jacquard-identity", features = ["cache"] }

crates/weaver-common/Cargo.toml

···
 dev = []
 native = ["jacquard/dns"]
 iroh = ["dep:iroh", "dep:iroh-gossip", "dep:iroh-tickets"]
+telemetry = ["dep:metrics", "dep:metrics-exporter-prometheus", "dep:tracing-subscriber", "dep:tracing-loki"]
 
 [dependencies]
 n0-future = { workspace = true }
···
 iroh = { version = "0.95", default-features = false, optional = true }
 iroh-gossip = { version = "0.95", default-features = false, features = ["net"], optional = true }
 iroh-tickets = { version = "0.2", optional = true }
+
+# Telemetry (optional, native-only)
+metrics = { version = "0.24.2", optional = true }
+metrics-exporter-prometheus = { version = "0.17.2", optional = true }
+tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"], optional = true }
+tracing-loki = { version = "0.2", optional = true }
 
 getrandom = { version = "0.3", features = [] }
 ring = { version = "0.17", default-features = false }

crates/weaver-common/src/lib.rs

···
 pub mod constellation;
 pub mod error;
 pub mod resolve;
+#[cfg(feature = "telemetry")]
+pub mod telemetry;
 pub mod transport;
 pub mod worker_rt;
 

crates/weaver-common/src/telemetry.rs

···
+//! Telemetry infrastructure for weaver services.
+//!
+//! Provides:
+//! - Prometheus metrics with `/metrics` endpoint
+//! - Tracing with pretty console output + optional Loki push
+//!
+//! # Usage
+//!
+//! ```ignore
+//! use weaver_common::telemetry::{self, TelemetryConfig};
+//!
+//! #[tokio::main]
+//! async fn main() {
+//!     // Initialize telemetry (metrics + tracing)
+//!     let config = TelemetryConfig::from_env("weaver-index");
+//!     telemetry::init(config).await;
+//!
+//!     // Mount the metrics endpoint in your axum router
+//!     let app = Router::new()
+//!         .route("/metrics", get(|| async { telemetry::render() }));
+//!
+//!     // Use metrics
+//!     metrics::counter!("requests_total").increment(1);
+//!
+//!     // Use tracing (goes to both console and loki if configured)
+//!     tracing::info!("server started");
+//! }
+//! ```
+
+use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
+use std::sync::OnceLock;
+use tracing::Level;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::{EnvFilter, Layer};
+
+static PROMETHEUS_HANDLE: OnceLock<PrometheusHandle> = OnceLock::new();
+
+/// Telemetry configuration
+#[derive(Debug, Clone)]
+pub struct TelemetryConfig {
+    /// Service name for labeling (e.g., "weaver-index", "weaver-app")
+    pub service_name: String,
+    /// Loki push URL (e.g., "http://localhost:3100"). None disables Loki.
+    pub loki_url: Option<String>,
+    /// Console log level (default: INFO, DEBUG in debug builds)
+    pub console_level: Level,
+}
+
+impl TelemetryConfig {
+    /// Load config from environment variables.
+    ///
+    /// - `LOKI_URL`: Loki push endpoint (optional)
+    /// - `RUST_LOG`: Standard env filter (optional, overrides console_level)
+    pub fn from_env(service_name: impl Into<String>) -> Self {
+        let console_level = if cfg!(debug_assertions) {
+            Level::DEBUG
+        } else {
+            Level::INFO
+        };
+
+        Self {
+            service_name: service_name.into(),
+            loki_url: std::env::var("LOKI_URL").ok(),
+            console_level,
+        }
+    }
+}
+
+/// Initialize telemetry (metrics + tracing).
+///
+/// Call once at application startup. If `LOKI_URL` is set, spawns a background
+/// task to push logs to Loki.
+pub async fn init(config: TelemetryConfig) {
+    // Initialize prometheus metrics
+    init_metrics();
+
+    // Initialize tracing
+    init_tracing(config).await;
+}
+
+/// Initialize just the prometheus metrics recorder.
+pub fn init_metrics() -> &'static PrometheusHandle {
+    PROMETHEUS_HANDLE.get_or_init(|| {
+        PrometheusBuilder::new()
+            .install_recorder()
+            .expect("failed to install prometheus recorder")
+    })
+}
+
+/// Initialize tracing with console + optional Loki layers.
+async fn init_tracing(config: TelemetryConfig) {
+    let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| {
+        EnvFilter::new(format!(
+            "{}",
+            config.console_level.as_str().to_lowercase()
+        ))
+    });
+
+    // Pretty console layer for human-readable stdout
+    let console_layer = tracing_subscriber::fmt::layer()
+        .with_target(true)
+        .with_thread_ids(false)
+        .with_file(false)
+        .with_line_number(false)
+        .compact()
+        .with_filter(env_filter);
+
+    // Optional Loki layer for structured logs
+    if let Some(loki_url) = config.loki_url {
+        match tracing_loki::url::Url::parse(&loki_url) {
+            Ok(url) => {
+                let (loki_layer, loki_task) = tracing_loki::builder()
+                    .label("service", config.service_name.clone())
+                    .expect("invalid label")
+                    .build_url(url)
+                    .expect("failed to build loki layer");
+
+                tracing_subscriber::registry()
+                    .with(console_layer)
+                    .with(loki_layer)
+                    .init();
+
+                // Spawn the background task that pushes to Loki
+                tokio::spawn(loki_task);
+
+                tracing::info!(
+                    service = %config.service_name,
+                    loki_url = %loki_url,
+                    "telemetry initialized with loki"
+                );
+            }
+            Err(e) => {
+                // Invalid URL - fall back to console only
+                tracing_subscriber::registry().with(console_layer).init();
+
+                tracing::warn!(
+                    error = %e,
+                    loki_url = %loki_url,
+                    "invalid LOKI_URL, falling back to console only"
+                );
+            }
+        }
+    } else {
+        // No Loki URL - console only
+        tracing_subscriber::registry().with(console_layer).init();
+
+        tracing::debug!(
+            service = %config.service_name,
+            "telemetry initialized (console only, set LOKI_URL to enable loki)"
+        );
+    }
+}
+
+/// Get the prometheus handle.
+pub fn handle() -> &'static PrometheusHandle {
+    PROMETHEUS_HANDLE.get_or_init(|| {
+        PrometheusBuilder::new()
+            .install_recorder()
+            .expect("failed to install prometheus recorder")
+    })
+}
+
+/// Render metrics in prometheus text format.
+pub fn render() -> String {
+    handle().render()
+}
+
+// Re-export the metrics crate for convenience
+pub use metrics::{counter, gauge, histogram};

crates/weaver-index/Cargo.toml

···
 [dependencies]
 # Internal
 weaver-api = { path = "../weaver-api", features = ["streaming"] }
+weaver-common = { path = "../weaver-common", features = ["telemetry"] }
 
 # AT Protocol / Jacquard
-jacquard = { workspace = true, features = ["websocket", "zstd"] }
+jacquard = { workspace = true, features = ["websocket", "zstd", "dns", "cache"] }
 jacquard-common = { workspace = true }
 jacquard-repo = { workspace = true }
+jacquard-axum = { workspace = true }
 
 # ClickHouse
 clickhouse = { version = "0.14", features = ["inserter", "chrono", "rustls-tls-ring", "rustls-tls-webpki-roots"] }
···
 base64 = "0.22"
 dashmap = "6"
 include_dir = "0.7.4"
+
+# WebSocket (for tap consumer)
+tokio-tungstenite = { version = "0.26", features = ["native-tls"] }
+futures-util = "0.3"
+
+# HTTP server
+axum = { version = "0.8.7", features = ["macros"] }
+tower = "0.5.2"
+tower-http = { version = "0.6.7", features = ["trace", "cors"] }
+
+# SQLite (shard storage)
+rusqlite = { version = "0.37.0", features = ["bundled"] }
+rusqlite_migration = { version = "2.1.0", features = ["from-directory"] }

crates/weaver-index/Dockerfile

···
+# Build stage
+FROM rust:1.91-trixie AS builder
+
+WORKDIR /build
+
+# Install build dependencies
+RUN apt-get update && apt-get install -y \
+    pkg-config \
+    libssl-dev \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy workspace files
+COPY Cargo.toml Cargo.lock ./
+COPY crates ./crates
+
+# Build release binary
+RUN cargo build --release -p weaver-index --bin indexer
+
+# Runtime stage
+FROM debian:trixie-slim
+
+RUN apt-get update && apt-get install -y \
+    ca-certificates \
+    libssl3 \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+# Copy binary from builder
+COPY --from=builder /build/target/release/indexer /app/indexer
+
+# Default environment variables
+ENV RUST_LOG=info
+ENV RUST_BACKTRACE=1
+ENV CLICKHOUSE_URL=http://clickhouse:8123
+ENV CLICKHOUSE_DATABASE=default
+
+# Run the indexer
+ENTRYPOINT ["/app/indexer"]

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

···
     -- Populated by async batch validation, not in hot path
     validation_state LowCardinality(String) DEFAULT 'unchecked',
 
+    -- Whether this came from live firehose (true) or backfill (false)
+    -- Backfill events may not reflect current state until repo is fully synced
+    is_live Bool DEFAULT true,
+
     -- Materialized AT URI for convenience
     uri String MATERIALIZED concat('at://', did, '/', collection, '/', rkey),
 

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

···
     let firehose_config = FirehoseConfig::from_env()?;
 
     info!(
-        "Connecting to ClickHouse at {} (database: {})",
+        "Connecting to ClickHouse at:\n{} (database: {})",
         ch_config.url, ch_config.database
     );
     let client = Client::new(&ch_config)?;
···
         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);
+    info!("Connecting to firehose at:\n {}", firehose_config.relay_url);
     let consumer = FirehoseConsumer::new(firehose_config);
     let mut stream = consumer.connect().await?;
 
···
         }
     }
 
-    // Final flush
     info!("Flushing remaining records...");
     json_inserter
         .end()
···
             source: e,
         })?;
 
-    // Final report
     info!("\n========== FINAL RESULTS ==========");
     report_progress(
         &client,
···
         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)",

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

···
 use clap::{Parser, Subcommand};
-use miette::IntoDiagnostic;
-use tracing::{Level, info, warn};
-use tracing_subscriber::EnvFilter;
+use tracing::{error, info, warn};
 use weaver_index::clickhouse::{Client, Migrator, Tables};
-use weaver_index::config::{ClickHouseConfig, FirehoseConfig, IndexerConfig};
+use weaver_index::config::{
+    ClickHouseConfig, FirehoseConfig, IndexerConfig, ShardConfig, SourceMode, TapConfig,
+};
 use weaver_index::firehose::FirehoseConsumer;
-use weaver_index::{Indexer, load_cursor};
+use weaver_index::server::{AppState, ServerConfig, TelemetryConfig, telemetry};
+use weaver_index::{FirehoseIndexer, TapIndexer, load_cursor};
 
 #[derive(Parser)]
 #[command(name = "indexer")]
···
     /// Check database connectivity
     Health,
 
-    /// Start the indexer service (not yet implemented)
+    /// Start the full service (indexer + HTTP server)
     Run,
+
+    /// Start only the HTTP server (no indexing)
+    Serve,
+
+    /// Start only the indexer (no HTTP server)
+    Index,
 }
 
 #[tokio::main]
 async fn main() -> miette::Result<()> {
     dotenvy::dotenv().ok();
 
-    let console_level = if cfg!(debug_assertions) {
-        Level::DEBUG
-    } else {
-        Level::INFO
-    };
-
-    tracing_subscriber::fmt()
-        .with_env_filter(
-            tracing_subscriber::EnvFilter::builder()
-                .from_env_lossy()
-                .add_directive(console_level.into())
-                .add_directive("hyper_util=info".parse().into_diagnostic()?),
-        )
-        .init();
+    // Initialize telemetry (metrics + tracing with optional Loki)
+    let telemetry_config = TelemetryConfig::from_env("weaver-index");
+    telemetry::init(telemetry_config).await;
 
     let args = Args::parse();
 
     match args.command {
         Command::Migrate { dry_run, reset } => run_migrate(dry_run, reset).await,
         Command::Health => run_health().await,
-        Command::Run => run_indexer().await,
+        Command::Run => run_full().await,
+        Command::Serve => run_server_only().await,
+        Command::Index => run_indexer_only().await,
     }
 }
 
···
     Ok(())
 }
 
-async fn run_indexer() -> miette::Result<()> {
+/// Run both indexer and HTTP server concurrently (production mode)
+async fn run_full() -> miette::Result<()> {
+    let ch_config = ClickHouseConfig::from_env()?;
+    let shard_config = ShardConfig::from_env();
+    let server_config = ServerConfig::from_env();
+    let indexer_config = IndexerConfig::from_env();
+    let source_mode = SourceMode::from_env();
+
+    info!(
+        "Connecting to ClickHouse at {} (database: {})",
+        ch_config.url, ch_config.database
+    );
+    info!("SQLite shards at {}", shard_config.base_path.display());
+
+    // Create separate clients for indexer and server
+    let indexer_client = Client::new(&ch_config)?;
+    let server_client = Client::new(&ch_config)?;
+
+    // Build AppState for server
+    let state = AppState::new(server_client, shard_config);
+
+    // Spawn the indexer task
+    let indexer_handle = match source_mode {
+        SourceMode::Firehose => {
+            let mut firehose_config = FirehoseConfig::from_env()?;
+            if firehose_config.cursor.is_none() {
+                if let Some(cursor) = load_cursor(&indexer_client).await? {
+                    firehose_config.cursor = Some(cursor);
+                }
+            }
+            info!(
+                "Connecting to firehose at {} (cursor: {:?})",
+                firehose_config.relay_url, firehose_config.cursor
+            );
+            let consumer = FirehoseConsumer::new(firehose_config);
+            let indexer = FirehoseIndexer::new(indexer_client, consumer, indexer_config).await?;
+            info!("Starting firehose indexer");
+            tokio::spawn(async move { indexer.run().await })
+        }
+        SourceMode::Tap => {
+            let tap_config = TapConfig::from_env()?;
+            let indexer = TapIndexer::new(indexer_client, tap_config, indexer_config);
+            info!("Starting tap indexer");
+            tokio::spawn(async move { indexer.run().await })
+        }
+    };
+
+    // Run server, monitoring indexer health
+    tokio::select! {
+        result = weaver_index::server::run(state, server_config) => {
+            result?;
+        }
+        result = indexer_handle => {
+            match result {
+                Ok(Ok(())) => info!("Indexer completed"),
+                Ok(Err(e)) => error!("Indexer failed: {}", e),
+                Err(e) => error!("Indexer task panicked: {}", e),
+            }
+        }
+    }
+
+    Ok(())
+}
+
+/// Run only the indexer (no HTTP server)
+async fn run_indexer_only() -> miette::Result<()> {
     let ch_config = ClickHouseConfig::from_env()?;
-    let mut firehose_config = FirehoseConfig::from_env()?;
     let indexer_config = IndexerConfig::from_env();
+    let source_mode = SourceMode::from_env();
 
     info!(
         "Connecting to ClickHouse at {} (database: {})",
···
     );
     let client = Client::new(&ch_config)?;
 
+    match source_mode {
+        SourceMode::Firehose => run_firehose_indexer(client, indexer_config).await,
+        SourceMode::Tap => {
+            let tap_config = TapConfig::from_env()?;
+            run_tap_indexer(client, tap_config, indexer_config).await
+        }
+    }
+}
+
+async fn run_firehose_indexer(client: Client, indexer_config: IndexerConfig) -> miette::Result<()> {
+    let mut firehose_config = FirehoseConfig::from_env()?;
+
     // Load cursor from ClickHouse if not overridden by env var
     if firehose_config.cursor.is_none() {
         if let Some(cursor) = load_cursor(&client).await? {
···
     );
     let consumer = FirehoseConsumer::new(firehose_config);
 
-    let indexer = Indexer::new(client, consumer, indexer_config).await?;
+    let indexer = FirehoseIndexer::new(client, consumer, indexer_config).await?;
+
+    info!("Starting firehose indexer");
+    indexer.run().await?;
 
-    info!("Starting indexer");
+    Ok(())
+}
+
+async fn run_tap_indexer(
+    client: Client,
+    tap_config: TapConfig,
+    indexer_config: IndexerConfig,
+) -> miette::Result<()> {
+    let indexer = TapIndexer::new(client, tap_config, indexer_config);
+
+    info!("Starting tap indexer");
     indexer.run().await?;
 
     Ok(())
 }
+
+async fn run_server_only() -> miette::Result<()> {
+    let ch_config = ClickHouseConfig::from_env()?;
+    let shard_config = ShardConfig::from_env();
+    let server_config = ServerConfig::from_env();
+
+    info!(
+        "Connecting to ClickHouse at {} (database: {})",
+        ch_config.url, ch_config.database
+    );
+    info!("SQLite shards at {}", shard_config.base_path.display());
+
+    let client = Client::new(&ch_config)?;
+
+    let state = AppState::new(client, shard_config);
+    weaver_index::server::run(state, server_config).await?;
+
+    Ok(())
+}

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

···
+use std::time::Duration;
+
 use crate::config::ClickHouseConfig;
 use crate::error::{ClickHouseError, IndexError};
 use clickhouse::Row;
 use clickhouse::inserter::Inserter;
+use serde::Deserialize;
 
 /// ClickHouse client wrapper with connection pooling and batched inserts
 pub struct Client {
···
             // Enable JSON type support (treated as string at transport level)
             .with_option("allow_experimental_json_type", "1")
             .with_option("input_format_binary_read_json_as_string", "1")
-            .with_option("output_format_binary_write_json_as_string", "1");
+            .with_option("output_format_binary_write_json_as_string", "1")
+            .with_option("send_timeout", "120")
+            .with_option("receive_timeout", "120");
 
         Ok(Self { inner })
     }
···
             .inserter(table)
             .with_max_rows(1000)
             .with_period_bias(0.1)
+            .with_period(Some(Duration::from_secs(1)))
             .with_max_bytes(1_048_576)
     }
 
···
     pub fn inner(&self) -> &clickhouse::Client {
         &self.inner
     }
+
+    /// Get a single record by (did, collection, rkey)
+    ///
+    /// Returns the latest non-deleted version from raw_records.
+    pub async fn get_record(
+        &self,
+        did: &str,
+        collection: &str,
+        rkey: &str,
+    ) -> Result<Option<RecordRow>, IndexError> {
+        // FINAL ensures ReplacingMergeTree deduplication is applied
+        let query = r#"
+            SELECT cid, record
+            FROM raw_records FINAL
+            WHERE did = ?
+              AND collection = ?
+              AND rkey = ?
+              AND operation != 'delete'
+            ORDER BY event_time DESC
+            LIMIT 1
+        "#;
+
+        let row = self
+            .inner
+            .query(query)
+            .bind(did)
+            .bind(collection)
+            .bind(rkey)
+            .fetch_optional::<RecordRow>()
+            .await
+            .map_err(|e| ClickHouseError::Query {
+                message: "failed to get record".into(),
+                source: e,
+            })?;
+
+        Ok(row)
+    }
+
+    /// Insert a single record (for cache-on-miss)
+    ///
+    /// Used when fetching a record from upstream that wasn't in our cache.
+    pub async fn insert_record(
+        &self,
+        did: &str,
+        collection: &str,
+        rkey: &str,
+        cid: &str,
+        record_json: &str,
+    ) -> Result<(), IndexError> {
+        use crate::clickhouse::schema::RawRecordInsert;
+        use chrono::DateTime;
+        use smol_str::SmolStr;
+
+        let row = RawRecordInsert {
+            did: SmolStr::new(did),
+            collection: SmolStr::new(collection),
+            rkey: SmolStr::new(rkey),
+            cid: SmolStr::new(cid),
+            rev: SmolStr::new_static(""), // Unknown from upstream fetch
+            record: SmolStr::new(record_json),
+            operation: SmolStr::new_static("cache"), // Distinguish from firehose ops
+            seq: 0,                                  // Not from firehose
+            event_time: DateTime::UNIX_EPOCH,        // Sort behind canonical firehose data
+            is_live: false,                          // Fetched on-demand, not from firehose
+        };
+
+        let mut insert = self
+            .inner
+            .insert::<RawRecordInsert>("raw_records")
+            .await
+            .map_err(|e| ClickHouseError::Insert {
+                message: "failed to create insert".into(),
+                source: e,
+            })?;
+
+        insert
+            .write(&row)
+            .await
+            .map_err(|e| ClickHouseError::Insert {
+                message: "failed to write record".into(),
+                source: e,
+            })?;
+
+        insert.end().await.map_err(|e| ClickHouseError::Insert {
+            message: "failed to flush insert".into(),
+            source: e,
+        })?;
+
+        Ok(())
+    }
+
+    /// List records for a repo+collection
+    ///
+    /// Returns non-deleted records ordered by rkey, with cursor-based pagination.
+    pub async fn list_records(
+        &self,
+        did: &str,
+        collection: &str,
+        limit: u32,
+        cursor: Option<&str>,
+        reverse: bool,
+    ) -> Result<Vec<RecordListRow>, IndexError> {
+        let order = if reverse { "DESC" } else { "ASC" };
+        let cursor_op = if reverse { "<" } else { ">" };
+
+        // Build query with optional cursor
+        let query = if cursor.is_some() {
+            format!(
+                r#"
+                SELECT rkey, cid, record
+                FROM raw_records FINAL
+                WHERE did = ?
+                  AND collection = ?
+                  AND rkey {cursor_op} ?
+                  AND operation != 'delete'
+                ORDER BY rkey {order}
+                LIMIT ?
+                "#,
+            )
+        } else {
+            format!(
+                r#"
+                SELECT rkey, cid, record
+                FROM raw_records FINAL
+                WHERE did = ?
+                  AND collection = ?
+                  AND operation != 'delete'
+                ORDER BY rkey {order}
+                LIMIT ?
+                "#,
+            )
+        };
+
+        let mut q = self.inner.query(&query).bind(did).bind(collection);
+
+        if let Some(cursor_rkey) = cursor {
+            q = q.bind(cursor_rkey);
+        }
+
+        let rows = q
+            .bind(limit)
+            .fetch_all::<RecordListRow>()
+            .await
+            .map_err(|e| ClickHouseError::Query {
+                message: "failed to list records".into(),
+                source: e,
+            })?;
+
+        Ok(rows)
+    }
 }
 
 /// Table size statistics from system.parts
···
         }
     }
 }
+
+/// Single record from raw_records (for getRecord)
+#[derive(Debug, Clone, Row, Deserialize)]
+pub struct RecordRow {
+    pub cid: String,
+    pub record: String, // JSON string
+}
+
+/// Record with rkey from raw_records (for listRecords)
+#[derive(Debug, Clone, Row, Deserialize)]
+pub struct RecordListRow {
+    pub rkey: String,
+    pub cid: String,
+    pub record: String, // JSON string
+}

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

···
 
 /// Validation states for records
 pub mod validation {
+    #[allow(dead_code)]
     pub const UNCHECKED: &str = "unchecked";
+    #[allow(dead_code)]
     pub const VALID: &str = "valid";
+    #[allow(dead_code)]
     pub const INVALID_REV: &str = "invalid_rev";
+    #[allow(dead_code)]
     pub const INVALID_GAP: &str = "invalid_gap";
+    #[allow(dead_code)]
     pub const INVALID_ACCOUNT: &str = "invalid_account";
 }
 
···
     pub seq: u64,
     #[serde(with = "clickhouse::serde::chrono::datetime64::millis")]
     pub event_time: DateTime<Utc>,
+    /// Whether this came from live firehose (true) or backfill (false)
+    pub is_live: bool,
     // Note: indexed_at has DEFAULT now64(3), omit from insert
     // Note: validation_state has DEFAULT 'unchecked', omit from insert
 }

crates/weaver-index/src/config.rs

···
     }
 }
 
+/// Tap connection configuration
+#[derive(Debug, Clone)]
+pub struct TapConfig {
+    pub url: Url,
+    pub send_acks: bool,
+}
+
+impl TapConfig {
+    /// Default tap URL (local)
+    pub const DEFAULT_URL: &'static str = "ws://localhost:2480/channel";
+
+    /// Load configuration from environment variables.
+    ///
+    /// Optional env vars:
+    /// - `TAP_URL`: Tap WebSocket URL (default: ws://localhost:2480/channel)
+    /// - `TAP_SEND_ACKS`: Whether to send acks (default: true)
+    pub fn from_env() -> Result<Self, IndexError> {
+        let url_str = std::env::var("TAP_URL").unwrap_or_else(|_| Self::DEFAULT_URL.to_string());
+
+        let url = Url::parse(&url_str).map_err(|e| ConfigError::UrlParse {
+            url: url_str,
+            message: e.to_string(),
+        })?;
+
+        let send_acks = std::env::var("TAP_SEND_ACKS")
+            .map(|s| s.to_lowercase() != "false")
+            .unwrap_or(true);
+
+        Ok(Self { url, send_acks })
+    }
+}
+
+/// Source mode for the indexer
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum SourceMode {
+    /// Direct firehose connection
+    #[default]
+    Firehose,
+    /// Consume from tap
+    Tap,
+}
+
+impl SourceMode {
+    pub fn from_env() -> Self {
+        match std::env::var("INDEXER_SOURCE").as_deref() {
+            Ok("tap") => SourceMode::Tap,
+            _ => SourceMode::Firehose,
+        }
+    }
+}
+
+/// SQLite shard configuration
+#[derive(Debug, Clone)]
+pub struct ShardConfig {
+    pub base_path: std::path::PathBuf,
+}
+
+impl Default for ShardConfig {
+    fn default() -> Self {
+        Self {
+            base_path: std::path::PathBuf::from("./shards"),
+        }
+    }
+}
+
+impl ShardConfig {
+    /// Load configuration from environment variables.
+    ///
+    /// Optional env vars:
+    /// - `SHARD_BASE_PATH`: Base directory for SQLite shards (default: ./shards)
+    pub fn from_env() -> Self {
+        let base_path = std::env::var("SHARD_BASE_PATH")
+            .map(std::path::PathBuf::from)
+            .unwrap_or_else(|_| std::path::PathBuf::from("./shards"));
+
+        Self { base_path }
+    }
+}
+
 /// Combined configuration for the indexer
 #[derive(Debug, Clone)]
 pub struct Config {
     pub clickhouse: ClickHouseConfig,
     pub firehose: FirehoseConfig,
+    pub tap: TapConfig,
     pub indexer: IndexerConfig,
+    pub shard: ShardConfig,
+    pub source: SourceMode,
 }
 
 impl Config {
···
         Ok(Self {
             clickhouse: ClickHouseConfig::from_env()?,
             firehose: FirehoseConfig::from_env()?,
+            tap: TapConfig::from_env()?,
             indexer: IndexerConfig::from_env(),
+            shard: ShardConfig::from_env(),
+            source: SourceMode::from_env(),
         })
     }
 }

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

···
+//! XRPC endpoint handlers for the appview.
+
+pub mod repo;

crates/weaver-index/src/endpoints/repo.rs

···
+//! com.atproto.repo.* endpoint handlers
+//!
+//! These serve as a record cache, reading from the raw_records table
+//! populated by firehose/tap ingestion. On cache miss, fetches from
+//! upstream via Slingshot and caches the result.
+
+use axum::{Json, extract::State, http::StatusCode, response::IntoResponse};
+use jacquard::IntoStatic;
+use jacquard::api::com_atproto::repo::{
+    get_record::{GetRecordOutput, GetRecordRequest},
+    list_records::{ListRecordsOutput, ListRecordsRequest, Record},
+};
+use jacquard::client::AgentSessionExt;
+use jacquard::identity::resolver::IdentityResolver;
+use jacquard::types::ident::AtIdentifier;
+use jacquard::types::string::{AtUri, Cid};
+use jacquard::types::value::Data;
+use jacquard_axum::ExtractXrpc;
+use serde_json::json;
+
+use crate::server::AppState;
+
+/// Error response for XRPC endpoints
+pub struct XrpcErrorResponse {
+    pub status: StatusCode,
+    pub error: String,
+    pub message: Option<String>,
+}
+
+impl XrpcErrorResponse {
+    pub fn not_found(message: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::NOT_FOUND,
+            error: "RecordNotFound".to_string(),
+            message: Some(message.into()),
+        }
+    }
+
+    pub fn invalid_request(message: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::BAD_REQUEST,
+            error: "InvalidRequest".to_string(),
+            message: Some(message.into()),
+        }
+    }
+
+    pub fn internal_error(message: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::INTERNAL_SERVER_ERROR,
+            error: "InternalServerError".to_string(),
+            message: Some(message.into()),
+        }
+    }
+}
+
+impl IntoResponse for XrpcErrorResponse {
+    fn into_response(self) -> axum::response::Response {
+        let body = json!({
+            "error": self.error,
+            "message": self.message,
+        });
+        (self.status, Json(body)).into_response()
+    }
+}
+
+/// Handle com.atproto.repo.getRecord
+///
+/// Fetches a single record from the raw_records cache. On cache miss,
+/// fetches from upstream via Slingshot and caches the result.
+pub async fn get_record(
+    State(state): State<AppState>,
+    ExtractXrpc(args): ExtractXrpc<GetRecordRequest>,
+) -> Result<Json<GetRecordOutput<'static>>, XrpcErrorResponse> {
+    // Resolve identifier to DID
+    let did = match &args.repo {
+        AtIdentifier::Did(did) => did.clone(),
+        AtIdentifier::Handle(handle) => {
+            state.resolver.resolve_handle(handle).await.map_err(|e| {
+                tracing::warn!("Handle resolution failed for {}: {}", handle, e);
+                XrpcErrorResponse::invalid_request(format!("Could not resolve handle: {}", handle))
+            })?
+        }
+    };
+
+    let collection = args.collection.as_str();
+    let rkey: &str = args.rkey.as_ref();
+
+    // Query ClickHouse for the record
+    let cached = state
+        .clickhouse
+        .get_record(did.as_str(), collection, rkey)
+        .await
+        .map_err(|e| {
+            tracing::error!("ClickHouse query failed: {}", e);
+            XrpcErrorResponse::internal_error("Database query failed")
+        })?;
+
+    if let Some(row) = cached {
+        // Cache hit - return from ClickHouse
+        let value: Data<'_> = serde_json::from_str(&row.record).map_err(|e| {
+            tracing::error!("Failed to parse record JSON: {}", e);
+            XrpcErrorResponse::internal_error("Failed to parse stored record")
+        })?;
+
+        let uri_str = format!("at://{}/{}/{}", did, collection, rkey);
+        let uri = AtUri::new_owned(uri_str).map_err(|e| {
+            tracing::error!("Failed to construct AT URI: {}", e);
+            XrpcErrorResponse::internal_error("Failed to construct URI")
+        })?;
+
+        let cid = Cid::new(row.cid.as_bytes()).map_err(|e| {
+            tracing::error!("Invalid CID in database: {}", e);
+            XrpcErrorResponse::internal_error("Invalid CID stored")
+        })?;
+
+        return Ok(Json(
+            GetRecordOutput {
+                cid: Some(cid),
+                uri,
+                value,
+                extra_data: None,
+            }
+            .into_static(),
+        ));
+    }
+
+    // Cache miss - fetch from Slingshot
+    tracing::debug!(
+        "Cache miss for {}/{}/{}, fetching from Slingshot",
+        did,
+        collection,
+        rkey
+    );
+
+    let uri_str = format!("at://{}/{}/{}", did, collection, rkey);
+    let uri = AtUri::new_owned(uri_str.clone()).map_err(|e| {
+        tracing::error!("Failed to construct AT URI: {}", e);
+        XrpcErrorResponse::internal_error("Failed to construct URI")
+    })?;
+
+    let upstream = state
+        .resolver
+        .fetch_record_slingshot(&uri)
+        .await
+        .map_err(|e| {
+            tracing::warn!("Slingshot fetch failed for {}: {}", uri, e);
+            XrpcErrorResponse::not_found("Record not found")
+        })?;
+
+    // Cache the fetched record (fire-and-forget, don't block response)
+    let cid_str = upstream
+        .cid
+        .as_ref()
+        .map(|c| c.as_str().to_string())
+        .unwrap_or_default();
+    let record_json = serde_json::to_string(&upstream.value).unwrap_or_default();
+
+    if !cid_str.is_empty() && !record_json.is_empty() {
+        let clickhouse = state.clickhouse.clone();
+        let did_str = did.as_str().to_string();
+        let collection_str = collection.to_string();
+        let rkey_str = rkey.to_string();
+
+        tokio::spawn(async move {
+            if let Err(e) = clickhouse
+                .insert_record(&did_str, &collection_str, &rkey_str, &cid_str, &record_json)
+                .await
+            {
+                tracing::warn!("Failed to cache fetched record: {}", e);
+            }
+        });
+    }
+
+    Ok(Json(upstream))
+}
+
+/// Handle com.atproto.repo.listRecords
+///
+/// Lists records for a repo+collection from the raw_records cache.
+pub async fn list_records(
+    State(state): State<AppState>,
+    ExtractXrpc(args): ExtractXrpc<ListRecordsRequest>,
+) -> Result<Json<ListRecordsOutput<'static>>, XrpcErrorResponse> {
+    // Resolve identifier to DID
+    let did = match &args.repo {
+        AtIdentifier::Did(did) => did.clone(),
+        AtIdentifier::Handle(handle) => {
+            state.resolver.resolve_handle(handle).await.map_err(|e| {
+                tracing::warn!("Handle resolution failed for {}: {}", handle, e);
+                XrpcErrorResponse::invalid_request(format!("Could not resolve handle: {}", handle))
+            })?
+        }
+    };
+
+    let collection = args.collection.as_str();
+    let limit = args.limit.unwrap_or(50).clamp(1, 100) as u32;
+    let cursor = args.cursor.as_deref();
+    let reverse = args.reverse.unwrap_or(false);
+
+    // Query ClickHouse for records
+    let rows = state
+        .clickhouse
+        .list_records(did.as_str(), collection, limit, cursor, reverse)
+        .await
+        .map_err(|e| {
+            tracing::error!("ClickHouse query failed: {}", e);
+            XrpcErrorResponse::internal_error("Database query failed")
+        })?;
+
+    // Convert rows to Record output
+    let mut records = Vec::with_capacity(rows.len());
+    for row in &rows {
+        let value: Data<'_> = serde_json::from_str(&row.record).map_err(|e| {
+            tracing::error!("Failed to parse record JSON: {}", e);
+            XrpcErrorResponse::internal_error("Failed to parse stored record")
+        })?;
+
+        let uri_str = format!("at://{}/{}/{}", did, collection, row.rkey);
+        let uri = AtUri::new_owned(uri_str)
+            .map_err(|_| XrpcErrorResponse::internal_error("Failed to construct URI"))?;
+
+        let cid = Cid::new(row.cid.as_bytes())
+            .map_err(|_| XrpcErrorResponse::internal_error("Invalid CID stored"))?;
+
+        records.push(
+            Record {
+                uri,
+                cid,
+                value,
+                extra_data: None,
+            }
+            .into_static(),
+        );
+    }
+
+    // Cursor is the rkey of the last record, if we have more
+    let next_cursor = if records.len() == limit as usize {
+        rows.last().map(|r| r.rkey.clone().into())
+    } else {
+        None
+    };
+
+    Ok(Json(ListRecordsOutput {
+        records,
+        cursor: next_cursor,
+        extra_data: None,
+    }))
+}

crates/weaver-index/src/error.rs

···
+use std::path::PathBuf;
+
 use miette::Diagnostic;
 use thiserror::Error;
 
···
     #[error(transparent)]
     #[diagnostic(transparent)]
     Config(#[from] ConfigError),
+
+    #[error(transparent)]
+    #[diagnostic(transparent)]
+    Server(#[from] ServerError),
+
+    #[error(transparent)]
+    #[diagnostic(transparent)]
+    Sqlite(#[from] SqliteError),
+}
+
+/// HTTP server errors
+#[derive(Debug, Error, Diagnostic)]
+pub enum ServerError {
+    #[error("failed to bind to {addr}")]
+    #[diagnostic(code(server::bind))]
+    Bind {
+        addr: std::net::SocketAddr,
+        #[source]
+        source: std::io::Error,
+    },
+
+    #[error("server terminated unexpectedly")]
+    #[diagnostic(code(server::serve))]
+    Serve {
+        #[source]
+        source: std::io::Error,
+    },
+}
+
+/// SQLite shard errors
+#[derive(Debug, Error, Diagnostic)]
+pub enum SqliteError {
+    #[error("failed to open database at {}", path.display())]
+    #[diagnostic(code(sqlite::open))]
+    Open {
+        path: PathBuf,
+        #[source]
+        source: rusqlite::Error,
+    },
+
+    #[error("failed to create directory {}", path.display())]
+    #[diagnostic(code(sqlite::io))]
+    Io {
+        path: PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
+
+    #[error("failed to set pragma {pragma}")]
+    #[diagnostic(code(sqlite::pragma))]
+    Pragma {
+        pragma: &'static str,
+        #[source]
+        source: rusqlite::Error,
+    },
+
+    #[error("migration failed: {message}")]
+    #[diagnostic(code(sqlite::migration))]
+    Migration { message: String },
+
+    #[error("query failed: {message}")]
+    #[diagnostic(code(sqlite::query))]
+    Query { message: String },
+
+    #[error("shard lock poisoned")]
+    #[diagnostic(code(sqlite::lock))]
+    LockPoisoned,
 }
 
 /// ClickHouse database errors

crates/weaver-index/src/indexer.rs

···
 use dashmap::DashMap;
 use n0_future::StreamExt;
 use smol_str::{SmolStr, ToSmolStr};
-use tracing::{debug, info, warn};
+use tracing::{debug, info, trace, warn};
 
 use chrono::DateTime;
 
···
     AccountRevState, Client, FirehoseCursor, RawAccountEvent, RawIdentityEvent, RawRecordInsert,
 };
 use crate::config::IndexerConfig;
-use crate::error::{IndexError, Result};
+use crate::config::TapConfig;
+use crate::error::{ClickHouseError, IndexError, Result};
 use crate::firehose::{
     Account, Commit, ExtractedRecord, FirehoseConsumer, Identity, MessageStream,
     SubscribeReposMessage, extract_records,
 };
+use crate::tap::{TapConfig as TapConsumerConfig, TapConsumer, TapEvent};
 
 /// Default consumer ID for cursor tracking
 const CONSUMER_ID: &str = "main";
···
     }
 }
 
-/// Main indexer that consumes firehose and writes to ClickHouse
-pub struct Indexer {
+/// Firehose indexer that consumes AT Protocol firehose and writes to ClickHouse
+pub struct FirehoseIndexer {
     client: Arc<Client>,
     consumer: FirehoseConsumer,
     rev_cache: RevCache,
     config: IndexerConfig,
 }
 
-impl Indexer {
-    /// Create a new indexer
+impl FirehoseIndexer {
+    /// Create a new firehose indexer
     pub async fn new(
         client: Client,
         consumer: FirehoseConsumer,
···
 
         info!("starting indexer loop");
 
-        while let Some(result) = stream.next().await {
+        loop {
+            // Get time until next required flush - must commit before socket timeout (30s)
+            let records_time = records.time_left().unwrap_or(Duration::from_secs(10));
+            let identities_time = identities.time_left().unwrap_or(Duration::from_secs(10));
+            let accounts_time = accounts.time_left().unwrap_or(Duration::from_secs(10));
+            let time_left = records_time.min(identities_time).min(accounts_time);
+
+            let result =
+                match tokio::time::timeout(time_left, stream.next()).await {
+                    Ok(Some(result)) => result,
+                    Ok(None) => {
+                        // Stream ended
+                        break;
+                    }
+                    Err(_) => {
+                        // Timeout - flush inserters to keep INSERT alive
+                        debug!("flush timeout, committing inserters");
+                        records.commit().await.map_err(|e| {
+                            crate::error::ClickHouseError::Query {
+                                message: "periodic records commit failed".into(),
+                                source: e,
+                            }
+                        })?;
+                        identities.commit().await.map_err(|e| {
+                            crate::error::ClickHouseError::Query {
+                                message: "periodic identities commit failed".into(),
+                                source: e,
+                            }
+                        })?;
+                        accounts.commit().await.map_err(|e| {
+                            crate::error::ClickHouseError::Query {
+                                message: "periodic accounts commit failed".into(),
+                                source: e,
+                            }
+                        })?;
+                        continue;
+                    }
+                };
+
             let msg = match result {
                 Ok(msg) => msg,
                 Err(e) => {
···
                     operation: record.operation.clone(),
                     seq: record.seq as u64,
                     event_time: record.event_time,
+                    is_live: true,
                 })
                 .await
                 .map_err(|e| crate::error::ClickHouseError::Query {
···
 /// Minimal struct for delete lookups - just the fields we need to process the delete
 #[derive(Debug, Clone, clickhouse::Row, serde::Deserialize)]
 struct LookupRawRecord {
+    #[allow(dead_code)]
     did: SmolStr,
+    #[allow(dead_code)]
     collection: SmolStr,
+    #[allow(dead_code)]
     rkey: SmolStr,
+    #[allow(dead_code)]
     record: SmolStr, // JSON string of the original record
 }
 
···
         tokio::time::sleep(Duration::from_secs(1)).await;
     }
 }
+
+// ============================================================================
+// TapIndexer - consumes from tap websocket
+// ============================================================================
+
+/// Consumer ID for tap cursor tracking
+const TAP_CONSUMER_ID: &str = "tap";
+
+/// Tap indexer that consumes from tap websocket and writes to ClickHouse
+pub struct TapIndexer {
+    client: Arc<Client>,
+    tap_config: TapConfig,
+    config: IndexerConfig,
+}
+
+impl TapIndexer {
+    /// Create a new tap indexer
+    pub fn new(client: Client, tap_config: TapConfig, config: IndexerConfig) -> Self {
+        Self {
+            client: Arc::new(client),
+            tap_config,
+            config,
+        }
+    }
+
+    /// Save tap cursor to ClickHouse for visibility
+    async fn save_cursor(&self, seq: u64) -> Result<()> {
+        let query = format!(
+            "INSERT INTO firehose_cursor (consumer_id, seq, event_time) VALUES ('{}', {}, now64(3))",
+            TAP_CONSUMER_ID, seq
+        );
+
+        self.client.execute(&query).await?;
+        debug!(seq, "saved tap cursor");
+        Ok(())
+    }
+
+    /// Run the tap indexer loop
+    pub async fn run(&self) -> Result<()> {
+        info!(url = %self.tap_config.url, "connecting to tap...");
+
+        let consumer_config = TapConsumerConfig::new(self.tap_config.url.clone())
+            .with_acks(self.tap_config.send_acks);
+        let consumer = TapConsumer::new(consumer_config);
+
+        let (mut events, ack_tx) = consumer.connect().await?;
+
+        let mut records = self.client.inserter::<RawRecordInsert>("raw_records");
+        let mut identities = self
+            .client
+            .inserter::<RawIdentityEvent>("raw_identity_events");
+
+        let mut processed: u64 = 0;
+        let mut last_seq: u64 = 0;
+        let mut last_stats = Instant::now();
+        let mut last_cursor_save = Instant::now();
+
+        info!("starting tap indexer loop");
+
+        loop {
+            // Get time until next required flush - must commit before socket timeout (30s)
+            let records_time = records.time_left().unwrap_or(Duration::from_secs(10));
+            let identities_time = identities.time_left().unwrap_or(Duration::from_secs(10));
+            let time_left = records_time.min(identities_time);
+
+            let event = match tokio::time::timeout(time_left, events.recv()).await {
+                Ok(Some(event)) => event,
+                Ok(None) => {
+                    // Channel closed, exit loop
+                    break;
+                }
+                Err(_) => {
+                    // Timeout - flush inserters to keep INSERT alive
+                    trace!("flush timeout, committing inserters");
+                    records.commit().await.map_err(|e| ClickHouseError::Query {
+                        message: "periodic records commit failed".into(),
+                        source: e,
+                    })?;
+                    identities
+                        .commit()
+                        .await
+                        .map_err(|e| ClickHouseError::Query {
+                            message: "periodic identities commit failed".into(),
+                            source: e,
+                        })?;
+                    continue;
+                }
+            };
+
+            let event_id = event.id();
+            last_seq = event_id;
+
+            match event {
+                TapEvent::Record(envelope) => {
+                    let record = &envelope.record;
+
+                    // Collection filter
+                    if !self.config.collections.matches(&record.collection) {
+                        // Still ack even if filtered
+                        let _ = ack_tx.send(event_id).await;
+                        continue;
+                    }
+
+                    let json = record
+                        .record
+                        .as_ref()
+                        .map(|v| serde_json::to_string(v).unwrap_or_default())
+                        .unwrap_or_default();
+
+                    debug!(
+                        op = record.action.as_str(),
+                        id = event_id,
+                        len = json.len(),
+                        "writing record"
+                    );
+
+                    records
+                        .write(&RawRecordInsert {
+                            did: record.did.clone(),
+                            collection: record.collection.clone(),
+                            rkey: record.rkey.clone(),
+                            cid: record.cid.clone(),
+                            rev: record.rev.clone(),
+                            record: json.to_smolstr(),
+                            operation: record.action.as_str().to_smolstr(),
+                            seq: event_id,
+                            event_time: Utc::now(),
+                            is_live: record.live,
+                        })
+                        .await
+                        .map_err(|e| ClickHouseError::Query {
+                            message: "record write failed".into(),
+                            source: e,
+                        })?;
+                    records.commit().await.map_err(|e| ClickHouseError::Query {
+                        message: format!("record commit failed for id {}", event_id),
+                        source: e,
+                    })?;
+
+                    processed += 1;
+                }
+                TapEvent::Identity(envelope) => {
+                    let identity = &envelope.identity;
+
+                    identities
+                        .write(&RawIdentityEvent {
+                            did: identity.did.clone(),
+                            handle: identity.handle.clone(),
+                            seq: event_id,
+                            event_time: Utc::now(),
+                        })
+                        .await
+                        .map_err(|e| ClickHouseError::Query {
+                            message: "identity write failed".into(),
+                            source: e,
+                        })?;
+                    identities
+                        .commit()
+                        .await
+                        .map_err(|e| ClickHouseError::Query {
+                            message: "identity commit failed".into(),
+                            source: e,
+                        })?;
+                }
+            }
+
+            // Send ack after successful write+commit
+            let _ = ack_tx.send(event_id).await;
+
+            // Periodic stats
+            if last_stats.elapsed() >= Duration::from_secs(10) {
+                info!(processed, last_seq, "tap indexer stats");
+                last_stats = Instant::now();
+            }
+
+            // Save cursor every 30s for visibility
+            if last_cursor_save.elapsed() >= Duration::from_secs(30) && last_seq > 0 {
+                if let Err(e) = self.save_cursor(last_seq).await {
+                    warn!(error = ?e, "failed to save tap cursor");
+                }
+                last_cursor_save = Instant::now();
+            }
+        }
+
+        // Final flush
+        records.end().await.map_err(|e| ClickHouseError::Query {
+            message: "final records flush failed".into(),
+            source: e,
+        })?;
+        identities.end().await.map_err(|e| ClickHouseError::Query {
+            message: "final identities flush failed".into(),
+            source: e,
+        })?;
+
+        // Final cursor save
+        if last_seq > 0 {
+            self.save_cursor(last_seq).await?;
+        }
+
+        info!(last_seq, "tap stream ended");
+        Ok(())
+    }
+}

crates/weaver-index/src/lib.rs

···
 pub mod clickhouse;
 pub mod config;
+pub mod endpoints;
 pub mod error;
 pub mod firehose;
 pub mod indexer;
+pub mod server;
+pub mod sqlite;
+pub mod tap;
 
 pub use config::Config;
 pub use error::{IndexError, Result};
-pub use indexer::{load_cursor, Indexer};
+pub use indexer::{FirehoseIndexer, TapIndexer, load_cursor};
+pub use server::{AppState, ServerConfig};
+pub use sqlite::{ShardKey, ShardRouter, SqliteShard};

crates/weaver-index/src/server.rs

···
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use axum::{Json, Router, extract::State, http::StatusCode, response::IntoResponse, routing::get};
+use jacquard::api::com_atproto::repo::{
+    get_record::GetRecordRequest, list_records::ListRecordsRequest,
+};
+use jacquard::client::UnauthenticatedSession;
+use jacquard::identity::JacquardResolver;
+use jacquard_axum::IntoRouter;
+use serde::Serialize;
+use tower_http::trace::TraceLayer;
+use tracing::info;
+
+use crate::clickhouse::Client;
+use crate::config::ShardConfig;
+use crate::endpoints::repo;
+use crate::error::{IndexError, ServerError};
+use crate::sqlite::ShardRouter;
+
+pub use weaver_common::telemetry::{self, TelemetryConfig};
+
+/// Identity resolver type (unauthenticated, just for handle/DID resolution)
+pub type Resolver = UnauthenticatedSession<JacquardResolver>;
+
+/// Shared application state
+#[derive(Clone)]
+pub struct AppState {
+    pub clickhouse: Arc<Client>,
+    pub shards: Arc<ShardRouter>,
+    pub resolver: Resolver,
+}
+
+impl AppState {
+    pub fn new(clickhouse: Client, shard_config: ShardConfig) -> Self {
+        Self {
+            clickhouse: Arc::new(clickhouse),
+            shards: Arc::new(ShardRouter::new(shard_config.base_path)),
+            resolver: UnauthenticatedSession::new_slingshot(),
+        }
+    }
+}
+
+/// Build the axum router with all XRPC endpoints
+pub fn router(state: AppState) -> Router {
+    Router::new()
+        .route("/xrpc/_health", get(health))
+        .route("/metrics", get(metrics))
+        // com.atproto.repo.* endpoints (record cache)
+        .merge(GetRecordRequest::into_router(repo::get_record))
+        .merge(ListRecordsRequest::into_router(repo::list_records))
+        .layer(TraceLayer::new_for_http())
+        .with_state(state)
+}
+
+/// Prometheus metrics endpoint
+async fn metrics() -> String {
+    telemetry::render()
+}
+
+/// Health check response
+#[derive(Serialize)]
+struct HealthResponse {
+    status: &'static str,
+    clickhouse: bool,
+    shard_count: usize,
+}
+
+/// Health check endpoint
+///
+/// Returns 200 OK with stats if healthy, 503 if ClickHouse unreachable.
+async fn health(State(state): State<AppState>) -> impl IntoResponse {
+    let clickhouse_ok = state.clickhouse.execute("SELECT 1").await.is_ok();
+    let shard_count = state.shards.shard_count();
+
+    let response = HealthResponse {
+        status: if clickhouse_ok { "ok" } else { "degraded" },
+        clickhouse: clickhouse_ok,
+        shard_count,
+    };
+
+    let status = if clickhouse_ok {
+        StatusCode::OK
+    } else {
+        StatusCode::SERVICE_UNAVAILABLE
+    };
+
+    (status, Json(response))
+}
+
+/// Server configuration
+#[derive(Debug, Clone)]
+pub struct ServerConfig {
+    pub host: String,
+    pub port: u16,
+}
+
+impl Default for ServerConfig {
+    fn default() -> Self {
+        Self {
+            host: "0.0.0.0".to_string(),
+            port: 3000,
+        }
+    }
+}
+
+impl ServerConfig {
+    pub fn from_env() -> Self {
+        let host = std::env::var("SERVER_HOST").unwrap_or_else(|_| "0.0.0.0".to_string());
+        let port = std::env::var("SERVER_PORT")
+            .ok()
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(3000);
+
+        Self { host, port }
+    }
+
+    pub fn addr(&self) -> SocketAddr {
+        format!("{}:{}", self.host, self.port)
+            .parse()
+            .expect("valid socket address")
+    }
+}
+
+/// Run the HTTP server
+pub async fn run(state: AppState, config: ServerConfig) -> Result<(), IndexError> {
+    let addr = config.addr();
+    let app = router(state);
+
+    info!("Starting HTTP server on {}", addr);
+
+    let listener = tokio::net::TcpListener::bind(addr)
+        .await
+        .map_err(|e| ServerError::Bind { addr, source: e })?;
+
+    axum::serve(listener, app)
+        .await
+        .map_err(|e| ServerError::Serve { source: e })?;
+
+    Ok(())
+}

crates/weaver-index/src/sqlite.rs

···
+use std::collections::hash_map::DefaultHasher;
+use std::fs;
+use std::hash::{Hash, Hasher};
+use std::path::{Path, PathBuf};
+use std::sync::Mutex;
+use std::time::Instant;
+
+use dashmap::DashMap;
+use rusqlite::Connection;
+use rusqlite_migration::{M, Migrations};
+use smol_str::SmolStr;
+
+use crate::error::{IndexError, SqliteError};
+
+/// Key for shard routing - (collection, rkey) tuple
+#[derive(Debug, Clone, Hash, PartialEq, Eq)]
+pub struct ShardKey {
+    pub collection: SmolStr,
+    pub rkey: SmolStr,
+}
+
+impl ShardKey {
+    pub fn new(collection: impl Into<SmolStr>, rkey: impl Into<SmolStr>) -> Self {
+        Self {
+            collection: collection.into(),
+            rkey: rkey.into(),
+        }
+    }
+
+    fn hash_prefix(&self) -> String {
+        let mut hasher = DefaultHasher::new();
+        self.hash(&mut hasher);
+        let hash = hasher.finish();
+        format!("{:02x}", (hash & 0xFF) as u8)
+    }
+
+    /// Directory path: {base}/{hash(collection,rkey)[0..2]}/{rkey}/
+    fn dir_path(&self, base: &Path) -> PathBuf {
+        base.join(self.hash_prefix()).join(self.rkey.as_str())
+    }
+
+    pub fn collection(&self) -> &str {
+        &self.collection
+    }
+
+    pub fn rkey(&self) -> &str {
+        &self.rkey
+    }
+}
+
+/// A single SQLite shard for a resource
+pub struct SqliteShard {
+    conn: Mutex<Connection>,
+    path: PathBuf,
+    last_accessed: Mutex<Instant>,
+}
+
+impl SqliteShard {
+    const DB_FILENAME: &'static str = "store.sqlite";
+
+    fn open(dir: &Path) -> Result<Self, IndexError> {
+        fs::create_dir_all(dir).map_err(|e| SqliteError::Io {
+            path: dir.to_path_buf(),
+            source: e,
+        })?;
+
+        let db_path = dir.join(Self::DB_FILENAME);
+        let mut conn = Connection::open(&db_path).map_err(|e| SqliteError::Open {
+            path: db_path.clone(),
+            source: e,
+        })?;
+
+        // Enable WAL mode for better concurrency
+        conn.pragma_update(None, "journal_mode", "WAL")
+            .map_err(|e| SqliteError::Pragma {
+                pragma: "journal_mode",
+                source: e,
+            })?;
+
+        // Run migrations
+        // PERF: rusqlite_migration checks user_version pragma, which is fast when
+        // no migrations needed. If shard open becomes a bottleneck, consider adding
+        // a signal file (e.g., .schema_v{N}) to skip migration check entirely.
+        Self::migrations()
+            .to_latest(&mut conn)
+            .map_err(|e| SqliteError::Migration {
+                message: e.to_string(),
+            })?;
+
+        Ok(Self {
+            conn: Mutex::new(conn),
+            path: db_path,
+            last_accessed: Mutex::new(Instant::now()),
+        })
+    }
+
+    fn migrations() -> Migrations<'static> {
+        Migrations::new(vec![
+            M::up(include_str!("sqlite/migrations/001_edit_graph.sql")),
+            M::up(include_str!("sqlite/migrations/002_collaboration.sql")),
+            M::up(include_str!("sqlite/migrations/003_permissions.sql")),
+        ])
+    }
+
+    pub fn path(&self) -> &Path {
+        &self.path
+    }
+
+    pub fn touch(&self) {
+        if let Ok(mut last) = self.last_accessed.lock() {
+            *last = Instant::now();
+        }
+    }
+
+    pub fn last_accessed(&self) -> Instant {
+        self.last_accessed
+            .lock()
+            .map(|t| *t)
+            .unwrap_or_else(|_| Instant::now())
+    }
+
+    /// Execute a read operation on the shard
+    pub fn read<F, T>(&self, f: F) -> Result<T, IndexError>
+    where
+        F: FnOnce(&Connection) -> Result<T, rusqlite::Error>,
+    {
+        self.touch();
+        let conn = self.conn.lock().map_err(|_| SqliteError::LockPoisoned)?;
+        f(&conn).map_err(|e| {
+            SqliteError::Query {
+                message: e.to_string(),
+            }
+            .into()
+        })
+    }
+
+    /// Execute a write operation on the shard
+    pub fn write<F, T>(&self, f: F) -> Result<T, IndexError>
+    where
+        F: FnOnce(&Connection) -> Result<T, rusqlite::Error>,
+    {
+        self.touch();
+        let conn = self.conn.lock().map_err(|_| SqliteError::LockPoisoned)?;
+        f(&conn).map_err(|e| {
+            SqliteError::Query {
+                message: e.to_string(),
+            }
+            .into()
+        })
+    }
+}
+
+/// Routes resources to their SQLite shards
+pub struct ShardRouter {
+    base_path: PathBuf,
+    shards: DashMap<ShardKey, std::sync::Arc<SqliteShard>>,
+}
+
+impl ShardRouter {
+    pub fn new(base_path: impl Into<PathBuf>) -> Self {
+        Self {
+            base_path: base_path.into(),
+            shards: DashMap::new(),
+        }
+    }
+
+    /// Get or create a shard for the given key
+    pub fn get_or_create(&self, key: &ShardKey) -> Result<std::sync::Arc<SqliteShard>, IndexError> {
+        // Fast path: already cached
+        if let Some(shard) = self.shards.get(key) {
+            shard.touch();
+            return Ok(shard.clone());
+        }
+
+        // Slow path: create new shard
+        let dir = key.dir_path(&self.base_path);
+        let shard = std::sync::Arc::new(SqliteShard::open(&dir)?);
+        self.shards.insert(key.clone(), shard.clone());
+
+        Ok(shard)
+    }
+
+    /// Get an existing shard without creating
+    pub fn get(&self, key: &ShardKey) -> Option<std::sync::Arc<SqliteShard>> {
+        self.shards.get(key).map(|s| {
+            s.touch();
+            s.clone()
+        })
+    }
+
+    /// Number of active shards
+    pub fn shard_count(&self) -> usize {
+        self.shards.len()
+    }
+
+    /// Iterate over shards that haven't been accessed since the given instant
+    pub fn idle_shards(&self, since: Instant) -> Vec<ShardKey> {
+        self.shards
+            .iter()
+            .filter(|entry| entry.value().last_accessed() < since)
+            .map(|entry| entry.key().clone())
+            .collect()
+    }
+
+    /// Remove a shard from the cache (for eviction)
+    pub fn evict(&self, key: &ShardKey) -> Option<std::sync::Arc<SqliteShard>> {
+        self.shards.remove(key).map(|(_, shard)| shard)
+    }
+}

crates/weaver-index/src/sqlite/migrations/001_edit_graph.sql

···
+-- Edit graph storage (roots and diffs)
+-- Supports DAG structure for future merge support
+
+CREATE TABLE edit_nodes (
+    -- Edit record identity (decomposed)
+    did TEXT NOT NULL,
+    collection TEXT NOT NULL,  -- 'sh.weaver.edit.root' or 'sh.weaver.edit.diff'
+    rkey TEXT NOT NULL,
+
+    -- Resource being edited (decomposed)
+    resource_did TEXT NOT NULL,
+    resource_collection TEXT NOT NULL,
+    resource_rkey TEXT NOT NULL,
+
+    node_type TEXT NOT NULL,  -- 'root' | 'diff'
+    created_at TEXT NOT NULL,
+    indexed_at TEXT NOT NULL,
+
+    PRIMARY KEY (did, collection, rkey)
+);
+
+CREATE INDEX idx_edit_nodes_resource ON edit_nodes(resource_did, resource_collection, resource_rkey);
+CREATE INDEX idx_edit_nodes_author ON edit_nodes(did);
+
+-- Edit graph edges (supports DAG)
+CREATE TABLE edit_edges (
+    -- Child reference (decomposed)
+    child_did TEXT NOT NULL,
+    child_collection TEXT NOT NULL,
+    child_rkey TEXT NOT NULL,
+
+    -- Parent reference (decomposed)
+    parent_did TEXT NOT NULL,
+    parent_collection TEXT NOT NULL,
+    parent_rkey TEXT NOT NULL,
+
+    edge_type TEXT NOT NULL,  -- 'prev' | 'merge' (future)
+
+    PRIMARY KEY (child_did, child_collection, child_rkey, parent_did, parent_collection, parent_rkey),
+    FOREIGN KEY (child_did, child_collection, child_rkey) REFERENCES edit_nodes(did, collection, rkey),
+    FOREIGN KEY (parent_did, parent_collection, parent_rkey) REFERENCES edit_nodes(did, collection, rkey)
+);
+
+CREATE INDEX idx_edit_edges_parent ON edit_edges(parent_did, parent_collection, parent_rkey);
+
+-- Fast path: track current head per resource
+CREATE TABLE edit_heads (
+    -- Resource identity (decomposed)
+    resource_did TEXT NOT NULL,
+    resource_collection TEXT NOT NULL,
+    resource_rkey TEXT NOT NULL,
+
+    -- Latest root reference (decomposed)
+    root_did TEXT,
+    root_collection TEXT,
+    root_rkey TEXT,
+
+    -- Current head reference (decomposed)
+    head_did TEXT,
+    head_collection TEXT,
+    head_rkey TEXT,
+
+    updated_at TEXT NOT NULL,
+
+    PRIMARY KEY (resource_did, resource_collection, resource_rkey)
+);

crates/weaver-index/src/sqlite/migrations/002_collaboration.sql

···
+-- Valid collaborators (invite + accept pairs)
+CREATE TABLE collaborators (
+    -- Resource reference (decomposed)
+    resource_did TEXT NOT NULL,
+    resource_collection TEXT NOT NULL,
+    resource_rkey TEXT NOT NULL,
+
+    collaborator_did TEXT NOT NULL,
+
+    -- Invite record reference (decomposed)
+    invite_did TEXT NOT NULL,
+    invite_rkey TEXT NOT NULL,
+
+    -- Accept record reference (decomposed)
+    accept_did TEXT NOT NULL,
+    accept_rkey TEXT NOT NULL,
+
+    scope TEXT NOT NULL,  -- 'direct' | 'inherited'
+    granted_at TEXT NOT NULL,
+    indexed_at TEXT NOT NULL,
+
+    PRIMARY KEY (resource_did, resource_collection, resource_rkey, collaborator_did)
+);
+
+CREATE INDEX idx_collaborators_did ON collaborators(collaborator_did);
+
+-- Active sessions (TTL-based, cleaned up on expiry)
+CREATE TABLE sessions (
+    -- Session record identity (decomposed)
+    did TEXT NOT NULL,
+    rkey TEXT NOT NULL,
+
+    -- Resource reference (decomposed)
+    resource_did TEXT NOT NULL,
+    resource_collection TEXT NOT NULL,
+    resource_rkey TEXT NOT NULL,
+
+    participant_did TEXT NOT NULL,
+    node_id TEXT NOT NULL,
+    relay_url TEXT,  -- NULL if no relay
+    created_at TEXT NOT NULL,
+    expires_at TEXT,  -- NULL = no expiry
+    indexed_at TEXT NOT NULL,
+
+    PRIMARY KEY (did, rkey)
+);
+
+CREATE INDEX idx_sessions_resource ON sessions(resource_did, resource_collection, resource_rkey);
+CREATE INDEX idx_sessions_expires ON sessions(expires_at);
+
+-- Pending invites (no accept yet)
+CREATE TABLE pending_invites (
+    -- Invite record identity (decomposed)
+    did TEXT NOT NULL,  -- inviter DID
+    rkey TEXT NOT NULL,
+
+    -- Resource reference (decomposed)
+    resource_did TEXT NOT NULL,
+    resource_collection TEXT NOT NULL,
+    resource_rkey TEXT NOT NULL,
+
+    inviter_did TEXT NOT NULL,  -- same as did
+    invitee_did TEXT NOT NULL,
+    message TEXT,  -- NULL if no message
+    expires_at TEXT,  -- NULL = no expiry
+    created_at TEXT NOT NULL,
+    indexed_at TEXT NOT NULL,
+
+    PRIMARY KEY (did, rkey)
+);
+
+CREATE INDEX idx_pending_invites_resource ON pending_invites(resource_did, resource_collection, resource_rkey);
+CREATE INDEX idx_pending_invites_invitee ON pending_invites(invitee_did);

crates/weaver-index/src/sqlite/migrations/003_permissions.sql

···
+-- Permissions cache
+-- Local cache of permissions for collab-related hot paths.
+-- ClickHouse is authoritative; this is populated on-demand for active resources.
+CREATE TABLE permissions (
+    -- Resource reference (decomposed)
+    resource_did TEXT NOT NULL,
+    resource_collection TEXT NOT NULL,
+    resource_rkey TEXT NOT NULL,
+
+    did TEXT NOT NULL,  -- user who has permission
+
+    scope TEXT NOT NULL,  -- 'owner' | 'direct' | 'inherited'
+
+    -- Source reference (decomposed) - resource itself for owner, invite for others
+    source_did TEXT NOT NULL,
+    source_collection TEXT NOT NULL,
+    source_rkey TEXT NOT NULL,
+
+    granted_at TEXT NOT NULL,
+
+    PRIMARY KEY (resource_did, resource_collection, resource_rkey, did)
+);
+
+CREATE INDEX idx_permissions_did ON permissions(did);

crates/weaver-index/src/tap.rs

···
+mod consumer;
+mod types;
+
+pub use consumer::{TapConfig, TapConsumer};
+pub use types::*;

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

···
+use std::time::Duration;
+
+use futures_util::{SinkExt, StreamExt};
+use tokio::sync::mpsc;
+use tokio_tungstenite::{connect_async, tungstenite::Message};
+use tracing::{debug, error, info, trace, warn};
+use url::Url;
+
+use crate::error::IndexError;
+
+use super::{TapAck, TapEvent};
+
+/// Messages sent to the writer task
+enum WriteCommand {
+    #[allow(dead_code)]
+    Ack(u64),
+    Pong(bytes::Bytes),
+}
+
+/// Configuration for tap consumer
+#[derive(Debug, Clone)]
+pub struct TapConfig {
+    /// WebSocket URL for tap (e.g., ws://localhost:2480/channel)
+    pub url: Url,
+    /// Whether to send acks (disable for fire-and-forget mode)
+    pub send_acks: bool,
+    /// Reconnect delay on connection failure
+    pub reconnect_delay: Duration,
+}
+
+impl TapConfig {
+    pub fn new(url: Url) -> Self {
+        Self {
+            url,
+            send_acks: true,
+            reconnect_delay: Duration::from_secs(5),
+        }
+    }
+
+    pub fn with_acks(mut self, send_acks: bool) -> Self {
+        self.send_acks = send_acks;
+        self
+    }
+}
+
+/// Consumer that connects to tap's websocket and yields events
+pub struct TapConsumer {
+    config: TapConfig,
+}
+
+impl TapConsumer {
+    pub fn new(config: TapConfig) -> Self {
+        Self { config }
+    }
+
+    /// Connect to tap and return channels for events and acks
+    ///
+    /// Returns a receiver for events and a sender for acks.
+    /// The consumer handles reconnection internally.
+    pub async fn connect(
+        &self,
+    ) -> Result<(mpsc::Receiver<TapEvent>, mpsc::Sender<u64>), IndexError> {
+        let (event_tx, event_rx) = mpsc::channel::<TapEvent>(10000);
+        let (ack_tx, ack_rx) = mpsc::channel::<u64>(10000);
+
+        let config = self.config.clone();
+        tokio::spawn(async move {
+            run_connection_loop(config, event_tx, ack_rx).await;
+        });
+
+        Ok((event_rx, ack_tx))
+    }
+}
+
+async fn run_connection_loop(
+    config: TapConfig,
+    event_tx: mpsc::Sender<TapEvent>,
+    ack_rx: mpsc::Receiver<u64>,
+) {
+    loop {
+        info!(url = %config.url, "connecting to tap");
+
+        match connect_async(config.url.as_str()).await {
+            Ok((ws_stream, _response)) => {
+                info!("connected to tap");
+
+                let (write, read) = ws_stream.split();
+
+                // Channel for reader -> writer communication (pongs, etc)
+                let (write_tx, write_rx) = mpsc::channel::<WriteCommand>(10000);
+
+                // Spawn writer task
+                let send_acks = config.send_acks;
+                let writer_handle = tokio::spawn(run_writer(write, write_rx, ack_rx, send_acks));
+
+                // Run reader in current task
+                let reader_result = run_reader(read, event_tx.clone(), write_tx, send_acks).await;
+
+                // Reader finished - abort writer and wait for it
+                writer_handle.abort();
+                let _ = writer_handle.await;
+
+                // Get back the ack_rx from... wait, we moved it. Need to restructure.
+                // For now, if reader dies we'll reconnect with a fresh ack channel state
+
+                match reader_result {
+                    ReaderResult::Closed => {
+                        info!("tap connection closed");
+                    }
+                    ReaderResult::Error(e) => {
+                        warn!(error = %e, "tap reader error");
+                    }
+                    ReaderResult::ChannelClosed => {
+                        error!("event channel closed, stopping tap consumer");
+                        return;
+                    }
+                }
+
+                // We lost the ack_rx to the writer task, need to break out
+                // and let caller reconnect if needed
+                break;
+            }
+            Err(e) => {
+                error!(error = ?e, "failed to connect to tap");
+            }
+        }
+
+        // Reconnect after delay
+        info!(delay = ?config.reconnect_delay, "reconnecting to tap");
+        tokio::time::sleep(config.reconnect_delay).await;
+    }
+}
+
+enum ReaderResult {
+    Closed,
+    Error(String),
+    ChannelClosed,
+}
+
+async fn run_reader<S>(
+    mut read: S,
+    event_tx: mpsc::Sender<TapEvent>,
+    write_tx: mpsc::Sender<WriteCommand>,
+    send_acks: bool,
+) -> ReaderResult
+where
+    S: StreamExt<Item = Result<Message, tokio_tungstenite::tungstenite::Error>> + Unpin,
+{
+    while let Some(msg) = read.next().await {
+        match msg {
+            Ok(Message::Text(text)) => match serde_json::from_str::<TapEvent>(&text) {
+                Ok(event) => {
+                    let event_id = event.id();
+                    if event_tx.send(event).await.is_err() {
+                        return ReaderResult::ChannelClosed;
+                    }
+
+                    if !send_acks {
+                        debug!(id = event_id, "event received (fire-and-forget)");
+                    }
+                }
+                Err(e) => {
+                    warn!(error = ?e, text = %text, "failed to parse tap event");
+                }
+            },
+            Ok(Message::Ping(data)) => {
+                if write_tx.send(WriteCommand::Pong(data)).await.is_err() {
+                    return ReaderResult::Error("writer channel closed".into());
+                }
+            }
+            Ok(Message::Close(_)) => {
+                return ReaderResult::Closed;
+            }
+            Ok(_) => {
+                // Ignore binary, pong, etc.
+            }
+            Err(e) => {
+                return ReaderResult::Error(e.to_string());
+            }
+        }
+    }
+    ReaderResult::Closed
+}
+
+async fn run_writer<S>(
+    mut write: S,
+    mut write_rx: mpsc::Receiver<WriteCommand>,
+    mut ack_rx: mpsc::Receiver<u64>,
+    send_acks: bool,
+) where
+    S: SinkExt<Message> + Unpin,
+    S::Error: std::fmt::Display,
+{
+    loop {
+        tokio::select! {
+            biased;
+
+            // Handle pongs and other write commands from reader
+            cmd = write_rx.recv() => {
+                match cmd {
+                    Some(WriteCommand::Pong(data)) => {
+                        if let Err(e) = write.send(Message::Pong(data)).await {
+                            warn!(error = %e, "failed to send pong");
+                            return;
+                        }
+                    }
+                    Some(WriteCommand::Ack(id)) => {
+                        if send_acks {
+                            if let Err(e) = send_ack(&mut write, id).await {
+                                warn!(error = %e, id, "failed to send ack");
+                                return;
+                            }
+                        }
+                    }
+                    None => {
+                        // Reader closed the channel, we're done
+                        return;
+                    }
+                }
+            }
+
+            // Handle acks from the indexer
+            id = ack_rx.recv(), if send_acks => {
+                match id {
+                    Some(id) => {
+                        if let Err(e) = send_ack(&mut write, id).await {
+                            warn!(error = %e, id, "failed to send ack");
+                            return;
+                        }
+                    }
+                    None => {
+                        // Ack channel closed, indexer is done
+                        return;
+                    }
+                }
+            }
+        }
+    }
+}
+
+async fn send_ack<S>(write: &mut S, id: u64) -> Result<(), String>
+where
+    S: SinkExt<Message> + Unpin,
+    S::Error: std::fmt::Display,
+{
+    let ack = TapAck::new(id);
+    let json = serde_json::to_string(&ack).map_err(|e| e.to_string())?;
+    write
+        .send(Message::Text(json.into()))
+        .await
+        .map_err(|e| e.to_string())?;
+    trace!(id, "sent ack");
+    Ok(())
+}

crates/weaver-index/src/tap/types.rs

···
+use serde::{Deserialize, Serialize};
+use smol_str::SmolStr;
+
+/// Event received from tap's websocket channel
+#[derive(Debug, Clone, Deserialize)]
+#[serde(tag = "type", rename_all = "lowercase")]
+pub enum TapEvent {
+    Record(TapRecordEnvelope),
+    Identity(TapIdentityEnvelope),
+}
+
+impl TapEvent {
+    pub fn id(&self) -> u64 {
+        match self {
+            TapEvent::Record(r) => r.id,
+            TapEvent::Identity(i) => i.id,
+        }
+    }
+}
+
+/// Envelope for record events
+#[derive(Debug, Clone, Deserialize)]
+pub struct TapRecordEnvelope {
+    pub id: u64,
+    pub record: TapRecordEvent,
+}
+
+/// Record event from tap
+#[derive(Debug, Clone, Deserialize)]
+pub struct TapRecordEvent {
+    /// Whether this is a live event (true) or backfill (false)
+    pub live: bool,
+    /// Repository revision
+    pub rev: SmolStr,
+    /// DID of the account
+    pub did: SmolStr,
+    /// Collection NSID (e.g., "app.bsky.feed.post")
+    pub collection: SmolStr,
+    /// Record key
+    pub rkey: SmolStr,
+    /// Operation: create, update, delete
+    pub action: RecordAction,
+    /// Content identifier
+    pub cid: SmolStr,
+    /// The actual record data (only present for create/update)
+    #[serde(default)]
+    pub record: Option<serde_json::Value>,
+}
+
+/// Record operation type
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RecordAction {
+    Create,
+    Update,
+    Delete,
+}
+
+impl RecordAction {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            RecordAction::Create => "create",
+            RecordAction::Update => "update",
+            RecordAction::Delete => "delete",
+        }
+    }
+}
+
+/// Envelope for identity events
+#[derive(Debug, Clone, Deserialize)]
+pub struct TapIdentityEnvelope {
+    pub id: u64,
+    pub identity: TapIdentityEvent,
+}
+
+/// Identity event from tap (handle or status changes)
+#[derive(Debug, Clone, Deserialize)]
+pub struct TapIdentityEvent {
+    pub did: SmolStr,
+    pub handle: SmolStr,
+    pub is_active: bool,
+    pub status: AccountStatus,
+}
+
+/// Account status
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum AccountStatus {
+    Active,
+    Deactivated,
+    Suspended,
+    Deleted,
+    Takendown,
+    #[serde(other)]
+    Unknown,
+}
+
+/// Ack message to send back to tap
+#[derive(Debug, Clone, Serialize)]
+pub struct TapAck {
+    #[serde(rename = "type")]
+    pub msg_type: &'static str,
+    pub id: u64,
+}
+
+impl TapAck {
+    pub fn new(id: u64) -> Self {
+        Self {
+            msg_type: "ack",
+            id,
+        }
+    }
+}

docker-compose.yml

···
+services:
+  # Tap - AT Protocol sync utility
+  # Build from local indigo checkout, or use pre-built image
+  tap:
+    container_name: weaver-tap
+    build:
+      # Build from local indigo checkout on sync-tool branch
+      # git clone https://github.com/bluesky-social/indigo.git && git checkout sync-tool
+      context: ../../Git_Repos/indigo
+      dockerfile: cmd/tap/Dockerfile
+    ports:
+      - "2480:2480"
+    volumes:
+      - tap_data:/data/tap
+    environment:
+      TAP_DATABASE_URL: sqlite:///data/tap/tap.db
+      TAP_BIND: ":2480"
+      TAP_DISABLE_ACKS: "false"
+      TAP_LOG_LEVEL: info
+      # Filter to weaver collections only
+      TAP_SIGNAL_COLLECTION: sh.weaver.edit.root
+      TAP_COLLECTION_FILTERS: "sh.weaver.*,app.bsky.actor.profile"
+    healthcheck:
+      test: ["CMD", "wget", "-q", "--spider", "http://localhost:2480/health"]
+      interval: 20s
+      timeout: 5s
+      retries: 3
+
+  # Weaver indexer - consumes from tap or direct firehose
+  indexer:
+    container_name: weaver-indexer
+    build:
+      context: .
+      dockerfile: crates/weaver-index/Dockerfile
+    command: ["run"]
+    ports:
+      - "3000:3000"
+    environment:
+      RUST_LOG: debug,weaver_index=debug
+      # ClickHouse connection (set these for your cloud/homelab instance)
+      CLICKHOUSE_URL: ${CLICKHOUSE_URL}
+      CLICKHOUSE_DATABASE: ${CLICKHOUSE_DATABASE:-weaver}
+      CLICKHOUSE_USER: ${CLICKHOUSE_USER}
+      CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD}
+      # Source mode: "firehose" or "tap"
+      INDEXER_SOURCE: tap
+      # Tap connection (when INDEXER_SOURCE=tap)
+      TAP_URL: ws://tap:2480/channel
+      TAP_SEND_ACKS: "true"
+      # Firehose connection (when INDEXER_SOURCE=firehose)
+      FIREHOSE_RELAY_URL: wss://bsky.network
+      # Collection filters
+      INDEXER_COLLECTIONS: "sh.weaver.*,app.bsky.actor.profile"
+    depends_on:
+      tap:
+        condition: service_healthy
+    healthcheck:
+      test: ["CMD", "wget", "-q", "--spider", "http://localhost:2480/xrpc/_health"]
+      interval: 20s
+      timeout: 5s
+      retries: 3
+
+volumes:
+  tap_data: