diff --git a/Cargo.lock b/Cargo.lock index 2cb4c53..2db799d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11789,6 +11789,7 @@ dependencies = [ "jacquard-repo", "miette 7.6.0", "n0-future 0.1.3", + "regex", "rusqlite", "rusqlite_migration", "serde", diff --git a/crates/weaver-index/Cargo.toml b/crates/weaver-index/Cargo.toml index 166ad2b..3ea61a9 100644 --- a/crates/weaver-index/Cargo.toml +++ b/crates/weaver-index/Cargo.toml @@ -68,6 +68,7 @@ humansize = "2.0" base64 = "0.22" dashmap = "6" include_dir = "0.7.4" +regex = "1" # WebSocket (for tap consumer) tokio-tungstenite = { version = "0.26", features = ["native-tls"] } diff --git a/crates/weaver-index/migrations/clickhouse/001_raw_records.sql b/crates/weaver-index/migrations/clickhouse/001_raw_records.sql index 8dfda30..93e7851 100644 --- a/crates/weaver-index/migrations/clickhouse/001_raw_records.sql +++ b/crates/weaver-index/migrations/clickhouse/001_raw_records.sql @@ -1,7 +1,8 @@ -- Raw records from firehose/jetstream -- Core table for all AT Protocol records before denormalization -- --- Uses ReplacingMergeTree to deduplicate on (collection, did, rkey) keeping latest indexed_at +-- Append-only log using plain MergeTree - all versions preserved for audit/rollback. +-- Query-time deduplication via ORDER BY + LIMIT or window functions. -- JSON column stores full record, extract fields only when needed for ORDER BY/WHERE/JOINs CREATE TABLE IF NOT EXISTS raw_records ( @@ -10,16 +11,16 @@ CREATE TABLE IF NOT EXISTS raw_records ( collection LowCardinality(String), rkey String, - -- Content identifier from the record + -- Content identifier from the record (content-addressed hash) cid String, - -- Repository revision (TID) - monotonically increasing per DID, used for dedup/ordering + -- Repository revision (TID) - monotonically increasing per DID, used for ordering rev String, -- Full record as native JSON (schema-flexible, queryable with record.field.subfield) record JSON, - -- Operation: 'create', 'update', 'delete' + -- Operation: 'create', 'update', 'delete', 'cache' (fetched on-demand) operation LowCardinality(String), -- Firehose sequence number (metadata only, not for ordering - can jump on relay restart) @@ -49,6 +50,5 @@ CREATE TABLE IF NOT EXISTS raw_records ( SELECT * ORDER BY (did, cid) ) ) -ENGINE = ReplacingMergeTree(indexed_at) -ORDER BY (collection, did, rkey, event_time, indexed_at) -SETTINGS deduplicate_merge_projection_mode = 'drop'; +ENGINE = MergeTree() +ORDER BY (collection, did, rkey, event_time, indexed_at); diff --git a/crates/weaver-index/migrations/clickhouse/008_handle_mappings.sql b/crates/weaver-index/migrations/clickhouse/008_handle_mappings.sql new file mode 100644 index 0000000..dc34ce8 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/008_handle_mappings.sql @@ -0,0 +1,37 @@ +-- Handle → DID mappings with account status tracking +-- +-- Updated from three sources: +-- 1. Identity events (firehose) - handle claims/changes via 009_handle_mappings_identity_mv.sql +-- 2. Account events (firehose) - takedowns/suspensions/deletions via 010_handle_mappings_account_mv.sql +-- 3. Resolution cache flush - XRPC handle resolution results (manual inserts) +-- +-- Query pattern: ORDER BY freed ASC, event_time DESC to get active mapping first + +CREATE TABLE IF NOT EXISTS handle_mappings ( + handle String, + did String, + + -- 0 = active, 1 = account deactivated/suspended/deleted + freed UInt8 DEFAULT 0, + + -- 'active' | 'takendown' | 'suspended' | 'deleted' | 'deactivated' + account_status LowCardinality(String) DEFAULT 'active', + + -- 'identity' (firehose) | 'account' (firehose) | 'resolution' (xrpc cache) + source LowCardinality(String), + + -- Canonical event time (firehose event or resolution time) + event_time DateTime64(3), + + -- When we indexed this mapping + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Fast DID → handle lookups (for account events, profile hydration) + -- Query with ORDER BY freed ASC, event_time DESC to get active mapping + PROJECTION by_did ( + SELECT * ORDER BY (did, freed, event_time) + ) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (handle, did) +SETTINGS deduplicate_merge_projection_mode = 'drop'; diff --git a/crates/weaver-index/migrations/clickhouse/009_handle_mappings_identity_mv.sql b/crates/weaver-index/migrations/clickhouse/009_handle_mappings_identity_mv.sql new file mode 100644 index 0000000..8b0e6e6 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/009_handle_mappings_identity_mv.sql @@ -0,0 +1,13 @@ +-- Auto-populate handle_mappings from identity events when handle is present + +CREATE MATERIALIZED VIEW IF NOT EXISTS handle_mappings_from_identity_mv TO handle_mappings AS +SELECT + handle, + did, + 0 as freed, + 'active' as account_status, + 'identity' as source, + event_time, + now64(3) as indexed_at +FROM raw_identity_events +WHERE handle != '' diff --git a/crates/weaver-index/migrations/clickhouse/010_handle_mappings_account_mv.sql b/crates/weaver-index/migrations/clickhouse/010_handle_mappings_account_mv.sql new file mode 100644 index 0000000..7016a79 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/010_handle_mappings_account_mv.sql @@ -0,0 +1,16 @@ +-- Auto-populate freed status from account events +-- JOINs against handle_mappings to find current handle for the DID +-- If no mapping exists yet, the JOIN fails silently (can't free unknown handles) + +CREATE MATERIALIZED VIEW IF NOT EXISTS handle_mappings_from_account_mv TO handle_mappings AS +SELECT + h.handle, + a.did, + 1 as freed, + a.status as account_status, + 'account' as source, + a.event_time, + now64(3) as indexed_at +FROM raw_account_events a +INNER JOIN handle_mappings h ON h.did = a.did AND h.freed = 0 +WHERE a.active = 0 AND a.status != '' diff --git a/crates/weaver-index/migrations/clickhouse/011_profiles_weaver.sql b/crates/weaver-index/migrations/clickhouse/011_profiles_weaver.sql new file mode 100644 index 0000000..4e4478b --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/011_profiles_weaver.sql @@ -0,0 +1,25 @@ +-- Weaver profile source table +-- Populated by MV from raw_records, merged into profiles by refreshable MV + +CREATE TABLE IF NOT EXISTS profiles_weaver ( + did String, + + -- Raw profile JSON + profile String, + + -- Extracted fields for coalescing + display_name String DEFAULT '', + description String DEFAULT '', + avatar_cid String DEFAULT '', + banner_cid String DEFAULT '', + + -- Timestamps + created_at DateTime64(3) DEFAULT toDateTime64(0, 3), + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Soft delete (epoch = not deleted) + deleted_at DateTime64(3) DEFAULT toDateTime64(0, 3) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY did diff --git a/crates/weaver-index/migrations/clickhouse/012_profiles_weaver_mv.sql b/crates/weaver-index/migrations/clickhouse/012_profiles_weaver_mv.sql new file mode 100644 index 0000000..dc5027c --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/012_profiles_weaver_mv.sql @@ -0,0 +1,16 @@ +-- Populate profiles_weaver from raw_records + +CREATE MATERIALIZED VIEW IF NOT EXISTS profiles_weaver_mv TO profiles_weaver AS +SELECT + did, + toString(record) as profile, + coalesce(record.displayName, '') as display_name, + coalesce(record.description, '') as description, + coalesce(record.avatar.ref.`$link`, '') as avatar_cid, + coalesce(record.banner.ref.`$link`, '') as banner_cid, + coalesce(toDateTime64(record.createdAt, 3), toDateTime64(0, 3)) as created_at, + event_time, + now64(3) as indexed_at, + if(operation = 'delete', event_time, toDateTime64(0, 3)) as deleted_at +FROM raw_records +WHERE collection = 'sh.weaver.actor.profile' diff --git a/crates/weaver-index/migrations/clickhouse/013_profiles_bsky.sql b/crates/weaver-index/migrations/clickhouse/013_profiles_bsky.sql new file mode 100644 index 0000000..67f0aef --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/013_profiles_bsky.sql @@ -0,0 +1,25 @@ +-- Bluesky profile source table +-- Populated by MV from raw_records, merged into profiles by refreshable MV + +CREATE TABLE IF NOT EXISTS profiles_bsky ( + did String, + + -- Raw profile JSON + profile String, + + -- Extracted fields for coalescing + display_name String DEFAULT '', + description String DEFAULT '', + avatar_cid String DEFAULT '', + banner_cid String DEFAULT '', + + -- Timestamps + created_at DateTime64(3) DEFAULT toDateTime64(0, 3), + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Soft delete (epoch = not deleted) + deleted_at DateTime64(3) DEFAULT toDateTime64(0, 3) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY did diff --git a/crates/weaver-index/migrations/clickhouse/014_profiles_bsky_mv.sql b/crates/weaver-index/migrations/clickhouse/014_profiles_bsky_mv.sql new file mode 100644 index 0000000..6193251 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/014_profiles_bsky_mv.sql @@ -0,0 +1,16 @@ +-- Populate profiles_bsky from raw_records + +CREATE MATERIALIZED VIEW IF NOT EXISTS profiles_bsky_mv TO profiles_bsky AS +SELECT + did, + toString(record) as profile, + coalesce(record.displayName, '') as display_name, + coalesce(record.description, '') as description, + coalesce(record.avatar.ref.`$link`, '') as avatar_cid, + coalesce(record.banner.ref.`$link`, '') as banner_cid, + coalesce(toDateTime64(record.createdAt, 3), toDateTime64(0, 3)) as created_at, + event_time, + now64(3) as indexed_at, + if(operation = 'delete', event_time, toDateTime64(0, 3)) as deleted_at +FROM raw_records +WHERE collection = 'app.bsky.actor.profile' diff --git a/crates/weaver-index/migrations/clickhouse/015_profiles.sql b/crates/weaver-index/migrations/clickhouse/015_profiles.sql new file mode 100644 index 0000000..3d450a1 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/015_profiles.sql @@ -0,0 +1,41 @@ +-- Unified profiles view +-- Refreshable MV that merges weaver + bsky profiles with handle resolution +-- Queries are pure reads, no merge computation needed + +CREATE MATERIALIZED VIEW IF NOT EXISTS profiles +REFRESH EVERY 1 MINUTE +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY did +AS SELECT + if(w.did != '', w.did, b.did) as did, + + -- Handle from handle_mappings (empty if not resolved yet) + coalesce(h.handle, '') as handle, + + -- Raw profiles per source + coalesce(w.profile, '') as weaver_profile, + coalesce(b.profile, '') as bsky_profile, + + -- Coalesced fields (weaver > bsky priority) + coalesce(nullIf(w.display_name, ''), b.display_name, '') as display_name, + coalesce(nullIf(w.description, ''), b.description, '') as description, + coalesce(nullIf(w.avatar_cid, ''), b.avatar_cid, '') as avatar_cid, + coalesce(nullIf(w.banner_cid, ''), b.banner_cid, '') as banner_cid, + + -- Presence flags + if(w.did != '', 1, 0) as has_weaver, + if(b.did != '', 1, 0) as has_bsky, + + -- Timestamps + coalesce(w.created_at, b.created_at, toDateTime64(0, 3)) as created_at, + greatest(coalesce(w.event_time, toDateTime64(0, 3)), coalesce(b.event_time, toDateTime64(0, 3))) as event_time, + now64(3) as indexed_at +FROM (SELECT * FROM profiles_weaver WHERE deleted_at = toDateTime64(0, 3)) w +FULL OUTER JOIN (SELECT * FROM profiles_bsky WHERE deleted_at = toDateTime64(0, 3)) b ON w.did = b.did +LEFT JOIN ( + SELECT did, argMax(handle, event_time) as handle + FROM handle_mappings + WHERE freed = 0 AND did != '' + GROUP BY did +) h ON h.did = if(w.did != '', w.did, b.did) +WHERE if(w.did != '', w.did, b.did) != '' diff --git a/crates/weaver-index/migrations/clickhouse/016_profile_counts.sql b/crates/weaver-index/migrations/clickhouse/016_profile_counts.sql new file mode 100644 index 0000000..fbe8d85 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/016_profile_counts.sql @@ -0,0 +1,17 @@ +-- Profile counts aggregated from graph tables +-- Updated by MVs from follows, notebooks, entries (added later with those tables) +-- Joined with profiles at query time + +CREATE TABLE IF NOT EXISTS profile_counts ( + did String, + + -- Signed for increment/decrement from MVs + follower_count Int64 DEFAULT 0, + following_count Int64 DEFAULT 0, + notebook_count Int64 DEFAULT 0, + entry_count Int64 DEFAULT 0, + + indexed_at DateTime64(3) DEFAULT now64(3) +) +ENGINE = SummingMergeTree((follower_count, following_count, notebook_count, entry_count)) +ORDER BY did diff --git a/crates/weaver-index/migrations/clickhouse/017_notebooks.sql b/crates/weaver-index/migrations/clickhouse/017_notebooks.sql new file mode 100644 index 0000000..b92665d --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/017_notebooks.sql @@ -0,0 +1,36 @@ +-- Notebooks table +-- Populated by MV from raw_records + +CREATE TABLE IF NOT EXISTS notebooks ( + -- Identity + did String, + rkey String, + cid String, + + -- Materialized URI for convenience + uri String MATERIALIZED concat('at://', did, '/sh.weaver.notebook.book/', rkey), + + -- Extracted fields + title String DEFAULT '', + path String DEFAULT '', + description String DEFAULT '', + tags Array(String) DEFAULT [], + publish_global UInt8 DEFAULT 0, + + -- Authors array (DIDs) + author_dids Array(String) DEFAULT [], + + -- Entry count (length of entryList) + entry_count UInt32 DEFAULT 0, + + -- Timestamps + created_at DateTime64(3) DEFAULT toDateTime64(0, 3), + updated_at DateTime64(3) DEFAULT toDateTime64(0, 3), + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Soft delete (epoch = not deleted) + deleted_at DateTime64(3) DEFAULT toDateTime64(0, 3) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (did, rkey) diff --git a/crates/weaver-index/migrations/clickhouse/018_notebooks_mv.sql b/crates/weaver-index/migrations/clickhouse/018_notebooks_mv.sql new file mode 100644 index 0000000..5d64174 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/018_notebooks_mv.sql @@ -0,0 +1,21 @@ +-- Populate notebooks from raw_records + +CREATE MATERIALIZED VIEW IF NOT EXISTS notebooks_mv TO notebooks AS +SELECT + did, + rkey, + cid, + coalesce(record.title, '') as title, + coalesce(record.path, '') as path, + '' as description, -- notebooks don't have description field + JSONExtract(toString(record), 'tags', 'Array(String)') as tags, + if(record.publishGlobal = true, 1, 0) as publish_global, + arrayMap(x -> JSONExtractString(x, 'did'), JSONExtractArrayRaw(toString(record), 'authors')) as author_dids, + length(JSONExtractArrayRaw(toString(record), 'entryList')) as entry_count, + coalesce(toDateTime64(record.createdAt, 3), toDateTime64(0, 3)) as created_at, + coalesce(toDateTime64(record.updatedAt, 3), toDateTime64(0, 3)) as updated_at, + event_time, + now64(3) as indexed_at, + if(operation = 'delete', event_time, toDateTime64(0, 3)) as deleted_at +FROM raw_records +WHERE collection = 'sh.weaver.notebook.book' diff --git a/crates/weaver-index/migrations/clickhouse/019_notebook_counts.sql b/crates/weaver-index/migrations/clickhouse/019_notebook_counts.sql new file mode 100644 index 0000000..cf8821c --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/019_notebook_counts.sql @@ -0,0 +1,17 @@ +-- Notebook engagement counts +-- Updated by MVs from likes, bookmarks, subscriptions (added later with graph tables) +-- Joined with notebooks at query time + +CREATE TABLE IF NOT EXISTS notebook_counts ( + did String, + rkey String, + + -- Signed for increment/decrement from MVs + like_count Int64 DEFAULT 0, + bookmark_count Int64 DEFAULT 0, + subscriber_count Int64 DEFAULT 0, + + indexed_at DateTime64(3) DEFAULT now64(3) +) +ENGINE = SummingMergeTree((like_count, bookmark_count, subscriber_count)) +ORDER BY (did, rkey) diff --git a/crates/weaver-index/migrations/clickhouse/020_entries.sql b/crates/weaver-index/migrations/clickhouse/020_entries.sql new file mode 100644 index 0000000..075612f --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/020_entries.sql @@ -0,0 +1,35 @@ +-- Entries table +-- Populated by MV from raw_records + +CREATE TABLE IF NOT EXISTS entries ( + -- Identity + did String, + rkey String, + cid String, + + -- Materialized URI for convenience + uri String MATERIALIZED concat('at://', did, '/sh.weaver.notebook.entry/', rkey), + + -- Extracted fields + title String DEFAULT '', + path String DEFAULT '', + content String DEFAULT '', + tags Array(String) DEFAULT [], + + -- Authors array (DIDs) + author_dids Array(String) DEFAULT [], + + -- Content preview (first 500 chars for feeds) + content_preview String DEFAULT '', + + -- Timestamps + created_at DateTime64(3) DEFAULT toDateTime64(0, 3), + updated_at DateTime64(3) DEFAULT toDateTime64(0, 3), + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Soft delete (epoch = not deleted) + deleted_at DateTime64(3) DEFAULT toDateTime64(0, 3) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (did, rkey) diff --git a/crates/weaver-index/migrations/clickhouse/021_entries_mv.sql b/crates/weaver-index/migrations/clickhouse/021_entries_mv.sql new file mode 100644 index 0000000..92ab66a --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/021_entries_mv.sql @@ -0,0 +1,20 @@ +-- Populate entries from raw_records + +CREATE MATERIALIZED VIEW IF NOT EXISTS entries_mv TO entries AS +SELECT + did, + rkey, + cid, + coalesce(record.title, '') as title, + coalesce(record.path, '') as path, + coalesce(record.content, '') as content, + JSONExtract(toString(record), 'tags', 'Array(String)') as tags, + arrayMap(x -> JSONExtractString(x, 'did'), JSONExtractArrayRaw(toString(record), 'authors')) as author_dids, + substring(coalesce(record.content, ''), 1, 500) as content_preview, + coalesce(toDateTime64(record.createdAt, 3), toDateTime64(0, 3)) as created_at, + coalesce(toDateTime64(record.updatedAt, 3), toDateTime64(0, 3)) as updated_at, + event_time, + now64(3) as indexed_at, + if(operation = 'delete', event_time, toDateTime64(0, 3)) as deleted_at +FROM raw_records +WHERE collection = 'sh.weaver.notebook.entry' diff --git a/crates/weaver-index/migrations/clickhouse/022_entry_counts.sql b/crates/weaver-index/migrations/clickhouse/022_entry_counts.sql new file mode 100644 index 0000000..1ae1fea --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/022_entry_counts.sql @@ -0,0 +1,16 @@ +-- Entry engagement counts +-- Updated by MVs from likes, bookmarks (added later with graph tables) +-- Joined with entries at query time + +CREATE TABLE IF NOT EXISTS entry_counts ( + did String, + rkey String, + + -- Signed for increment/decrement from MVs + like_count Int64 DEFAULT 0, + bookmark_count Int64 DEFAULT 0, + + indexed_at DateTime64(3) DEFAULT now64(3) +) +ENGINE = SummingMergeTree((like_count, bookmark_count)) +ORDER BY (did, rkey) diff --git a/crates/weaver-index/migrations/clickhouse/023_drafts.sql b/crates/weaver-index/migrations/clickhouse/023_drafts.sql new file mode 100644 index 0000000..b0184cf --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/023_drafts.sql @@ -0,0 +1,22 @@ +-- Draft stub records +-- Anchors for unpublished content, enables draft discovery via queries + +CREATE TABLE IF NOT EXISTS drafts ( + -- Identity + did String, + rkey String, + cid String, + + -- Materialized URI for convenience + uri String MATERIALIZED concat('at://', did, '/sh.weaver.edit.draft/', rkey), + + -- Timestamps + created_at DateTime64(3) DEFAULT toDateTime64(0, 3), + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Soft delete (epoch = not deleted) + deleted_at DateTime64(3) DEFAULT toDateTime64(0, 3) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (did, rkey) diff --git a/crates/weaver-index/migrations/clickhouse/024_drafts_mv.sql b/crates/weaver-index/migrations/clickhouse/024_drafts_mv.sql new file mode 100644 index 0000000..9da1cf3 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/024_drafts_mv.sql @@ -0,0 +1,13 @@ +-- Populate drafts from raw_records + +CREATE MATERIALIZED VIEW IF NOT EXISTS drafts_mv TO drafts AS +SELECT + did, + rkey, + cid, + coalesce(toDateTime64(record.createdAt, 3), toDateTime64(0, 3)) as created_at, + event_time, + now64(3) as indexed_at, + if(operation = 'delete', event_time, toDateTime64(0, 3)) as deleted_at +FROM raw_records +WHERE collection = 'sh.weaver.edit.draft' diff --git a/crates/weaver-index/migrations/clickhouse/025_edit_nodes.sql b/crates/weaver-index/migrations/clickhouse/025_edit_nodes.sql new file mode 100644 index 0000000..d1090f3 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/025_edit_nodes.sql @@ -0,0 +1,46 @@ +-- Edit nodes (roots and diffs combined) +-- Used for querying edit history from ClickHouse +-- Real-time collab queries use SQLite hot tier + +CREATE TABLE IF NOT EXISTS edit_nodes ( + -- Identity + did String, + rkey String, + cid String, + collection LowCardinality(String), -- 'sh.weaver.edit.root' or 'sh.weaver.edit.diff' + + -- Materialized URI + uri String MATERIALIZED concat('at://', did, '/', collection, '/', rkey), + + -- Node type derived from collection + node_type LowCardinality(String), -- 'root' or 'diff' + + -- Resource being edited (extracted from doc.value) + -- One of these will be populated depending on doc type + resource_type LowCardinality(String) DEFAULT '', -- 'entry', 'notebook', 'draft' + resource_did String DEFAULT '', + resource_rkey String DEFAULT '', + resource_collection LowCardinality(String) DEFAULT '', + + -- For diffs: reference to root + root_did String DEFAULT '', + root_rkey String DEFAULT '', + + -- For diffs: reference to previous node + prev_did String DEFAULT '', + prev_rkey String DEFAULT '', + + -- Whether this has inline diff data vs blob snapshot + has_inline_diff UInt8 DEFAULT 0, + has_snapshot UInt8 DEFAULT 0, + + -- Timestamps + created_at DateTime64(3) DEFAULT toDateTime64(0, 3), + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Soft delete (epoch = not deleted) + deleted_at DateTime64(3) DEFAULT toDateTime64(0, 3) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (did, rkey) diff --git a/crates/weaver-index/migrations/clickhouse/026_edit_roots_mv.sql b/crates/weaver-index/migrations/clickhouse/026_edit_roots_mv.sql new file mode 100644 index 0000000..12d6d5b --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/026_edit_roots_mv.sql @@ -0,0 +1,62 @@ +-- Populate edit_nodes from edit.root records + +CREATE MATERIALIZED VIEW IF NOT EXISTS edit_roots_mv TO edit_nodes AS +SELECT + did, + rkey, + cid, + 'sh.weaver.edit.root' as collection, + 'root' as node_type, + + -- Extract resource type and ref from doc.value union + multiIf( + toString(record.doc.value.entry.uri) != '', 'entry', + toString(record.doc.value.notebook.uri) != '', 'notebook', + toString(record.doc.value.draftKey) != '', 'draft', + '' + ) as resource_type, + + -- Extract resource DID (parse from URI or empty) + multiIf( + toString(record.doc.value.entry.uri) != '', + splitByChar('/', replaceOne(toString(record.doc.value.entry.uri), 'at://', ''))[1], + toString(record.doc.value.notebook.uri) != '', + splitByChar('/', replaceOne(toString(record.doc.value.notebook.uri), 'at://', ''))[1], + '' + ) as resource_did, + + -- Extract resource rkey (parse from URI or use draftKey) + multiIf( + toString(record.doc.value.entry.uri) != '', + splitByChar('/', replaceOne(toString(record.doc.value.entry.uri), 'at://', ''))[3], + toString(record.doc.value.notebook.uri) != '', + splitByChar('/', replaceOne(toString(record.doc.value.notebook.uri), 'at://', ''))[3], + toString(record.doc.value.draftKey) != '', + toString(record.doc.value.draftKey), + '' + ) as resource_rkey, + + -- Extract resource collection + multiIf( + toString(record.doc.value.entry.uri) != '', 'sh.weaver.notebook.entry', + toString(record.doc.value.notebook.uri) != '', 'sh.weaver.notebook.book', + toString(record.doc.value.draftKey) != '', 'sh.weaver.edit.draft', + '' + ) as resource_collection, + + -- Roots don't have root/prev refs + '' as root_did, + '' as root_rkey, + '' as prev_did, + '' as prev_rkey, + + -- Roots always have snapshot + 0 as has_inline_diff, + 1 as has_snapshot, + + event_time as created_at, + event_time, + now64(3) as indexed_at, + if(operation = 'delete', event_time, toDateTime64(0, 3)) as deleted_at +FROM raw_records +WHERE collection = 'sh.weaver.edit.root' diff --git a/crates/weaver-index/migrations/clickhouse/027_edit_diffs_mv.sql b/crates/weaver-index/migrations/clickhouse/027_edit_diffs_mv.sql new file mode 100644 index 0000000..c1afad3 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/027_edit_diffs_mv.sql @@ -0,0 +1,68 @@ +-- Populate edit_nodes from edit.diff records + +CREATE MATERIALIZED VIEW IF NOT EXISTS edit_diffs_mv TO edit_nodes AS +SELECT + did, + rkey, + cid, + 'sh.weaver.edit.diff' as collection, + 'diff' as node_type, + + -- Extract resource type and ref from doc.value union + multiIf( + toString(record.doc.value.entry.uri) != '', 'entry', + toString(record.doc.value.notebook.uri) != '', 'notebook', + toString(record.doc.value.draftKey) != '', 'draft', + '' + ) as resource_type, + + -- Extract resource DID + multiIf( + toString(record.doc.value.entry.uri) != '', + splitByChar('/', replaceOne(toString(record.doc.value.entry.uri), 'at://', ''))[1], + toString(record.doc.value.notebook.uri) != '', + splitByChar('/', replaceOne(toString(record.doc.value.notebook.uri), 'at://', ''))[1], + '' + ) as resource_did, + + -- Extract resource rkey + multiIf( + toString(record.doc.value.entry.uri) != '', + splitByChar('/', replaceOne(toString(record.doc.value.entry.uri), 'at://', ''))[3], + toString(record.doc.value.notebook.uri) != '', + splitByChar('/', replaceOne(toString(record.doc.value.notebook.uri), 'at://', ''))[3], + toString(record.doc.value.draftKey) != '', + toString(record.doc.value.draftKey), + '' + ) as resource_rkey, + + -- Extract resource collection + multiIf( + toString(record.doc.value.entry.uri) != '', 'sh.weaver.notebook.entry', + toString(record.doc.value.notebook.uri) != '', 'sh.weaver.notebook.book', + toString(record.doc.value.draftKey) != '', 'sh.weaver.edit.draft', + '' + ) as resource_collection, + + -- Root reference + splitByChar('/', replaceOne(toString(record.root.uri), 'at://', ''))[1] as root_did, + splitByChar('/', replaceOne(toString(record.root.uri), 'at://', ''))[3] as root_rkey, + + -- Prev reference (optional) + if(toString(record.prev.uri) != '', + splitByChar('/', replaceOne(toString(record.prev.uri), 'at://', ''))[1], + '') as prev_did, + if(toString(record.prev.uri) != '', + splitByChar('/', replaceOne(toString(record.prev.uri), 'at://', ''))[3], + '') as prev_rkey, + + -- Check for inline diff vs snapshot + if(length(toString(record.inlineDiff)) > 0, 1, 0) as has_inline_diff, + if(toString(record.snapshot.ref.`$link`) != '', 1, 0) as has_snapshot, + + coalesce(toDateTime64(record.createdAt, 3), event_time) as created_at, + event_time, + now64(3) as indexed_at, + if(operation = 'delete', event_time, toDateTime64(0, 3)) as deleted_at +FROM raw_records +WHERE collection = 'sh.weaver.edit.diff' diff --git a/crates/weaver-index/migrations/clickhouse/028_edit_heads.sql b/crates/weaver-index/migrations/clickhouse/028_edit_heads.sql new file mode 100644 index 0000000..706a3e0 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/028_edit_heads.sql @@ -0,0 +1,63 @@ +-- Edit heads per resource +-- Refreshable MV that tracks all branch heads for each resource +-- A head is a node with no children (nothing has prev pointing to it) +-- Multiple heads = divergent branches needing merge + +CREATE MATERIALIZED VIEW IF NOT EXISTS edit_heads +REFRESH EVERY 1 MINUTE +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (resource_did, resource_collection, resource_rkey, head_did, head_rkey) +AS +WITH + -- All nodes + all_nodes AS ( + SELECT + did, rkey, cid, collection, node_type, + resource_did, resource_collection, resource_rkey, + root_did, root_rkey, + prev_did, prev_rkey, + created_at + FROM edit_nodes + WHERE resource_did != '' + ), + -- Nodes that are pointed to by prev (have children) + has_children AS ( + SELECT DISTINCT prev_did as did, prev_rkey as rkey + FROM all_nodes + WHERE prev_did != '' + ), + -- Root CIDs lookup + root_cids AS ( + SELECT did, rkey, cid + FROM edit_nodes + WHERE node_type = 'root' + ) +-- Heads are nodes with no children +SELECT + n.resource_did, + n.resource_collection, + n.resource_rkey, + concat('at://', n.resource_did, '/', n.resource_collection, '/', n.resource_rkey) as resource_uri, + + -- This head + n.did as head_did, + n.rkey as head_rkey, + n.cid as head_cid, + n.collection as head_collection, + n.node_type as head_type, + concat('at://', n.did, '/', n.collection, '/', n.rkey) as head_uri, + + -- Root for this branch (with CID) + if(n.node_type = 'root', n.did, n.root_did) as root_did, + if(n.node_type = 'root', n.rkey, n.root_rkey) as root_rkey, + if(n.node_type = 'root', n.cid, coalesce(r.cid, '')) as root_cid, + if(n.node_type = 'root', + concat('at://', n.did, '/', n.collection, '/', n.rkey), + if(n.root_did != '', concat('at://', n.root_did, '/sh.weaver.edit.root/', n.root_rkey), '') + ) as root_uri, + + n.created_at as head_created_at, + now64(3) as indexed_at +FROM all_nodes n +LEFT ANTI JOIN has_children c ON n.did = c.did AND n.rkey = c.rkey +LEFT JOIN root_cids r ON r.did = n.root_did AND r.rkey = n.root_rkey diff --git a/crates/weaver-index/migrations/clickhouse/029_collab_invites.sql b/crates/weaver-index/migrations/clickhouse/029_collab_invites.sql new file mode 100644 index 0000000..bb8dcdd --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/029_collab_invites.sql @@ -0,0 +1,37 @@ +-- Collaboration invites +-- Half of a two-way collaboration agreement + +CREATE TABLE IF NOT EXISTS collab_invites ( + -- Invite record identity + did String, + rkey String, + cid String, + uri String MATERIALIZED concat('at://', did, '/sh.weaver.collab.invite/', rkey), + + -- Resource being shared (decomposed) + resource_did String, + resource_collection LowCardinality(String), + resource_rkey String, + resource_uri String MATERIALIZED concat('at://', resource_did, '/', resource_collection, '/', resource_rkey), + + -- Inviter is the record author (did) + inviter_did String MATERIALIZED did, + + -- Invitee + invitee_did String, + + -- Optional fields + scope LowCardinality(String) DEFAULT '', + message String DEFAULT '', + expires_at DateTime64(3) DEFAULT toDateTime64(0, 3), + + -- Timestamps + created_at DateTime64(3), + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Soft delete (epoch = not deleted) + deleted_at DateTime64(3) DEFAULT toDateTime64(0, 3) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (did, rkey) diff --git a/crates/weaver-index/migrations/clickhouse/030_collab_invites_mv.sql b/crates/weaver-index/migrations/clickhouse/030_collab_invites_mv.sql new file mode 100644 index 0000000..e9fc7cd --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/030_collab_invites_mv.sql @@ -0,0 +1,23 @@ +-- Populate collab_invites from raw_records + +CREATE MATERIALIZED VIEW IF NOT EXISTS collab_invites_mv TO collab_invites AS +SELECT + did, + rkey, + cid, + + -- Parse resource strongRef + splitByChar('/', replaceOne(toString(record.resource.uri), 'at://', ''))[1] as resource_did, + splitByChar('/', replaceOne(toString(record.resource.uri), 'at://', ''))[2] as resource_collection, + splitByChar('/', replaceOne(toString(record.resource.uri), 'at://', ''))[3] as resource_rkey, + + toString(record.invitee) as invitee_did, + coalesce(toString(record.scope), '') as scope, + coalesce(toString(record.message), '') as message, + coalesce(toDateTime64(record.expiresAt, 3), toDateTime64(0, 3)) as expires_at, + coalesce(toDateTime64(record.createdAt, 3), event_time) as created_at, + event_time, + now64(3) as indexed_at, + if(operation = 'delete', event_time, toDateTime64(0, 3)) as deleted_at +FROM raw_records +WHERE collection = 'sh.weaver.collab.invite' diff --git a/crates/weaver-index/migrations/clickhouse/031_collab_accepts.sql b/crates/weaver-index/migrations/clickhouse/031_collab_accepts.sql new file mode 100644 index 0000000..ff0702a --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/031_collab_accepts.sql @@ -0,0 +1,35 @@ +-- Collaboration accepts +-- Completes the two-way collaboration agreement + +CREATE TABLE IF NOT EXISTS collab_accepts ( + -- Accept record identity + did String, + rkey String, + cid String, + uri String MATERIALIZED concat('at://', did, '/sh.weaver.collab.accept/', rkey), + + -- Invite being accepted (decomposed) + invite_did String, + invite_rkey String, + invite_cid String DEFAULT '', + invite_uri String MATERIALIZED concat('at://', invite_did, '/sh.weaver.collab.invite/', invite_rkey), + + -- Resource (denormalized in the record) + resource_did String, + resource_collection LowCardinality(String), + resource_rkey String, + resource_uri String MATERIALIZED concat('at://', resource_did, '/', resource_collection, '/', resource_rkey), + + -- Accepter is the record author (did) + accepter_did String MATERIALIZED did, + + -- Timestamps + created_at DateTime64(3), + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Soft delete (epoch = not deleted) + deleted_at DateTime64(3) DEFAULT toDateTime64(0, 3) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (did, rkey) diff --git a/crates/weaver-index/migrations/clickhouse/032_collab_accepts_mv.sql b/crates/weaver-index/migrations/clickhouse/032_collab_accepts_mv.sql new file mode 100644 index 0000000..076227d --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/032_collab_accepts_mv.sql @@ -0,0 +1,24 @@ +-- Populate collab_accepts from raw_records + +CREATE MATERIALIZED VIEW IF NOT EXISTS collab_accepts_mv TO collab_accepts AS +SELECT + did, + rkey, + cid, + + -- Parse invite strongRef + splitByChar('/', replaceOne(toString(record.invite.uri), 'at://', ''))[1] as invite_did, + splitByChar('/', replaceOne(toString(record.invite.uri), 'at://', ''))[3] as invite_rkey, + toString(record.invite.cid) as invite_cid, + + -- Parse resource AT-URI + splitByChar('/', replaceOne(toString(record.resource), 'at://', ''))[1] as resource_did, + splitByChar('/', replaceOne(toString(record.resource), 'at://', ''))[2] as resource_collection, + splitByChar('/', replaceOne(toString(record.resource), 'at://', ''))[3] as resource_rkey, + + coalesce(toDateTime64(record.createdAt, 3), event_time) as created_at, + event_time, + now64(3) as indexed_at, + if(operation = 'delete', event_time, toDateTime64(0, 3)) as deleted_at +FROM raw_records +WHERE collection = 'sh.weaver.collab.accept' diff --git a/crates/weaver-index/migrations/clickhouse/033_collab_sessions.sql b/crates/weaver-index/migrations/clickhouse/033_collab_sessions.sql new file mode 100644 index 0000000..c32c29f --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/033_collab_sessions.sql @@ -0,0 +1,34 @@ +-- Active collaboration sessions +-- Published when joining collaborative editing, deleted on disconnect + +CREATE TABLE IF NOT EXISTS collab_sessions ( + -- Session record identity + did String, + rkey String, + cid String, + uri String MATERIALIZED concat('at://', did, '/sh.weaver.collab.session/', rkey), + + -- Resource being edited (decomposed) + resource_did String, + resource_collection LowCardinality(String), + resource_rkey String, + resource_uri String MATERIALIZED concat('at://', resource_did, '/', resource_collection, '/', resource_rkey), + + -- Participant is the record author (did) + participant_did String MATERIALIZED did, + + -- Connection info + node_id String, + relay_url String DEFAULT '', + + -- Timestamps + created_at DateTime64(3), + expires_at DateTime64(3) DEFAULT toDateTime64(0, 3), + event_time DateTime64(3), + indexed_at DateTime64(3) DEFAULT now64(3), + + -- Soft delete (epoch = not deleted) + deleted_at DateTime64(3) DEFAULT toDateTime64(0, 3) +) +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (did, rkey) diff --git a/crates/weaver-index/migrations/clickhouse/034_collab_sessions_mv.sql b/crates/weaver-index/migrations/clickhouse/034_collab_sessions_mv.sql new file mode 100644 index 0000000..7ac0c3c --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/034_collab_sessions_mv.sql @@ -0,0 +1,22 @@ +-- Populate collab_sessions from raw_records + +CREATE MATERIALIZED VIEW IF NOT EXISTS collab_sessions_mv TO collab_sessions AS +SELECT + did, + rkey, + cid, + + -- Parse resource strongRef + splitByChar('/', replaceOne(toString(record.resource.uri), 'at://', ''))[1] as resource_did, + splitByChar('/', replaceOne(toString(record.resource.uri), 'at://', ''))[2] as resource_collection, + splitByChar('/', replaceOne(toString(record.resource.uri), 'at://', ''))[3] as resource_rkey, + + toString(record.nodeId) as node_id, + coalesce(toString(record.relayUrl), '') as relay_url, + coalesce(toDateTime64(record.createdAt, 3), event_time) as created_at, + coalesce(toDateTime64(record.expiresAt, 3), toDateTime64(0, 3)) as expires_at, + event_time, + now64(3) as indexed_at, + if(operation = 'delete', event_time, toDateTime64(0, 3)) as deleted_at +FROM raw_records +WHERE collection = 'sh.weaver.collab.session' diff --git a/crates/weaver-index/migrations/clickhouse/035_collaborators.sql b/crates/weaver-index/migrations/clickhouse/035_collaborators.sql new file mode 100644 index 0000000..dad1f72 --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/035_collaborators.sql @@ -0,0 +1,47 @@ +-- Valid collaborators (matched invite + accept pairs) +-- Refreshable MV that joins invites and accepts + +CREATE MATERIALIZED VIEW IF NOT EXISTS collaborators +REFRESH EVERY 1 MINUTE +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (resource_did, resource_collection, resource_rkey, collaborator_did) +AS +SELECT + -- Resource + i.resource_did, + i.resource_collection, + i.resource_rkey, + concat('at://', i.resource_did, '/', i.resource_collection, '/', i.resource_rkey) as resource_uri, + + -- Collaborator (the invitee who accepted) + i.invitee_did as collaborator_did, + + -- Inviter + i.did as inviter_did, + + -- Invite record + i.did as invite_did, + i.rkey as invite_rkey, + i.cid as invite_cid, + concat('at://', i.did, '/sh.weaver.collab.invite/', i.rkey) as invite_uri, + + -- Accept record + a.did as accept_did, + a.rkey as accept_rkey, + a.cid as accept_cid, + concat('at://', a.did, '/sh.weaver.collab.accept/', a.rkey) as accept_uri, + + -- Scope + i.scope, + + -- Timestamps + i.created_at as invited_at, + a.created_at as accepted_at, + now64(3) as indexed_at +FROM collab_invites i +INNER JOIN collab_accepts a ON + a.invite_did = i.did + AND a.invite_rkey = i.rkey +WHERE + -- Invite not expired + (i.expires_at = toDateTime64(0, 3) OR i.expires_at > now64(3)) diff --git a/crates/weaver-index/migrations/clickhouse/036_permissions.sql b/crates/weaver-index/migrations/clickhouse/036_permissions.sql new file mode 100644 index 0000000..3b3751b --- /dev/null +++ b/crates/weaver-index/migrations/clickhouse/036_permissions.sql @@ -0,0 +1,66 @@ +-- Resource permissions +-- Refreshable MV that computes who can access each resource +-- Combines: owners (resource creator) + collaborators (invite+accept pairs) + +CREATE MATERIALIZED VIEW IF NOT EXISTS permissions +REFRESH EVERY 1 MINUTE +ENGINE = ReplacingMergeTree(indexed_at) +ORDER BY (resource_did, resource_collection, resource_rkey, grantee_did) +AS +-- Owners: resource creator has owner permission +SELECT + did as resource_did, + 'sh.weaver.notebook.entry' as resource_collection, + rkey as resource_rkey, + concat('at://', did, '/sh.weaver.notebook.entry/', rkey) as resource_uri, + + did as grantee_did, + 'owner' as scope, + + -- Source is the resource itself + did as source_did, + 'sh.weaver.notebook.entry' as source_collection, + rkey as source_rkey, + + created_at as granted_at, + now64(3) as indexed_at +FROM entries + +UNION ALL + +SELECT + did as resource_did, + 'sh.weaver.notebook.book' as resource_collection, + rkey as resource_rkey, + concat('at://', did, '/sh.weaver.notebook.book/', rkey) as resource_uri, + + did as grantee_did, + 'owner' as scope, + + did as source_did, + 'sh.weaver.notebook.book' as source_collection, + rkey as source_rkey, + + created_at as granted_at, + now64(3) as indexed_at +FROM notebooks + +UNION ALL + +-- Collaborators: invite+accept pairs grant permission +SELECT + resource_did, + resource_collection, + resource_rkey, + resource_uri, + + collaborator_did as grantee_did, + if(scope != '', scope, 'collaborator') as scope, + + invite_did as source_did, + 'sh.weaver.collab.invite' as source_collection, + invite_rkey as source_rkey, + + accepted_at as granted_at, + indexed_at +FROM collaborators diff --git a/crates/weaver-index/src/bin/weaver_indexer.rs b/crates/weaver-index/src/bin/weaver_indexer.rs index 3c12d82..5a62249 100644 --- a/crates/weaver-index/src/bin/weaver_indexer.rs +++ b/crates/weaver-index/src/bin/weaver_indexer.rs @@ -1,6 +1,6 @@ use clap::{Parser, Subcommand}; use tracing::{error, info, warn}; -use weaver_index::clickhouse::{Client, Migrator, Tables}; +use weaver_index::clickhouse::{Client, Migrator}; use weaver_index::config::{ ClickHouseConfig, FirehoseConfig, IndexerConfig, ShardConfig, SourceMode, TapConfig, }; @@ -71,18 +71,19 @@ async fn run_migrate(dry_run: bool, reset: bool) -> miette::Result<()> { let client = Client::new(&config)?; if reset { + let objects = Migrator::all_objects(); if dry_run { - info!("Would drop tables:"); - for table in Tables::ALL { - info!(" - {}", table); + info!("Would drop {} objects:", objects.len()); + for obj in &objects { + info!(" - {} ({:?})", obj.name, obj.object_type); } } else { - info!("Dropping all tables..."); - for table in Tables::ALL { - let query = format!("DROP TABLE IF EXISTS {}", table); + info!("Dropping all tables and views..."); + for obj in &objects { + let query = obj.drop_statement(); match client.execute(&query).await { - Ok(_) => info!(" dropped {}", table), - Err(e) => warn!(" failed to drop {}: {}", table, e), + Ok(_) => info!(" dropped {} ({:?})", obj.name, obj.object_type), + Err(e) => warn!(" failed to drop {}: {}", obj.name, e), } } } diff --git a/crates/weaver-index/src/clickhouse.rs b/crates/weaver-index/src/clickhouse.rs index 91f2e07..e307fba 100644 --- a/crates/weaver-index/src/clickhouse.rs +++ b/crates/weaver-index/src/clickhouse.rs @@ -3,7 +3,7 @@ mod migrations; mod schema; pub use client::{Client, TableSize}; -pub use migrations::{MigrationResult, Migrator}; +pub use migrations::{DbObject, MigrationResult, Migrator, ObjectType}; pub use schema::{ 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 5a3655a..c476be4 100644 --- a/crates/weaver-index/src/clickhouse/client.rs +++ b/crates/weaver-index/src/clickhouse/client.rs @@ -106,12 +106,11 @@ impl Client { collection: &str, rkey: &str, ) -> Result, IndexError> { - // FINAL ensures ReplacingMergeTree deduplication is applied // Order by event_time first (firehose data wins), then indexed_at as tiebreaker // Include deletes so we can return not-found for deleted records let query = r#" SELECT cid, record, operation - FROM raw_records FINAL + FROM raw_records WHERE did = ? AND collection = ? AND rkey = ? @@ -191,6 +190,7 @@ impl Client { /// List records for a repo+collection /// /// Returns non-deleted records ordered by rkey, with cursor-based pagination. + /// Uses window function to get latest operation per rkey and filter out deletes. pub async fn list_records( &self, did: &str, @@ -202,16 +202,20 @@ impl Client { let order = if reverse { "DESC" } else { "ASC" }; let cursor_op = if reverse { "<" } else { ">" }; - // Build query with optional cursor + // Use window function to get latest version per rkey, then filter out deletes 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' + FROM ( + SELECT rkey, cid, record, operation, + ROW_NUMBER() OVER (PARTITION BY rkey ORDER BY event_time DESC, indexed_at DESC) as rn + FROM raw_records + WHERE did = ? + AND collection = ? + AND rkey {cursor_op} ? + ) + WHERE rn = 1 AND operation != 'delete' ORDER BY rkey {order} LIMIT ? "#, @@ -220,10 +224,14 @@ impl Client { format!( r#" SELECT rkey, cid, record - FROM raw_records FINAL - WHERE did = ? - AND collection = ? - AND operation != 'delete' + FROM ( + SELECT rkey, cid, record, operation, + ROW_NUMBER() OVER (PARTITION BY rkey ORDER BY event_time DESC, indexed_at DESC) as rn + FROM raw_records + WHERE did = ? + AND collection = ? + ) + WHERE rn = 1 AND operation != 'delete' ORDER BY rkey {order} LIMIT ? "#, diff --git a/crates/weaver-index/src/clickhouse/migrations.rs b/crates/weaver-index/src/clickhouse/migrations.rs index 9de7bbd..64b2a7a 100644 --- a/crates/weaver-index/src/clickhouse/migrations.rs +++ b/crates/weaver-index/src/clickhouse/migrations.rs @@ -1,5 +1,6 @@ use crate::error::{ClickHouseError, IndexError}; use include_dir::{Dir, include_dir}; +use regex::Regex; use tracing::info; use super::Client; @@ -7,6 +8,33 @@ use super::Client; /// Embedded migrations directory - compiled into the binary static MIGRATIONS_DIR: Dir = include_dir!("$CARGO_MANIFEST_DIR/migrations/clickhouse"); +/// Type of database object +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ObjectType { + Table, + MaterializedView, + View, +} + +/// A database object (table or view) extracted from migrations +#[derive(Debug, Clone)] +pub struct DbObject { + pub name: String, + pub object_type: ObjectType, +} + +impl DbObject { + /// Get the DROP statement for this object + pub fn drop_statement(&self) -> String { + match self.object_type { + ObjectType::Table => format!("DROP TABLE IF EXISTS {}", self.name), + ObjectType::MaterializedView | ObjectType::View => { + format!("DROP VIEW IF EXISTS {}", self.name) + } + } + } +} + /// Migration runner for ClickHouse pub struct Migrator<'a> { client: &'a Client, @@ -32,6 +60,51 @@ impl<'a> Migrator<'a> { files } + /// Extract all database objects (tables, views) from migrations + /// Returns them in reverse order for safe dropping (MVs before their source tables) + pub fn all_objects() -> Vec { + let table_re = + Regex::new(r"(?i)CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+(\w+)").unwrap(); + let mv_re = + Regex::new(r"(?i)CREATE\s+MATERIALIZED\s+VIEW\s+IF\s+NOT\s+EXISTS\s+(\w+)").unwrap(); + let view_re = + Regex::new(r"(?i)CREATE\s+VIEW\s+IF\s+NOT\s+EXISTS\s+(\w+)").unwrap(); + + let mut objects = Vec::new(); + + for (_, sql) in Self::migrations() { + // Find all materialized views + for caps in mv_re.captures_iter(sql) { + objects.push(DbObject { + name: caps[1].to_string(), + object_type: ObjectType::MaterializedView, + }); + } + // Find all regular views (excluding MVs already matched) + for caps in view_re.captures_iter(sql) { + let name = caps[1].to_string(); + // Skip if already added as MV + if !objects.iter().any(|o| o.name == name) { + objects.push(DbObject { + name, + object_type: ObjectType::View, + }); + } + } + // Find all tables + for caps in table_re.captures_iter(sql) { + objects.push(DbObject { + name: caps[1].to_string(), + object_type: ObjectType::Table, + }); + } + } + + // Reverse so MVs/views come before their source tables + objects.reverse(); + objects + } + /// Run all pending migrations pub async fn run(&self) -> Result { // First, ensure the migrations table exists (bootstrap) @@ -57,7 +130,12 @@ impl<'a> Migrator<'a> { } info!(migration = %name, "applying migration"); - self.client.execute(sql).await?; + + // Split by semicolons and execute each statement + for statement in Self::split_statements(sql) { + self.client.execute(statement).await?; + } + self.record_migration(name).await?; applied_count += 1; } @@ -68,6 +146,14 @@ impl<'a> Migrator<'a> { }) } + /// Split SQL into individual statements + fn split_statements(sql: &str) -> Vec<&str> { + sql.split(';') + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + .collect() + } + /// Check which migrations would be applied without running them pub async fn pending(&self) -> Result, IndexError> { // Try to get applied migrations, but if table doesn't exist, all are pending diff --git a/crates/weaver-index/src/error.rs b/crates/weaver-index/src/error.rs index ba3e6b0..587e5cc 100644 --- a/crates/weaver-index/src/error.rs +++ b/crates/weaver-index/src/error.rs @@ -93,7 +93,7 @@ pub enum SqliteError { /// ClickHouse database errors #[derive(Debug, Error, Diagnostic)] pub enum ClickHouseError { - #[error("failed to connect to ClickHouse: {message}")] + #[error("failed to connect to ClickHouse: {message}: {source}")] #[diagnostic(code(clickhouse::connection))] Connection { message: String, @@ -101,7 +101,7 @@ pub enum ClickHouseError { source: clickhouse::error::Error, }, - #[error("ClickHouse query failed: {message}")] + #[error("ClickHouse query failed: {message}: {source}")] #[diagnostic(code(clickhouse::query))] Query { message: String, @@ -109,7 +109,7 @@ pub enum ClickHouseError { source: clickhouse::error::Error, }, - #[error("failed to insert batch: {message}")] + #[error("failed to insert batch: {message}: {source}")] #[diagnostic(code(clickhouse::insert))] Insert { message: String,