denormalized tables + view tables

Author: orual

Cargo.lock

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

crates/weaver-index/Cargo.toml

@@ -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);

crates/weaver-index/migrations/clickhouse/001_raw_records.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';

crates/weaver-index/migrations/clickhouse/008_handle_mappings.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 != ''

crates/weaver-index/migrations/clickhouse/009_handle_mappings_identity_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 != ''

crates/weaver-index/migrations/clickhouse/010_handle_mappings_account_mv.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

crates/weaver-index/migrations/clickhouse/011_profiles_weaver.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'

crates/weaver-index/migrations/clickhouse/012_profiles_weaver_mv.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

crates/weaver-index/migrations/clickhouse/013_profiles_bsky.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'