From 40eee0db4b7c6276c4882d65687fbe0d599c7cf5 Mon Sep 17 00:00:00 2001 From: Orual Date: Thu, 11 Dec 2025 10:06:01 -0500 Subject: [PATCH] initial consumer skeleton --- Cargo.lock | 21 + .../src/com_atproto/sync/subscribe_repos.rs | 370 +++++-------- crates/weaver-index/Cargo.toml | 2 + .../migrations/clickhouse/001_raw_records.sql | 19 +- .../clickhouse/006_account_rev_state.sql | 24 + .../clickhouse/007_account_rev_state_mv.sql | 12 + .../weaver-index/src/bin/storage_benchmark.rs | 9 +- crates/weaver-index/src/bin/weaver_indexer.rs | 81 ++- crates/weaver-index/src/clickhouse.rs | 3 +- crates/weaver-index/src/clickhouse/client.rs | 6 +- .../weaver-index/src/clickhouse/migrations.rs | 58 +- crates/weaver-index/src/clickhouse/schema.rs | 70 ++- crates/weaver-index/src/config.rs | 120 ++++ crates/weaver-index/src/firehose.rs | 4 +- crates/weaver-index/src/firehose/records.rs | 16 +- crates/weaver-index/src/indexer.rs | 512 ++++++++++++++++++ crates/weaver-index/src/lib.rs | 2 + 17 files changed, 1029 insertions(+), 300 deletions(-) create mode 100644 crates/weaver-index/migrations/clickhouse/006_account_rev_state.sql create mode 100644 crates/weaver-index/migrations/clickhouse/007_account_rev_state_mv.sql create mode 100644 crates/weaver-index/src/indexer.rs diff --git a/Cargo.lock b/Cargo.lock index 8974e3a..dc179f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5023,6 +5023,25 @@ version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edcd27d72f2f071c64249075f42e205ff93c9a4c5f6c6da53e79ed9f9832c285" +[[package]] +name = "include_dir" +version = "0.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "923d117408f1e49d914f1a379a309cffe4f18c05cf4e3d12e613a15fc81bd0dd" +dependencies = [ + "include_dir_macros", +] + +[[package]] +name = "include_dir_macros" +version = "0.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cab85a7ed0bd5f0e76d93846e0147172bed2e2d3f859bcc33a8d9699cad1a75" +dependencies = [ + "proc-macro2", + "quote", +] + [[package]] name = "indexmap" version = "1.9.3" @@ -11464,8 +11483,10 @@ dependencies = [ "cid", "clap", "clickhouse", + "dashmap 6.1.0", "dotenvy", "humansize", + "include_dir", "jacquard", "jacquard-common", "jacquard-repo", diff --git a/crates/weaver-api/src/com_atproto/sync/subscribe_repos.rs b/crates/weaver-api/src/com_atproto/sync/subscribe_repos.rs index 4ae22f2..7a3ee31 100644 --- a/crates/weaver-api/src/com_atproto/sync/subscribe_repos.rs +++ b/crates/weaver-api/src/com_atproto/sync/subscribe_repos.rs @@ -8,13 +8,7 @@ /// Represents a change to an account's status on a host (eg, PDS or Relay). The semantics of this event are that the status is at the host which emitted the event, not necessarily that at the currently active PDS. Eg, a Relay takedown would emit a takedown with active=false, even if the PDS is still active. #[jacquard_derive::lexicon] #[derive( - serde::Serialize, - serde::Deserialize, - Debug, - Clone, - PartialEq, - Eq, - jacquard_derive::IntoStatic + serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic, )] #[serde(rename_all = "camelCase")] pub struct Account<'a> { @@ -32,7 +26,7 @@ pub struct Account<'a> { pub mod account_state { - pub use crate::builder_types::{Set, Unset, IsSet, IsUnset}; + pub use crate::builder_types::{IsSet, IsUnset, Set, Unset}; #[allow(unused)] use ::core::marker::PhantomData; mod sealed { @@ -179,10 +173,7 @@ where S::Seq: account_state::IsUnset, { /// Set the `seq` field (required) - pub fn seq( - mut self, - value: impl Into, - ) -> AccountBuilder<'a, account_state::SetSeq> { + pub fn seq(mut self, value: impl Into) -> AccountBuilder<'a, account_state::SetSeq> { self.__unsafe_private_named.2 = ::core::option::Option::Some(value.into()); AccountBuilder { _phantom_state: ::core::marker::PhantomData, @@ -194,10 +185,7 @@ where impl<'a, S: account_state::State> AccountBuilder<'a, S> { /// Set the `status` field (optional) - pub fn status( - mut self, - value: impl Into>>, - ) -> Self { + pub fn status(mut self, value: impl Into>>) -> Self { self.__unsafe_private_named.3 = value.into(); self } @@ -265,9 +253,8 @@ where } } -fn lexicon_doc_com_atproto_sync_subscribeRepos() -> ::jacquard_lexicon::lexicon::LexiconDoc< - 'static, -> { +fn lexicon_doc_com_atproto_sync_subscribeRepos() -> ::jacquard_lexicon::lexicon::LexiconDoc<'static> +{ ::jacquard_lexicon::lexicon::LexiconDoc { lexicon: ::jacquard_lexicon::lexicon::Lexicon::Lexicon1, id: ::jacquard_common::CowStr::new_static("com.atproto.sync.subscribeRepos"), @@ -647,48 +634,54 @@ fn lexicon_doc_com_atproto_sync_subscribeRepos() -> ::jacquard_lexicon::lexicon: ); map.insert( ::jacquard_common::smol_str::SmolStr::new_static("info"), - ::jacquard_lexicon::lexicon::LexUserType::Object(::jacquard_lexicon::lexicon::LexObject { - description: None, - required: Some( - vec![::jacquard_common::smol_str::SmolStr::new_static("name")], - ), - nullable: None, - properties: { - #[allow(unused_mut)] - let mut map = ::std::collections::BTreeMap::new(); - map.insert( - ::jacquard_common::smol_str::SmolStr::new_static("message"), - ::jacquard_lexicon::lexicon::LexObjectProperty::String(::jacquard_lexicon::lexicon::LexString { - description: None, - format: None, - default: None, - min_length: None, - max_length: None, - min_graphemes: None, - max_graphemes: None, - r#enum: None, - r#const: None, - known_values: None, - }), - ); - map.insert( - ::jacquard_common::smol_str::SmolStr::new_static("name"), - ::jacquard_lexicon::lexicon::LexObjectProperty::String(::jacquard_lexicon::lexicon::LexString { - description: None, - format: None, - default: None, - min_length: None, - max_length: None, - min_graphemes: None, - max_graphemes: None, - r#enum: None, - r#const: None, - known_values: None, - }), - ); - map + ::jacquard_lexicon::lexicon::LexUserType::Object( + ::jacquard_lexicon::lexicon::LexObject { + description: None, + required: Some(vec![::jacquard_common::smol_str::SmolStr::new_static( + "name", + )]), + nullable: None, + properties: { + #[allow(unused_mut)] + let mut map = ::std::collections::BTreeMap::new(); + map.insert( + ::jacquard_common::smol_str::SmolStr::new_static("message"), + ::jacquard_lexicon::lexicon::LexObjectProperty::String( + ::jacquard_lexicon::lexicon::LexString { + description: None, + format: None, + default: None, + min_length: None, + max_length: None, + min_graphemes: None, + max_graphemes: None, + r#enum: None, + r#const: None, + known_values: None, + }, + ), + ); + map.insert( + ::jacquard_common::smol_str::SmolStr::new_static("name"), + ::jacquard_lexicon::lexicon::LexObjectProperty::String( + ::jacquard_lexicon::lexicon::LexString { + description: None, + format: None, + default: None, + min_length: None, + max_length: None, + min_graphemes: None, + max_graphemes: None, + r#enum: None, + r#const: None, + known_values: None, + }, + ), + ); + map + }, }, - }), + ), ); map.insert( ::jacquard_common::smol_str::SmolStr::new_static("main"), @@ -723,68 +716,70 @@ fn lexicon_doc_com_atproto_sync_subscribeRepos() -> ::jacquard_lexicon::lexicon: ); map.insert( ::jacquard_common::smol_str::SmolStr::new_static("repoOp"), - ::jacquard_lexicon::lexicon::LexUserType::Object(::jacquard_lexicon::lexicon::LexObject { - description: Some( - ::jacquard_common::CowStr::new_static( + ::jacquard_lexicon::lexicon::LexUserType::Object( + ::jacquard_lexicon::lexicon::LexObject { + description: Some(::jacquard_common::CowStr::new_static( "A repo operation, ie a mutation of a single record.", - ), - ), - required: Some( - vec![ + )), + required: Some(vec![ ::jacquard_common::smol_str::SmolStr::new_static("action"), ::jacquard_common::smol_str::SmolStr::new_static("path"), - ::jacquard_common::smol_str::SmolStr::new_static("cid") - ], - ), - nullable: None, - properties: { - #[allow(unused_mut)] - let mut map = ::std::collections::BTreeMap::new(); - map.insert( - ::jacquard_common::smol_str::SmolStr::new_static("action"), - ::jacquard_lexicon::lexicon::LexObjectProperty::String(::jacquard_lexicon::lexicon::LexString { - description: None, - format: None, - default: None, - min_length: None, - max_length: None, - min_graphemes: None, - max_graphemes: None, - r#enum: None, - r#const: None, - known_values: None, - }), - ); - map.insert( ::jacquard_common::smol_str::SmolStr::new_static("cid"), - ::jacquard_lexicon::lexicon::LexObjectProperty::CidLink(::jacquard_lexicon::lexicon::LexCidLink { - description: None, - }), - ); - map.insert( - ::jacquard_common::smol_str::SmolStr::new_static("path"), - ::jacquard_lexicon::lexicon::LexObjectProperty::String(::jacquard_lexicon::lexicon::LexString { - description: None, - format: None, - default: None, - min_length: None, - max_length: None, - min_graphemes: None, - max_graphemes: None, - r#enum: None, - r#const: None, - known_values: None, - }), - ); - map.insert( - ::jacquard_common::smol_str::SmolStr::new_static("prev"), - ::jacquard_lexicon::lexicon::LexObjectProperty::CidLink(::jacquard_lexicon::lexicon::LexCidLink { - description: None, - }), - ); - map + ]), + nullable: None, + properties: { + #[allow(unused_mut)] + let mut map = ::std::collections::BTreeMap::new(); + map.insert( + ::jacquard_common::smol_str::SmolStr::new_static("action"), + ::jacquard_lexicon::lexicon::LexObjectProperty::String( + ::jacquard_lexicon::lexicon::LexString { + description: None, + format: None, + default: None, + min_length: None, + max_length: None, + min_graphemes: None, + max_graphemes: None, + r#enum: None, + r#const: None, + known_values: None, + }, + ), + ); + map.insert( + ::jacquard_common::smol_str::SmolStr::new_static("cid"), + ::jacquard_lexicon::lexicon::LexObjectProperty::CidLink( + ::jacquard_lexicon::lexicon::LexCidLink { description: None }, + ), + ); + map.insert( + ::jacquard_common::smol_str::SmolStr::new_static("path"), + ::jacquard_lexicon::lexicon::LexObjectProperty::String( + ::jacquard_lexicon::lexicon::LexString { + description: None, + format: None, + default: None, + min_length: None, + max_length: None, + min_graphemes: None, + max_graphemes: None, + r#enum: None, + r#const: None, + known_values: None, + }, + ), + ); + map.insert( + ::jacquard_common::smol_str::SmolStr::new_static("prev"), + ::jacquard_lexicon::lexicon::LexObjectProperty::CidLink( + ::jacquard_lexicon::lexicon::LexCidLink { description: None }, + ), + ); + map + }, }, - }), + ), ); map.insert( ::jacquard_common::smol_str::SmolStr::new_static("sync"), @@ -916,13 +911,7 @@ impl<'a> ::jacquard_lexicon::schema::LexiconSchema for Account<'a> { /// Represents an update of repository state. Note that empty commits are allowed, which include no repo data changes, but an update to rev and signature. #[jacquard_derive::lexicon] #[derive( - serde::Serialize, - serde::Deserialize, - Debug, - Clone, - PartialEq, - Eq, - jacquard_derive::IntoStatic + serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic, )] #[serde(rename_all = "camelCase")] pub struct Commit<'a> { @@ -959,7 +948,7 @@ pub struct Commit<'a> { pub mod commit_state { - pub use crate::builder_types::{Set, Unset, IsSet, IsUnset}; + pub use crate::builder_types::{IsSet, IsUnset, Set, Unset}; #[allow(unused)] use ::core::marker::PhantomData; mod sealed { @@ -1176,9 +1165,7 @@ pub struct CommitBuilder<'a, S: commit_state::State> { ::core::option::Option>>, ::core::option::Option, ::core::option::Option>, - ::core::option::Option< - Vec>, - >, + ::core::option::Option>>, ::core::option::Option>, ::core::option::Option, ::core::option::Option>, @@ -1204,18 +1191,7 @@ impl<'a> CommitBuilder<'a, commit_state::Empty> { CommitBuilder { _phantom_state: ::core::marker::PhantomData, __unsafe_private_named: ( - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, + None, None, None, None, None, None, None, None, None, None, None, None, ), _phantom: ::core::marker::PhantomData, } @@ -1380,10 +1356,7 @@ where S::Seq: commit_state::IsUnset, { /// Set the `seq` field (required) - pub fn seq( - mut self, - value: impl Into, - ) -> CommitBuilder<'a, commit_state::SetSeq> { + pub fn seq(mut self, value: impl Into) -> CommitBuilder<'a, commit_state::SetSeq> { self.__unsafe_private_named.8 = ::core::option::Option::Some(value.into()); CommitBuilder { _phantom_state: ::core::marker::PhantomData, @@ -1395,18 +1368,12 @@ where impl<'a, S: commit_state::State> CommitBuilder<'a, S> { /// Set the `since` field (optional) - pub fn since( - mut self, - value: impl Into>, - ) -> Self { + pub fn since(mut self, value: impl Into>) -> Self { self.__unsafe_private_named.9 = value.into(); self } /// Set the `since` field to an Option value (optional) - pub fn maybe_since( - mut self, - value: Option, - ) -> Self { + pub fn maybe_since(mut self, value: Option) -> Self { self.__unsafe_private_named.9 = value; self } @@ -1526,9 +1493,7 @@ impl<'a> ::jacquard_lexicon::schema::LexiconSchema for Commit<'a> { #[allow(unused_comparisons)] if value.len() > 200usize { return Err(::jacquard_lexicon::validation::ConstraintError::MaxLength { - path: ::jacquard_lexicon::validation::ValidationPath::from_field( - "ops", - ), + path: ::jacquard_lexicon::validation::ValidationPath::from_field("ops"), max: 200usize, actual: value.len(), }); @@ -1541,13 +1506,7 @@ impl<'a> ::jacquard_lexicon::schema::LexiconSchema for Commit<'a> { /// Represents a change to an account's identity. Could be an updated handle, signing key, or pds hosting endpoint. Serves as a prod to all downstream services to refresh their identity cache. #[jacquard_derive::lexicon] #[derive( - serde::Serialize, - serde::Deserialize, - Debug, - Clone, - PartialEq, - Eq, - jacquard_derive::IntoStatic + serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic, )] #[serde(rename_all = "camelCase")] pub struct Identity<'a> { @@ -1563,7 +1522,7 @@ pub struct Identity<'a> { pub mod identity_state { - pub use crate::builder_types::{Set, Unset, IsSet, IsUnset}; + pub use crate::builder_types::{IsSet, IsUnset, Set, Unset}; #[allow(unused)] use ::core::marker::PhantomData; mod sealed { @@ -1693,10 +1652,7 @@ where S::Seq: identity_state::IsUnset, { /// Set the `seq` field (required) - pub fn seq( - mut self, - value: impl Into, - ) -> IdentityBuilder<'a, identity_state::SetSeq> { + pub fn seq(mut self, value: impl Into) -> IdentityBuilder<'a, identity_state::SetSeq> { self.__unsafe_private_named.2 = ::core::option::Option::Some(value.into()); IdentityBuilder { _phantom_state: ::core::marker::PhantomData, @@ -1786,7 +1742,7 @@ impl<'a> ::jacquard_lexicon::schema::LexiconSchema for Identity<'a> { PartialEq, Eq, jacquard_derive::IntoStatic, - Default + Default, )] #[serde(rename_all = "camelCase")] pub struct Info<'a> { @@ -1815,13 +1771,7 @@ impl<'a> ::jacquard_lexicon::schema::LexiconSchema for Info<'a> { } #[derive( - serde::Serialize, - serde::Deserialize, - Debug, - Clone, - PartialEq, - Eq, - jacquard_derive::IntoStatic + serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic, )] #[serde(rename_all = "camelCase")] pub struct SubscribeRepos { @@ -1831,7 +1781,7 @@ pub struct SubscribeRepos { pub mod subscribe_repos_state { - pub use crate::builder_types::{Set, Unset, IsSet, IsUnset}; + pub use crate::builder_types::{IsSet, IsUnset, Set, Unset}; #[allow(unused)] use ::core::marker::PhantomData; mod sealed { @@ -1898,13 +1848,7 @@ where #[jacquard_derive::open_union] #[derive( - serde::Serialize, - serde::Deserialize, - Debug, - Clone, - PartialEq, - Eq, - jacquard_derive::IntoStatic + serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic, )] #[serde(tag = "$type")] #[serde(bound(deserialize = "'de: 'a"))] @@ -1926,9 +1870,7 @@ impl<'a> SubscribeReposMessage<'a> { pub fn decode_framed<'de: 'a>( bytes: &'de [u8], ) -> Result, jacquard_common::error::DecodeError> { - let (header, body) = jacquard_common::xrpc::subscription::parse_event_header( - bytes, - )?; + let (header, body) = jacquard_common::xrpc::subscription::parse_event_header(bytes)?; match header.t.as_str() { "#commit" => { let variant = serde_ipld_dagcbor::from_slice(body)?; @@ -1950,11 +1892,9 @@ impl<'a> SubscribeReposMessage<'a> { let variant = serde_ipld_dagcbor::from_slice(body)?; Ok(Self::Info(Box::new(variant))) } - unknown => { - Err( - jacquard_common::error::DecodeError::UnknownEventType(unknown.into()), - ) - } + unknown => Err(jacquard_common::error::DecodeError::UnknownEventType( + unknown.into(), + )), } } } @@ -1969,7 +1909,7 @@ impl<'a> SubscribeReposMessage<'a> { Eq, thiserror::Error, miette::Diagnostic, - jacquard_derive::IntoStatic + jacquard_derive::IntoStatic, )] #[serde(tag = "error", content = "message")] #[serde(bound(deserialize = "'de: 'a"))] @@ -2008,7 +1948,8 @@ impl std::fmt::Display for SubscribeReposError<'_> { pub struct SubscribeReposStream; impl jacquard_common::xrpc::SubscriptionResp for SubscribeReposStream { const NSID: &'static str = "com.atproto.sync.subscribeRepos"; - const ENCODING: jacquard_common::xrpc::MessageEncoding = jacquard_common::xrpc::MessageEncoding::DagCbor; + const ENCODING: jacquard_common::xrpc::MessageEncoding = + jacquard_common::xrpc::MessageEncoding::DagCbor; type Message<'de> = SubscribeReposMessage<'de>; type Error<'de> = SubscribeReposError<'de>; fn decode_message<'de>( @@ -2020,14 +1961,16 @@ impl jacquard_common::xrpc::SubscriptionResp for SubscribeReposStream { impl jacquard_common::xrpc::XrpcSubscription for SubscribeRepos { const NSID: &'static str = "com.atproto.sync.subscribeRepos"; - const ENCODING: jacquard_common::xrpc::MessageEncoding = jacquard_common::xrpc::MessageEncoding::DagCbor; + const ENCODING: jacquard_common::xrpc::MessageEncoding = + jacquard_common::xrpc::MessageEncoding::DagCbor; type Stream = SubscribeReposStream; } pub struct SubscribeReposEndpoint; impl jacquard_common::xrpc::SubscriptionEndpoint for SubscribeReposEndpoint { const PATH: &'static str = "/xrpc/com.atproto.sync.subscribeRepos"; - const ENCODING: jacquard_common::xrpc::MessageEncoding = jacquard_common::xrpc::MessageEncoding::DagCbor; + const ENCODING: jacquard_common::xrpc::MessageEncoding = + jacquard_common::xrpc::MessageEncoding::DagCbor; type Params<'de> = SubscribeRepos; type Stream = SubscribeReposStream; } @@ -2035,13 +1978,7 @@ impl jacquard_common::xrpc::SubscriptionEndpoint for SubscribeReposEndpoint { /// A repo operation, ie a mutation of a single record. #[jacquard_derive::lexicon] #[derive( - serde::Serialize, - serde::Deserialize, - Debug, - Clone, - PartialEq, - Eq, - jacquard_derive::IntoStatic + serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic, )] #[serde(rename_all = "camelCase")] pub struct RepoOp<'a> { @@ -2060,7 +1997,7 @@ pub struct RepoOp<'a> { pub mod repo_op_state { - pub use crate::builder_types::{Set, Unset, IsSet, IsUnset}; + pub use crate::builder_types::{IsSet, IsUnset, Set, Unset}; #[allow(unused)] use ::core::marker::PhantomData; mod sealed { @@ -2161,10 +2098,7 @@ impl<'a, S: repo_op_state::State> RepoOpBuilder<'a, S> { self } /// Set the `cid` field to an Option value (optional) - pub fn maybe_cid( - mut self, - value: Option>, - ) -> Self { + pub fn maybe_cid(mut self, value: Option>) -> Self { self.__unsafe_private_named.1 = value; self } @@ -2199,10 +2133,7 @@ impl<'a, S: repo_op_state::State> RepoOpBuilder<'a, S> { self } /// Set the `prev` field to an Option value (optional) - pub fn maybe_prev( - mut self, - value: Option>, - ) -> Self { + pub fn maybe_prev(mut self, value: Option>) -> Self { self.__unsafe_private_named.3 = value; self } @@ -2262,13 +2193,7 @@ impl<'a> ::jacquard_lexicon::schema::LexiconSchema for RepoOp<'a> { /// Updates the repo to a new state, without necessarily including that state on the firehose. Used to recover from broken commit streams, data loss incidents, or in situations where upstream host does not know recent state of the repository. #[jacquard_derive::lexicon] #[derive( - serde::Serialize, - serde::Deserialize, - Debug, - Clone, - PartialEq, - Eq, - jacquard_derive::IntoStatic + serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic, )] #[serde(rename_all = "camelCase")] pub struct Sync<'a> { @@ -2289,7 +2214,7 @@ pub struct Sync<'a> { pub mod sync_state { - pub use crate::builder_types::{Set, Unset, IsSet, IsUnset}; + pub use crate::builder_types::{IsSet, IsUnset, Set, Unset}; #[allow(unused)] use ::core::marker::PhantomData; mod sealed { @@ -2473,10 +2398,7 @@ where S::Seq: sync_state::IsUnset, { /// Set the `seq` field (required) - pub fn seq( - mut self, - value: impl Into, - ) -> SyncBuilder<'a, sync_state::SetSeq> { + pub fn seq(mut self, value: impl Into) -> SyncBuilder<'a, sync_state::SetSeq> { self.__unsafe_private_named.3 = ::core::option::Option::Some(value.into()); SyncBuilder { _phantom_state: ::core::marker::PhantomData, @@ -2559,4 +2481,4 @@ impl<'a> ::jacquard_lexicon::schema::LexiconSchema for Sync<'a> { ) -> ::std::result::Result<(), ::jacquard_lexicon::validation::ConstraintError> { Ok(()) } -} \ No newline at end of file +} diff --git a/crates/weaver-index/Cargo.toml b/crates/weaver-index/Cargo.toml index dde2787..527b069 100644 --- a/crates/weaver-index/Cargo.toml +++ b/crates/weaver-index/Cargo.toml @@ -64,3 +64,5 @@ cid = "0.11" # Utilities humansize = "2.0" base64 = "0.22" +dashmap = "6" +include_dir = "0.7.4" diff --git a/crates/weaver-index/migrations/clickhouse/001_raw_records.sql b/crates/weaver-index/migrations/clickhouse/001_raw_records.sql index 2e1d6ce..15bd8ef 100644 --- a/crates/weaver-index/migrations/clickhouse/001_raw_records.sql +++ b/crates/weaver-index/migrations/clickhouse/001_raw_records.sql @@ -13,6 +13,9 @@ CREATE TABLE IF NOT EXISTS raw_records ( -- Content identifier from the record cid String, + -- Repository revision (TID) - monotonically increasing per DID, used for dedup/ordering + rev String, + -- Full record as native JSON (schema-flexible, queryable with record.field.subfield) record JSON, @@ -28,8 +31,20 @@ CREATE TABLE IF NOT EXISTS raw_records ( -- When we indexed this record indexed_at DateTime64(3) DEFAULT now64(3), + -- Validation state: 'unchecked', 'valid', 'invalid_rev', 'invalid_gap', 'invalid_account' + -- Populated by async batch validation, not in hot path + validation_state LowCardinality(String) DEFAULT 'unchecked', + -- Materialized AT URI for convenience - uri String MATERIALIZED concat('at://', did, '/', collection, '/', rkey) + uri String MATERIALIZED concat('at://', did, '/', collection, '/', rkey), + + -- Projection for fast delete lookups by (did, cid) + -- Delete events include CID, so we can O(1) lookup the original record + -- to know what to decrement (e.g., which notebook's like count) + PROJECTION by_did_cid ( + SELECT * ORDER BY (did, cid) + ) ) ENGINE = ReplacingMergeTree(indexed_at) -ORDER BY (collection, did, rkey, indexed_at); +ORDER BY (collection, did, rkey, event_time) +SETTINGS deduplicate_merge_projection_mode = 'drop'; diff --git a/crates/weaver-index/migrations/clickhouse/006_account_rev_state.sql b/crates/weaver-index/migrations/clickhouse/006_account_rev_state.sql new file mode 100644 index 0000000..dcc7fd3 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/006_account_rev_state.sql @@ -0,0 +1,24 @@ +-- Per-account revision state tracking +-- Maintains latest rev/cid per DID for dedup and gap detection +-- +-- AggregatingMergeTree with incremental MV from raw_records +-- Query with argMaxMerge/maxMerge to finalize aggregates + +CREATE TABLE IF NOT EXISTS account_rev_state ( + -- Account DID + did String, + + -- Latest revision (TID) seen for this account + last_rev AggregateFunction(argMax, String, DateTime64(3)), + + -- CID of the latest revision + last_cid AggregateFunction(argMax, String, DateTime64(3)), + + -- Latest sequence number seen + last_seq AggregateFunction(max, UInt64), + + -- Latest event time seen + last_event_time AggregateFunction(max, DateTime64(3)) +) +ENGINE = AggregatingMergeTree() +ORDER BY did diff --git a/crates/weaver-index/migrations/clickhouse/007_account_rev_state_mv.sql b/crates/weaver-index/migrations/clickhouse/007_account_rev_state_mv.sql new file mode 100644 index 0000000..43d72e8 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/007_account_rev_state_mv.sql @@ -0,0 +1,12 @@ +-- Incremental MV: fires on each insert to raw_records, maintains aggregate state +-- Must be created after both account_rev_state (target) and raw_records (source) exist + +CREATE MATERIALIZED VIEW IF NOT EXISTS account_rev_state_mv TO account_rev_state AS +SELECT + did, + argMaxState(rev, event_time) as last_rev, + argMaxState(cid, event_time) as last_cid, + maxState(seq) as last_seq, + maxState(event_time) as last_event_time +FROM raw_records +GROUP BY did diff --git a/crates/weaver-index/src/bin/storage_benchmark.rs b/crates/weaver-index/src/bin/storage_benchmark.rs index fb11f15..6428eb5 100644 --- a/crates/weaver-index/src/bin/storage_benchmark.rs +++ b/crates/weaver-index/src/bin/storage_benchmark.rs @@ -305,18 +305,17 @@ async fn main() -> miette::Result<()> { } }; - let event_time = DateTime::from_timestamp_millis(record.event_time_ms).unwrap(); // Insert JSON record json_inserter .write(&RawRecordJson { did: record.did.clone(), collection: record.collection.clone(), rkey: record.rkey.clone(), - cid: record.cid.clone(), + cid: record.cid.to_string(), record: json_str, operation: record.operation.clone(), seq: record.seq as u64, - event_time: event_time.clone(), + event_time: record.event_time, }) .await .map_err(|e| weaver_index::error::ClickHouseError::Insert { @@ -330,11 +329,11 @@ async fn main() -> miette::Result<()> { did: record.did, collection: record.collection, rkey: record.rkey, - cid: record.cid, + cid: record.cid.to_string(), record: cbor_bytes.clone(), operation: record.operation, seq: record.seq as u64, - event_time, + event_time: record.event_time, }) .await .map_err(|e| weaver_index::error::ClickHouseError::Insert { diff --git a/crates/weaver-index/src/bin/weaver_indexer.rs b/crates/weaver-index/src/bin/weaver_indexer.rs index ffff545..7ab91d6 100644 --- a/crates/weaver-index/src/bin/weaver_indexer.rs +++ b/crates/weaver-index/src/bin/weaver_indexer.rs @@ -1,10 +1,14 @@ use clap::{Parser, Subcommand}; -use tracing::info; -use weaver_index::clickhouse::{Client, Migrator}; -use weaver_index::config::ClickHouseConfig; +use miette::IntoDiagnostic; +use tracing::{Level, info, warn}; +use tracing_subscriber::EnvFilter; +use weaver_index::clickhouse::{Client, Migrator, Tables}; +use weaver_index::config::{ClickHouseConfig, FirehoseConfig, IndexerConfig}; +use weaver_index::firehose::FirehoseConsumer; +use weaver_index::{Indexer, load_cursor}; #[derive(Parser)] -#[command(name = "weaver-indexer")] +#[command(name = "indexer")] #[command(about = "Weaver index service - firehose ingestion and query serving")] struct Args { #[command(subcommand)] @@ -18,6 +22,10 @@ enum Command { /// Show what would be run without executing #[arg(long)] dry_run: bool, + + /// Drop all tables before running migrations (for testing) + #[arg(long)] + reset: bool, }, /// Check database connectivity @@ -31,24 +39,31 @@ enum Command { 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::from_default_env() - .add_directive("weaver_index=info".parse().unwrap()) - .add_directive("weaver_indexer=info".parse().unwrap()), + tracing_subscriber::EnvFilter::builder() + .from_env_lossy() + .add_directive(console_level.into()) + .add_directive("hyper_util=info".parse().into_diagnostic()?), ) .init(); let args = Args::parse(); match args.command { - Command::Migrate { dry_run } => run_migrate(dry_run).await, + Command::Migrate { dry_run, reset } => run_migrate(dry_run, reset).await, Command::Health => run_health().await, Command::Run => run_indexer().await, } } -async fn run_migrate(dry_run: bool) -> miette::Result<()> { +async fn run_migrate(dry_run: bool, reset: bool) -> miette::Result<()> { let config = ClickHouseConfig::from_env()?; info!( "Connecting to ClickHouse at {} (database: {})", @@ -56,6 +71,25 @@ async fn run_migrate(dry_run: bool) -> miette::Result<()> { ); let client = Client::new(&config)?; + + if reset { + if dry_run { + info!("Would drop tables:"); + for table in Tables::ALL { + info!(" - {}", table); + } + } else { + info!("Dropping all tables..."); + for table in Tables::ALL { + let query = format!("DROP TABLE IF EXISTS {}", table); + match client.execute(&query).await { + Ok(_) => info!(" dropped {}", table), + Err(e) => warn!(" failed to drop {}: {}", table, e), + } + } + } + } + let migrator = Migrator::new(&client); if dry_run { @@ -93,6 +127,33 @@ async fn run_health() -> miette::Result<()> { } async fn run_indexer() -> miette::Result<()> { - info!("Indexer not yet implemented"); + let ch_config = ClickHouseConfig::from_env()?; + let mut firehose_config = FirehoseConfig::from_env()?; + let indexer_config = IndexerConfig::from_env(); + + info!( + "Connecting to ClickHouse at {} (database: {})", + ch_config.url, ch_config.database + ); + let client = Client::new(&ch_config)?; + + // Load cursor from ClickHouse if not overridden by env var + if firehose_config.cursor.is_none() { + if let Some(cursor) = load_cursor(&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 = Indexer::new(client, consumer, indexer_config).await?; + + info!("Starting indexer"); + indexer.run().await?; + Ok(()) } diff --git a/crates/weaver-index/src/clickhouse.rs b/crates/weaver-index/src/clickhouse.rs index 5451d6a..91f2e07 100644 --- a/crates/weaver-index/src/clickhouse.rs +++ b/crates/weaver-index/src/clickhouse.rs @@ -5,5 +5,6 @@ mod schema; pub use client::{Client, TableSize}; pub use migrations::{MigrationResult, Migrator}; pub use schema::{ - FirehoseCursor, RawAccountEvent, RawEventDlq, RawIdentityEvent, RawRecord, Tables, + AccountRevState, FirehoseCursor, RawAccountEvent, RawEventDlq, RawIdentityEvent, + RawRecordInsert, Tables, }; diff --git a/crates/weaver-index/src/clickhouse/client.rs b/crates/weaver-index/src/clickhouse/client.rs index 3c05bb7..f23d564 100644 --- a/crates/weaver-index/src/clickhouse/client.rs +++ b/crates/weaver-index/src/clickhouse/client.rs @@ -15,7 +15,11 @@ impl Client { .with_url(config.url.as_str()) .with_database(&config.database) .with_user(&config.user) - .with_password(&config.password); + .with_password(&config.password) + // 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"); Ok(Self { inner }) } diff --git a/crates/weaver-index/src/clickhouse/migrations.rs b/crates/weaver-index/src/clickhouse/migrations.rs index 194b647..9de7bbd 100644 --- a/crates/weaver-index/src/clickhouse/migrations.rs +++ b/crates/weaver-index/src/clickhouse/migrations.rs @@ -1,35 +1,11 @@ use crate::error::{ClickHouseError, IndexError}; +use include_dir::{Dir, include_dir}; use tracing::info; use super::Client; -/// Embedded migrations - compiled into the binary -const MIGRATIONS: &[(&str, &str)] = &[ - ( - "000_migrations.sql", - include_str!("../../migrations/clickhouse/000_migrations.sql"), - ), - ( - "001_raw_records.sql", - include_str!("../../migrations/clickhouse/001_raw_records.sql"), - ), - ( - "002_identity_events.sql", - include_str!("../../migrations/clickhouse/002_identity_events.sql"), - ), - ( - "003_account_events.sql", - include_str!("../../migrations/clickhouse/003_account_events.sql"), - ), - ( - "004_events_dlq.sql", - include_str!("../../migrations/clickhouse/004_events_dlq.sql"), - ), - ( - "005_firehose_cursor.sql", - include_str!("../../migrations/clickhouse/005_firehose_cursor.sql"), - ), -]; +/// Embedded migrations directory - compiled into the binary +static MIGRATIONS_DIR: Dir = include_dir!("$CARGO_MANIFEST_DIR/migrations/clickhouse"); /// Migration runner for ClickHouse pub struct Migrator<'a> { @@ -41,6 +17,21 @@ impl<'a> Migrator<'a> { Self { client } } + /// Get sorted list of migration files from embedded directory + fn migrations() -> Vec<(&'static str, &'static str)> { + let mut files: Vec<_> = MIGRATIONS_DIR + .files() + .filter(|f| f.path().extension().is_some_and(|ext| ext == "sql")) + .filter_map(|f| { + let name = f.path().file_name()?.to_str()?; + let contents = f.contents_utf8()?; + Some((name, contents)) + }) + .collect(); + files.sort_by_key(|(name, _)| *name); + files + } + /// Run all pending migrations pub async fn run(&self) -> Result { // First, ensure the migrations table exists (bootstrap) @@ -52,10 +43,9 @@ impl<'a> Migrator<'a> { let mut applied_count = 0; let mut skipped_count = 0; - for (name, sql) in MIGRATIONS { + for (name, sql) in Self::migrations() { // Skip the bootstrap migration after first run - if *name == "000_migrations.sql" && applied.contains(&"000_migrations.sql".to_string()) - { + if name == "000_migrations.sql" && applied.contains(&"000_migrations.sql".to_string()) { skipped_count += 1; continue; } @@ -86,8 +76,8 @@ impl<'a> Migrator<'a> { Err(_) => vec![], }; - let pending: Vec = MIGRATIONS - .iter() + let pending: Vec = Self::migrations() + .into_iter() .filter(|(name, _)| !applied.contains(&name.to_string())) .map(|(name, _)| name.to_string()) .collect(); @@ -97,8 +87,8 @@ impl<'a> Migrator<'a> { async fn ensure_migrations_table(&self) -> Result<(), IndexError> { // Run the bootstrap migration directly - let (_, sql) = MIGRATIONS - .iter() + let (_, sql) = Self::migrations() + .into_iter() .find(|(name, _)| *name == "000_migrations.sql") .expect("bootstrap migration must exist"); diff --git a/crates/weaver-index/src/clickhouse/schema.rs b/crates/weaver-index/src/clickhouse/schema.rs index 26162c0..6c51302 100644 --- a/crates/weaver-index/src/clickhouse/schema.rs +++ b/crates/weaver-index/src/clickhouse/schema.rs @@ -1,5 +1,6 @@ use chrono::{DateTime, Utc}; use clickhouse::Row; +use smol_str::SmolStr; /// Table names for production schema pub struct Tables; @@ -10,28 +11,55 @@ impl Tables { pub const RAW_ACCOUNT_EVENTS: &'static str = "raw_account_events"; pub const RAW_EVENTS_DLQ: &'static str = "raw_events_dlq"; pub const FIREHOSE_CURSOR: &'static str = "firehose_cursor"; + pub const ACCOUNT_REV_STATE: &'static str = "account_rev_state"; + pub const ACCOUNT_REV_STATE_MV: &'static str = "account_rev_state_mv"; + pub const MIGRATIONS: &'static str = "_migrations"; + + /// All tables and views in drop order (MVs before their source tables) + pub const ALL: &'static [&'static str] = &[ + Self::ACCOUNT_REV_STATE_MV, // MV first, depends on raw_records + Self::ACCOUNT_REV_STATE, + Self::RAW_RECORDS, + Self::RAW_IDENTITY_EVENTS, + Self::RAW_ACCOUNT_EVENTS, + Self::RAW_EVENTS_DLQ, + Self::FIREHOSE_CURSOR, + Self::MIGRATIONS, + ]; +} + +/// Validation states for records +pub mod validation { + pub const UNCHECKED: &str = "unchecked"; + pub const VALID: &str = "valid"; + pub const INVALID_REV: &str = "invalid_rev"; + pub const INVALID_GAP: &str = "invalid_gap"; + pub const INVALID_ACCOUNT: &str = "invalid_account"; } /// Row type for raw_records table /// Schema defined in migrations/clickhouse/001_raw_records.sql #[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] -pub struct RawRecord { - pub did: String, - pub collection: String, - pub rkey: String, - pub cid: String, - pub record: String, // JSON string - ClickHouse JSON type accepts string - pub operation: String, +pub struct RawRecordInsert { + pub did: SmolStr, + pub collection: SmolStr, + pub rkey: SmolStr, + pub cid: SmolStr, + pub rev: SmolStr, + pub record: SmolStr, // JSON string - ClickHouse JSON type accepts string + pub operation: SmolStr, pub seq: u64, #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] pub event_time: DateTime, + // Note: indexed_at has DEFAULT now64(3), omit from insert + // Note: validation_state has DEFAULT 'unchecked', omit from insert } /// Row type for raw_identity_events table #[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] pub struct RawIdentityEvent { - pub did: String, - pub handle: String, + pub did: SmolStr, + pub handle: SmolStr, pub seq: u64, #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] pub event_time: DateTime, @@ -40,9 +68,9 @@ pub struct RawIdentityEvent { /// Row type for raw_account_events table #[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] pub struct RawAccountEvent { - pub did: String, + pub did: SmolStr, pub active: u8, - pub status: String, + pub status: SmolStr, pub seq: u64, #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] pub event_time: DateTime, @@ -51,17 +79,29 @@ pub struct RawAccountEvent { /// Row type for raw_events_dlq table #[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] pub struct RawEventDlq { - pub event_type: String, - pub raw_data: String, // JSON string - pub error_message: String, + pub event_type: SmolStr, + pub raw_data: SmolStr, // JSON string + pub error_message: SmolStr, pub seq: u64, } /// Row type for firehose_cursor table #[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] pub struct FirehoseCursor { - pub consumer_id: String, + pub consumer_id: SmolStr, pub seq: u64, #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] pub event_time: DateTime, } + +/// Row type for reading finalized account_rev_state +/// Query with: SELECT did, argMaxMerge(last_rev), argMaxMerge(last_cid), maxMerge(last_seq), maxMerge(last_event_time) FROM account_rev_state GROUP BY did +#[derive(Debug, Clone, Row, serde::Serialize, serde::Deserialize)] +pub struct AccountRevState { + pub did: SmolStr, + pub last_rev: SmolStr, + pub last_cid: SmolStr, + pub last_seq: u64, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub last_event_time: DateTime, +} diff --git a/crates/weaver-index/src/config.rs b/crates/weaver-index/src/config.rs index 48cf2ea..1c23c89 100644 --- a/crates/weaver-index/src/config.rs +++ b/crates/weaver-index/src/config.rs @@ -1,4 +1,5 @@ use crate::error::{ConfigError, IndexError}; +use dashmap::DashSet; use url::Url; /// ClickHouse connection configuration @@ -84,11 +85,129 @@ impl FirehoseConfig { } } +use smol_str::{SmolStr, ToSmolStr}; + +/// Pre-parsed collection filter for efficient matching +#[derive(Debug, Clone)] +pub struct CollectionFilter { + /// Prefix patterns (from "foo.*" -> "foo.") + prefixes: Vec, + /// Exact match patterns (HashSet for O(1) lookup) + exact: DashSet, + /// True if filter is empty (accept all) + accept_all: bool, +} + +impl CollectionFilter { + /// Parse filter patterns into prefixes and exact matches + pub fn new(patterns: Vec) -> Self { + let mut prefixes = Vec::new(); + let exact = DashSet::new(); + + for pattern in patterns { + if let Some(prefix) = pattern.strip_suffix('*') { + prefixes.push(SmolStr::new(prefix)); + } else { + exact.insert(SmolStr::new(&pattern)); + } + } + + let accept_all = prefixes.is_empty() && exact.is_empty(); + Self { + prefixes, + exact, + accept_all, + } + } + + /// Check if a collection matches any pattern + #[inline] + pub fn matches(&self, collection: &str) -> bool { + if self.accept_all { + return true; + } + + // O(1) exact match check first + if self.exact.contains(collection) { + return true; + } + + // Prefix check - for small N, linear scan is fine + // Accumulate without early return to help branch predictor + let mut matched = false; + for prefix in &self.prefixes { + matched |= collection.starts_with(prefix.as_str()); + } + matched + } +} + +/// Indexer runtime configuration +#[derive(Debug, Clone)] +pub struct IndexerConfig { + /// Maximum records to batch before flushing to ClickHouse + pub batch_size: usize, + /// Maximum time (ms) before flushing even if batch isn't full + pub flush_interval_ms: u64, + /// Collection filter (pre-parsed patterns) + pub collections: CollectionFilter, +} + +impl Default for IndexerConfig { + fn default() -> Self { + Self { + batch_size: 1000, + flush_interval_ms: 1000, + collections: CollectionFilter::new(vec![ + SmolStr::new_static("sh.weaver.*"), + SmolStr::new_static("app.bsky.actor.profile"), + ]), + } + } +} + +impl IndexerConfig { + /// Load configuration from environment variables. + /// + /// Optional env vars: + /// - `INDEXER_BATCH_SIZE`: Max records per batch (default: 1000) + /// - `INDEXER_FLUSH_INTERVAL_MS`: Max ms between flushes (default: 1000) + /// - `INDEXER_COLLECTIONS`: Comma-separated collection patterns (default: sh.weaver.*,app.bsky.actor.profile) + /// Use * suffix for prefix matching, e.g., "sh.weaver.*" matches all sh.weaver.* collections + pub fn from_env() -> Self { + let batch_size = std::env::var("INDEXER_BATCH_SIZE") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(1000); + + let flush_interval_ms = std::env::var("INDEXER_FLUSH_INTERVAL_MS") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(1000); + + let patterns: Vec = std::env::var("INDEXER_COLLECTIONS") + .map(|s| s.split(',').map(|p| p.trim().to_smolstr()).collect()) + .unwrap_or_else(|_| { + vec![ + SmolStr::new_static("sh.weaver.*"), + SmolStr::new_static("app.bsky.actor.profile"), + ] + }); + + Self { + batch_size, + flush_interval_ms, + collections: CollectionFilter::new(patterns), + } + } +} + /// Combined configuration for the indexer #[derive(Debug, Clone)] pub struct Config { pub clickhouse: ClickHouseConfig, pub firehose: FirehoseConfig, + pub indexer: IndexerConfig, } impl Config { @@ -97,6 +216,7 @@ impl Config { Ok(Self { clickhouse: ClickHouseConfig::from_env()?, firehose: FirehoseConfig::from_env()?, + indexer: IndexerConfig::from_env(), }) } } diff --git a/crates/weaver-index/src/firehose.rs b/crates/weaver-index/src/firehose.rs index 3a1c7ba..6ba0676 100644 --- a/crates/weaver-index/src/firehose.rs +++ b/crates/weaver-index/src/firehose.rs @@ -2,6 +2,6 @@ mod consumer; mod records; pub use consumer::{ - FirehoseConsumer, MessageStream, SubscribeReposMessage, Commit, Identity, Account, Sync, + Account, Commit, FirehoseConsumer, Identity, MessageStream, SubscribeReposMessage, Sync, }; -pub use records::{extract_records, ExtractedRecord}; +pub use records::{ExtractedRecord, extract_records}; diff --git a/crates/weaver-index/src/firehose/records.rs b/crates/weaver-index/src/firehose/records.rs index fa64762..4aacd8d 100644 --- a/crates/weaver-index/src/firehose/records.rs +++ b/crates/weaver-index/src/firehose/records.rs @@ -1,5 +1,6 @@ use crate::error::{CarError, IndexError}; use bytes::Bytes; +use chrono::{DateTime, Utc}; use jacquard_repo::car::reader::parse_car_bytes; use smol_str::{SmolStr, ToSmolStr}; @@ -15,15 +16,17 @@ pub struct ExtractedRecord { /// Record key within the collection pub rkey: SmolStr, /// Content identifier - pub cid: String, + pub cid: SmolStr, + /// Repository revision (TID) - monotonically increasing per DID + pub rev: SmolStr, /// Operation type: "create", "update", or "delete" pub operation: SmolStr, /// Raw DAG-CBOR bytes of the record (None for deletes) pub cbor_bytes: Option, /// Sequence number from the firehose event pub seq: i64, - /// Event timestamp (milliseconds since epoch) - pub event_time_ms: i64, + /// Event timestamp + pub event_time: DateTime, } impl ExtractedRecord { @@ -61,7 +64,7 @@ pub async fn extract_records(commit: &Commit<'_>) -> Result message: e.to_string(), })?; - let event_time_ms = commit.time.as_ref().timestamp_millis(); + let event_time = commit.time.as_ref().with_timezone(&Utc); let mut records = Vec::with_capacity(commit.ops.len()); for op in &commit.ops { @@ -77,7 +80,7 @@ pub async fn extract_records(commit: &Commit<'_>) -> Result }; let operation = op.action.to_smolstr(); - let cid_str = op.cid.as_ref().map(|c| c.to_string()).unwrap_or_default(); + let cid_str = op.cid.as_ref().map(|c| c.to_smolstr()).unwrap_or_default(); // For creates/updates, look up the record in the CAR blocks let cbor_bytes = if let Some(cid_link) = &op.cid { @@ -97,10 +100,11 @@ pub async fn extract_records(commit: &Commit<'_>) -> Result collection, rkey, cid: cid_str, + rev: commit.rev.to_smolstr(), operation, cbor_bytes, seq: commit.seq, - event_time_ms, + event_time, }); } diff --git a/crates/weaver-index/src/indexer.rs b/crates/weaver-index/src/indexer.rs new file mode 100644 index 0000000..48de565 --- /dev/null +++ b/crates/weaver-index/src/indexer.rs @@ -0,0 +1,512 @@ +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use chrono::Utc; +use dashmap::DashMap; +use n0_future::StreamExt; +use smol_str::{SmolStr, ToSmolStr}; +use tracing::{debug, info, warn}; + +use chrono::DateTime; + +use crate::clickhouse::{ + AccountRevState, Client, FirehoseCursor, RawAccountEvent, RawIdentityEvent, RawRecordInsert, +}; +use crate::config::IndexerConfig; +use crate::error::{IndexError, Result}; +use crate::firehose::{ + Account, Commit, ExtractedRecord, FirehoseConsumer, Identity, MessageStream, + SubscribeReposMessage, extract_records, +}; + +/// Default consumer ID for cursor tracking +const CONSUMER_ID: &str = "main"; + +/// Per-account revision state for deduplication +#[derive(Debug, Clone)] +pub struct RevState { + pub last_rev: SmolStr, + pub last_cid: SmolStr, +} + +/// In-memory cache of per-account revision state +/// +/// Used for fast deduplication without hitting ClickHouse on every event. +/// Populated from account_rev_state table on startup, updated as events are processed. +pub struct RevCache { + inner: DashMap, +} + +impl RevCache { + pub fn new() -> Self { + Self { + inner: DashMap::new(), + } + } + + /// Load cache from ClickHouse account_rev_state table + pub async fn load_from_clickhouse(client: &Client) -> Result { + let query = r#" + SELECT + did, + argMaxMerge(last_rev) as last_rev, + argMaxMerge(last_cid) as last_cid, + maxMerge(last_seq) as last_seq, + maxMerge(last_event_time) as last_event_time + FROM account_rev_state + GROUP BY did + "#; + + let rows: Vec = + client.inner().query(query).fetch_all().await.map_err(|e| { + IndexError::ClickHouse(crate::error::ClickHouseError::Query { + message: "failed to load account rev state".into(), + source: e, + }) + })?; + + let cache = Self::new(); + for row in rows { + cache.inner.insert( + SmolStr::new(&row.did), + RevState { + last_rev: SmolStr::new(&row.last_rev), + last_cid: SmolStr::new(&row.last_cid), + }, + ); + } + + info!( + accounts = cache.inner.len(), + "loaded rev cache from clickhouse" + ); + Ok(cache) + } + + /// Check if we should process this commit (returns false if already seen) + pub fn should_process(&self, did: &str, rev: &str) -> bool { + match self.inner.get(did) { + Some(state) => rev > state.last_rev.as_str(), + None => true, // new account, always process + } + } + + /// Update cache after processing a commit + pub fn update(&self, did: &SmolStr, rev: &SmolStr, cid: &SmolStr) { + self.inner.insert( + did.clone(), + RevState { + last_rev: rev.clone(), + last_cid: cid.clone(), + }, + ); + } + + /// Get current cache size (number of accounts tracked) + pub fn len(&self) -> usize { + self.inner.len() + } + + pub fn is_empty(&self) -> bool { + self.inner.is_empty() + } +} + +impl Default for RevCache { + fn default() -> Self { + Self::new() + } +} + +/// Safety margin when resuming - back up this many sequence numbers +/// to ensure no gaps from incomplete batches or race conditions +const CURSOR_REWIND: i64 = 1000; + +/// Load cursor from ClickHouse for resuming +/// +/// Returns cursor with safety margin subtracted to ensure overlap +pub async fn load_cursor(client: &Client) -> Result> { + let query = format!( + r#" + SELECT consumer_id, seq, event_time + FROM firehose_cursor FINAL + WHERE consumer_id = '{}' + LIMIT 1 + "#, + CONSUMER_ID + ); + + let cursor: Option = client + .inner() + .query(&query) + .fetch_optional() + .await + .map_err(|e| crate::error::ClickHouseError::Query { + message: "failed to load cursor".into(), + source: e, + })?; + + if let Some(c) = &cursor { + let resume_at = (c.seq as i64).saturating_sub(CURSOR_REWIND); + info!( + saved_seq = c.seq, + resume_seq = resume_at, + rewind = CURSOR_REWIND, + "loaded cursor from clickhouse (with safety margin)" + ); + Ok(Some(resume_at)) + } else { + Ok(None) + } +} + +/// Main indexer that consumes firehose and writes to ClickHouse +pub struct Indexer { + client: Arc, + consumer: FirehoseConsumer, + rev_cache: RevCache, + config: IndexerConfig, +} + +impl Indexer { + /// Create a new indexer + pub async fn new( + client: Client, + consumer: FirehoseConsumer, + config: IndexerConfig, + ) -> Result { + let client = Arc::new(client); + + // Load rev cache from ClickHouse + let rev_cache = RevCache::load_from_clickhouse(&client).await?; + + Ok(Self { + client, + consumer, + rev_cache, + config, + }) + } + + /// Save cursor to ClickHouse + async fn save_cursor(&self, seq: u64, event_time: DateTime) -> Result<()> { + let query = format!( + "INSERT INTO firehose_cursor (consumer_id, seq, event_time) VALUES ('{}', {}, {})", + CONSUMER_ID, + seq, + event_time.timestamp_millis() + ); + + self.client.execute(&query).await?; + debug!(seq, "saved cursor"); + Ok(()) + } + + /// Run the indexer loop + pub async fn run(&self) -> Result<()> { + info!("connecting to firehose..."); + let mut stream: MessageStream = self.consumer.connect().await?; + + // Inserters handle batching internally based on config + let mut records = self.client.inserter::("raw_records"); + let mut identities = self + .client + .inserter::("raw_identity_events"); + let mut accounts = self + .client + .inserter::("raw_account_events"); + + // Stats and cursor tracking + let mut processed: u64 = 0; + let mut skipped: u64 = 0; + let mut last_seq: u64 = 0; + let mut last_event_time = Utc::now(); + let mut last_stats = Instant::now(); + let mut last_cursor_save = Instant::now(); + + info!("starting indexer loop"); + + while let Some(result) = stream.next().await { + let msg = match result { + Ok(msg) => msg, + Err(e) => { + warn!(error = ?e, "firehose stream error"); + continue; + } + }; + + // Track seq from any message type that has it + match &msg { + SubscribeReposMessage::Commit(c) => { + last_seq = c.seq as u64; + last_event_time = c.time.as_ref().with_timezone(&Utc); + } + SubscribeReposMessage::Identity(i) => { + last_seq = i.seq as u64; + last_event_time = i.time.as_ref().with_timezone(&Utc); + } + SubscribeReposMessage::Account(a) => { + last_seq = a.seq as u64; + last_event_time = a.time.as_ref().with_timezone(&Utc); + } + _ => {} + } + + match msg { + SubscribeReposMessage::Commit(commit) => { + if self + .process_commit(&commit, &mut records, &mut skipped) + .await? + { + processed += 1; + } + } + SubscribeReposMessage::Identity(identity) => { + write_identity(&identity, &mut identities).await?; + } + SubscribeReposMessage::Account(account) => { + write_account(&account, &mut accounts).await?; + } + SubscribeReposMessage::Sync(_) => { + debug!("received sync (tooBig) event, skipping"); + } + _ => {} + } + + // commit() flushes if internal thresholds met, otherwise no-op + records + .commit() + .await + .map_err(|e| crate::error::ClickHouseError::Query { + message: "commit failed".into(), + source: e, + })?; + + // Periodic stats and cursor save (every 10s) + if last_stats.elapsed() >= Duration::from_secs(10) { + info!( + processed, + skipped, + last_seq, + rev_cache_size = self.rev_cache.len(), + "indexer stats" + ); + last_stats = Instant::now(); + } + + // Save cursor every 30s + if last_cursor_save.elapsed() >= Duration::from_secs(30) && last_seq > 0 { + if let Err(e) = self.save_cursor(last_seq, last_event_time).await { + warn!(error = ?e, "failed to save cursor"); + } + last_cursor_save = Instant::now(); + } + } + + // Final flush + records + .end() + .await + .map_err(|e| crate::error::ClickHouseError::Query { + message: "final flush failed".into(), + source: e, + })?; + identities + .end() + .await + .map_err(|e| crate::error::ClickHouseError::Query { + message: "final flush failed".into(), + source: e, + })?; + accounts + .end() + .await + .map_err(|e| crate::error::ClickHouseError::Query { + message: "final flush failed".into(), + source: e, + })?; + + // Final cursor save + if last_seq > 0 { + self.save_cursor(last_seq, last_event_time).await?; + } + + info!(last_seq, "firehose stream ended"); + Ok(()) + } + + async fn process_commit( + &self, + commit: &Commit<'_>, + inserter: &mut clickhouse::inserter::Inserter, + skipped: &mut u64, + ) -> Result { + let did = commit.repo.as_ref(); + let rev = commit.rev.as_ref(); + + // Dedup check + if !self.rev_cache.should_process(did, rev) { + *skipped += 1; + return Ok(false); + } + + // Extract and write records + for record in extract_records(commit).await? { + // Collection filter - skip early before JSON conversion + if !self.config.collections.matches(&record.collection) { + continue; + } + + let json = record.to_json()?.unwrap_or_else(|| "{}".to_string()); + + // Fire and forget delete handling + if record.operation == "delete" { + let client = self.client.clone(); + let record_clone = record.clone(); + tokio::spawn(async move { + if let Err(e) = handle_delete(&client, record_clone).await { + warn!(error = ?e, "delete handling failed"); + } + }); + } + + inserter + .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.operation.clone(), + seq: record.seq as u64, + event_time: record.event_time, + }) + .await + .map_err(|e| crate::error::ClickHouseError::Query { + message: "write failed".into(), + source: e, + })?; + } + + // Update rev cache + self.rev_cache.update( + &SmolStr::new(did), + &SmolStr::new(rev), + &commit.commit.0.to_smolstr(), + ); + + Ok(true) + } +} + +async fn write_identity( + identity: &Identity<'_>, + inserter: &mut clickhouse::inserter::Inserter, +) -> Result<()> { + inserter + .write(&RawIdentityEvent { + did: identity.did.to_smolstr(), + handle: identity + .handle + .as_ref() + .map(|h| h.as_ref().to_smolstr()) + .unwrap_or_default(), + seq: identity.seq as u64, + event_time: identity.time.as_ref().with_timezone(&Utc), + }) + .await + .map_err(|e| crate::error::ClickHouseError::Query { + message: "write failed".into(), + source: e, + })?; + Ok(()) +} + +async fn write_account( + account: &Account<'_>, + inserter: &mut clickhouse::inserter::Inserter, +) -> Result<()> { + inserter + .write(&RawAccountEvent { + did: account.did.to_smolstr(), + active: if account.active { 1 } else { 0 }, + status: account + .status + .as_ref() + .map(|s| s.as_ref().to_smolstr()) + .unwrap_or_default(), + seq: account.seq as u64, + event_time: account.time.as_ref().with_timezone(&Utc), + }) + .await + .map_err(|e| crate::error::ClickHouseError::Query { + message: "write failed".into(), + source: e, + })?; + Ok(()) +} + +/// Handle a delete event with poll-then-stub logic +/// +/// For deletes, we need to look up the original record to know what was deleted +/// (e.g., which notebook a like was for). If the record doesn't exist yet +/// (out-of-order events), we poll for up to 15 seconds before creating a stub tombstone. +/// Minimal struct for delete lookups - just the fields we need to process the delete +#[derive(Debug, Clone, clickhouse::Row, serde::Deserialize)] +struct LookupRawRecord { + did: SmolStr, + collection: SmolStr, + rkey: SmolStr, + record: SmolStr, // JSON string of the original record +} + +async fn handle_delete(client: &Client, record: ExtractedRecord) -> Result<()> { + let deadline = Instant::now() + Duration::from_secs(15); + + loop { + // Try to find the record by CID + let query = format!( + r#" + SELECT did, collection, rkey, record + FROM raw_records + WHERE did = '{}' AND cid = '{}' + ORDER BY event_time DESC + LIMIT 1 + "#, + record.did, record.cid + ); + + let original: Option = client + .inner() + .query(&query) + .fetch_optional() + .await + .map_err(|e| crate::error::ClickHouseError::Query { + message: "delete lookup failed".into(), + source: e, + })?; + + if let Some(_original) = original { + // Found the record - the main insert path already handles creating + // the delete row, so we're done. In phase 2, this is where we'd + // parse original.record and insert count deltas for denormalized tables. + debug!(did = %record.did, cid = %record.cid, "delete found original record"); + return Ok(()); + } + + if Instant::now() > deadline { + // Gave up - create stub tombstone + // The record will be inserted via the main batch path with operation='delete' + // and empty record content, which serves as our stub tombstone + warn!( + did = %record.did, + cid = %record.cid, + "delete timeout, stub tombstone will be created" + ); + return Ok(()); + } + + tokio::time::sleep(Duration::from_secs(1)).await; + } +} diff --git a/crates/weaver-index/src/lib.rs b/crates/weaver-index/src/lib.rs index f175934..70a2b6b 100644 --- a/crates/weaver-index/src/lib.rs +++ b/crates/weaver-index/src/lib.rs @@ -2,6 +2,8 @@ pub mod clickhouse; pub mod config; pub mod error; pub mod firehose; +pub mod indexer; pub use config::Config; pub use error::{IndexError, Result}; +pub use indexer::{load_cursor, Indexer}; -- knotmirror 0.0.1