+316 -21

crates/server/src/firehose.rs

··· 6 6 use crate::db::DbPool; 7 7 use async_trait::async_trait; 8 8 use atproto_jetstream::{Consumer, ConsumerTaskConfig, EventHandler, JetstreamEvent}; 9 9 + use chrono::{DateTime, Utc}; 10 10 + use serde::Deserialize; 11 11 + use serde_json::Value; 9 12 use tokio_util::sync::CancellationToken; 10 13 11 14 /// Default Jetstream endpoint (Bluesky's public instance). ··· 14 17 /// Collections we're interested in indexing. 15 18 pub const MALFESTIO_COLLECTIONS: &[&str] = &["app.malfestio.deck", "app.malfestio.card", "app.malfestio.note"]; 16 19 20 20 + /// Deck record structure matching the Lexicon schema. 21 21 + #[derive(Debug, Deserialize)] 22 22 + #[serde(rename_all = "camelCase")] 23 23 + pub struct DeckRecord { 24 24 + pub title: String, 25 25 + #[serde(default)] 26 26 + pub description: Option<String>, 27 27 + #[serde(default)] 28 28 + pub tags: Vec<String>, 29 29 + #[serde(default)] 30 30 + pub card_refs: Vec<String>, 31 31 + #[serde(default)] 32 32 + pub source_refs: Vec<String>, 33 33 + #[serde(default)] 34 34 + pub license: Option<String>, 35 35 + pub created_at: String, 36 36 + } 37 37 + 38 38 + /// Card record structure matching the Lexicon schema. 39 39 + #[derive(Debug, Deserialize)] 40 40 + #[serde(rename_all = "camelCase")] 41 41 + pub struct CardRecord { 42 42 + pub deck_ref: String, 43 43 + pub front: String, 44 44 + pub back: String, 45 45 + #[serde(default)] 46 46 + pub card_type: Option<String>, 47 47 + #[serde(default)] 48 48 + pub hints: Vec<String>, 49 49 + pub created_at: String, 50 50 + } 51 51 + 52 52 + /// Note record structure matching the Lexicon schema. 53 53 + #[derive(Debug, Deserialize)] 54 54 + #[serde(rename_all = "camelCase")] 55 55 + pub struct NoteRecord { 56 56 + pub title: String, 57 57 + pub body: String, 58 58 + #[serde(default)] 59 59 + pub tags: Vec<String>, 60 60 + #[serde(default)] 61 61 + pub visibility: Option<String>, 62 62 + pub created_at: String, 63 63 + } 64 64 + 65 65 + /// Parse a datetime string from record into chrono DateTime. 66 66 + fn parse_record_datetime(dt_str: &str) -> DateTime<Utc> { 67 67 + DateTime::parse_from_rfc3339(dt_str) 68 68 + .map(|dt| dt.with_timezone(&Utc)) 69 69 + .unwrap_or_else(|_| Utc::now()) 70 70 + } 71 71 + 17 72 /// Event handler for Malfestio records from Jetstream. 18 73 pub struct MalfestioEventHandler { 19 74 pool: DbPool, ··· 26 81 Self { pool, handler_id: "malfestio-indexer".to_string() } 27 82 } 28 83 29 29 - /// Index a record into the database. 30 30 - async fn index_record( 84 84 + /// Index a deck record with full content. 85 85 + async fn index_deck( 86 86 + &self, did: &str, rkey: &str, rev: &str, record: &Value, 87 87 + ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { 88 88 + let at_uri = format!("at://{}/app.malfestio.deck/{}", did, rkey); 89 89 + let deck: DeckRecord = serde_json::from_value(record.clone())?; 90 90 + let created_at = parse_record_datetime(&deck.created_at); 91 91 + 92 92 + let client = self.pool.get().await?; 93 93 + client 94 94 + .execute( 95 95 + "INSERT INTO indexed_decks (at_uri, did, rkey, title, description, tags, card_refs, source_refs, license, record_created_at, indexed_at) 96 96 + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, NOW()) 97 97 + ON CONFLICT (at_uri) DO UPDATE SET 98 98 + title = $4, description = $5, tags = $6, card_refs = $7, source_refs = $8, 99 99 + license = $9, record_created_at = $10, indexed_at = NOW(), deleted_at = NULL", 100 100 + &[ 101 101 + &at_uri, 102 102 + &did, 103 103 + &rkey, 104 104 + &deck.title, 105 105 + &deck.description, 106 106 + &deck.tags, 107 107 + &deck.card_refs, 108 108 + &deck.source_refs, 109 109 + &deck.license, 110 110 + &created_at, 111 111 + ], 112 112 + ) 113 113 + .await?; 114 114 + 115 115 + self.update_repo_state(did, rev).await?; 116 116 + 117 117 + tracing::debug!("Indexed deck: {}", at_uri); 118 118 + Ok(()) 119 119 + } 120 120 + 121 121 + /// Index a card record with full content. 122 122 + async fn index_card( 123 123 + &self, did: &str, rkey: &str, rev: &str, record: &Value, 124 124 + ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { 125 125 + let at_uri = format!("at://{}/app.malfestio.card/{}", did, rkey); 126 126 + let card: CardRecord = serde_json::from_value(record.clone())?; 127 127 + let created_at = parse_record_datetime(&card.created_at); 128 128 + let card_type = card.card_type.unwrap_or_else(|| "basic".to_string()); 129 129 + 130 130 + let client = self.pool.get().await?; 131 131 + 132 132 + client 133 133 + .execute( 134 134 + "INSERT INTO indexed_cards (at_uri, did, rkey, deck_ref, front, back, card_type, hints, record_created_at, indexed_at) 135 135 + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NOW()) 136 136 + ON CONFLICT (at_uri) DO UPDATE SET 137 137 + deck_ref = $4, front = $5, back = $6, card_type = $7, hints = $8, 138 138 + record_created_at = $9, indexed_at = NOW(), deleted_at = NULL", 139 139 + &[ 140 140 + &at_uri, 141 141 + &did, 142 142 + &rkey, 143 143 + &card.deck_ref, 144 144 + &card.front, 145 145 + &card.back, 146 146 + &card_type, 147 147 + &card.hints, 148 148 + &created_at, 149 149 + ], 150 150 + ) 151 151 + .await?; 152 152 + 153 153 + self.update_repo_state(did, rev).await?; 154 154 + 155 155 + tracing::debug!("Indexed card: {}", at_uri); 156 156 + Ok(()) 157 157 + } 158 158 + 159 159 + /// Index a note record with full content. 160 160 + async fn index_note( 161 161 + &self, did: &str, rkey: &str, rev: &str, record: &Value, 162 162 + ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { 163 163 + let at_uri = format!("at://{}/app.malfestio.note/{}", did, rkey); 164 164 + let note: NoteRecord = serde_json::from_value(record.clone())?; 165 165 + let created_at = parse_record_datetime(&note.created_at); 166 166 + let visibility = note.visibility.unwrap_or_else(|| "public".to_string()); 167 167 + 168 168 + let client = self.pool.get().await?; 169 169 + 170 170 + client 171 171 + .execute( 172 172 + "INSERT INTO indexed_notes (at_uri, did, rkey, title, body, tags, visibility, record_created_at, indexed_at) 173 173 + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW()) 174 174 + ON CONFLICT (at_uri) DO UPDATE SET 175 175 + title = $4, body = $5, tags = $6, visibility = $7, 176 176 + record_created_at = $8, indexed_at = NOW(), deleted_at = NULL", 177 177 + &[ 178 178 + &at_uri, 179 179 + &did, 180 180 + &rkey, 181 181 + &note.title, 182 182 + &note.body, 183 183 + &note.tags, 184 184 + &visibility, 185 185 + &created_at, 186 186 + ], 187 187 + ) 188 188 + .await?; 189 189 + 190 190 + self.update_repo_state(did, rev).await?; 191 191 + 192 192 + tracing::debug!("Indexed note: {}", at_uri); 193 193 + Ok(()) 194 194 + } 195 195 + 196 196 + /// Handle deletion of a record (soft delete by setting deleted_at). 197 197 + async fn handle_delete( 31 198 &self, did: &str, collection: &str, rkey: &str, 32 199 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { 33 200 let at_uri = format!("at://{}/{}/{}", did, collection, rkey); 201 201 + let client = self.pool.get().await?; 34 202 35 35 - let client = self.pool.get().await?; 203 203 + let table = match collection { 204 204 + "app.malfestio.deck" => "indexed_decks", 205 205 + "app.malfestio.card" => "indexed_cards", 206 206 + "app.malfestio.note" => "indexed_notes", 207 207 + _ => return Ok(()), 208 208 + }; 209 209 + 210 210 + let query = format!( 211 211 + "UPDATE {} SET deleted_at = NOW() WHERE at_uri = $1 AND deleted_at IS NULL", 212 212 + table 213 213 + ); 214 214 + client.execute(&query, &[&at_uri]).await?; 36 215 37 37 - // Upsert into indexed_records table 216 216 + tracing::info!("Soft-deleted record: {}", at_uri); 217 217 + Ok(()) 218 218 + } 219 219 + 220 220 + /// Update the repo sync state with the latest processed revision. 221 221 + async fn update_repo_state(&self, did: &str, rev: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { 222 222 + let client = self.pool.get().await?; 38 223 client 39 224 .execute( 40 40 - "INSERT INTO indexed_records (at_uri, did, collection, rkey, indexed_at) 41 41 - VALUES ($1, $2, $3, $4, NOW()) 42 42 - ON CONFLICT (at_uri) DO UPDATE SET indexed_at = NOW()", 43 43 - &[&at_uri, &did, &collection, &rkey], 225 225 + "INSERT INTO repo_sync_state (did, latest_rev, indexed_at) 226 226 + VALUES ($1, $2, NOW()) 227 227 + ON CONFLICT (did) DO UPDATE SET latest_rev = $2, indexed_at = NOW() 228 228 + WHERE repo_sync_state.latest_rev < $2 OR repo_sync_state.latest_rev IS NULL", 229 229 + &[&did, &rev], 44 230 ) 45 231 .await?; 46 46 - 47 47 - tracing::debug!("Indexed record: {}", at_uri); 48 232 Ok(()) 49 233 } 50 234 51 235 /// Update cursor position in database for reconnection. 52 52 - #[allow(dead_code)] 53 236 pub async fn save_cursor(&self, cursor_us: i64) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { 54 237 let client = self.pool.get().await?; 55 238 client ··· 85 268 86 269 async fn handle_event(&self, event: JetstreamEvent) -> Result<(), anyhow::Error> { 87 270 match event { 88 88 - JetstreamEvent::Commit { did, commit, .. } => { 271 271 + JetstreamEvent::Commit { did, time_us, commit, .. } => { 89 272 let collection = &commit.collection; 90 90 - 91 91 - // Only process our collections 92 273 if !MALFESTIO_COLLECTIONS.iter().any(|c| collection == *c) { 93 274 return Ok(()); 94 275 } 95 276 96 277 let rkey = &commit.rkey; 278 278 + let rev = &commit.rev; 279 279 + let operation = &commit.operation; 97 280 98 98 - tracing::info!("Received {} event: did={}, rkey={}", collection, did, rkey); 281 281 + tracing::info!( 282 282 + "Received {} {} event: did={}, rkey={}", 283 283 + collection, 284 284 + operation, 285 285 + did, 286 286 + rkey 287 287 + ); 288 288 + 289 289 + match operation.as_str() { 290 290 + "create" | "update" => { 291 291 + let result = match collection.as_str() { 292 292 + "app.malfestio.deck" => self.index_deck(&did, rkey, rev, &commit.record).await, 293 293 + "app.malfestio.card" => self.index_card(&did, rkey, rev, &commit.record).await, 294 294 + "app.malfestio.note" => self.index_note(&did, rkey, rev, &commit.record).await, 295 295 + _ => Ok(()), 296 296 + }; 99 297 100 100 - // Index the record 101 101 - if let Err(e) = self.index_record(&did, collection, rkey).await { 102 102 - tracing::warn!("Failed to index record: {}", e); 298 298 + if let Err(e) = result { 299 299 + tracing::warn!("Failed to index record: {}", e); 300 300 + } 301 301 + } 302 302 + "delete" => { 303 303 + if let Err(e) = self.handle_delete(&did, collection, rkey).await { 304 304 + tracing::warn!("Failed to handle delete: {}", e); 305 305 + } 306 306 + } 307 307 + _ => tracing::debug!("Unknown operation type: {}", operation), 308 308 + } 309 309 + 310 310 + if let Err(e) = self.save_cursor(time_us as i64).await { 311 311 + tracing::warn!("Failed to save cursor: {}", e); 103 312 } 104 313 } 105 105 - JetstreamEvent::Identity { .. } | JetstreamEvent::Account { .. } | JetstreamEvent::Delete { .. } => { 106 106 - // Ignore identity, account, and delete events 314 314 + JetstreamEvent::Delete { did, commit, .. } => { 315 315 + let collection = &commit.collection; 316 316 + 317 317 + if MALFESTIO_COLLECTIONS.iter().any(|c| collection == *c) { 318 318 + let rkey = &commit.rkey; 319 319 + tracing::info!( 320 320 + "Received delete event: did={}, collection={}, rkey={}", 321 321 + did, 322 322 + collection, 323 323 + rkey 324 324 + ); 325 325 + 326 326 + if let Err(e) = self.handle_delete(&did, collection, rkey).await { 327 327 + tracing::warn!("Failed to handle delete: {}", e); 328 328 + } 329 329 + } 107 330 } 331 331 + JetstreamEvent::Identity { .. } | JetstreamEvent::Account { .. } => (), 108 332 } 109 333 Ok(()) 110 334 } ··· 139 363 140 364 let handler = MalfestioEventHandler::new(pool); 141 365 142 142 - // Build consumer config 143 366 let task_config = ConsumerTaskConfig { 144 367 user_agent: "malfestio-indexer/0.1.0".to_string(), 145 368 compression: config.compress, ··· 177 400 178 401 #[cfg(test)] 179 402 mod tests { 403 403 + use chrono::Datelike; 404 404 + 180 405 use super::*; 181 406 182 407 #[test] ··· 192 417 assert!(MALFESTIO_COLLECTIONS.contains(&"app.malfestio.deck")); 193 418 assert!(MALFESTIO_COLLECTIONS.contains(&"app.malfestio.card")); 194 419 assert!(MALFESTIO_COLLECTIONS.contains(&"app.malfestio.note")); 420 420 + } 421 421 + 422 422 + #[test] 423 423 + fn test_parse_deck_record() { 424 424 + let json = serde_json::json!({ 425 425 + "title": "Test Deck", 426 426 + "description": "A test deck", 427 427 + "tags": ["rust", "learning"], 428 428 + "cardRefs": ["at://did:plc:abc/app.malfestio.card/123"], 429 429 + "sourceRefs": [], 430 430 + "license": "CC-BY-4.0", 431 431 + "createdAt": "2024-01-01T00:00:00Z" 432 432 + }); 433 433 + 434 434 + let deck: DeckRecord = serde_json::from_value(json).unwrap(); 435 435 + assert_eq!(deck.title, "Test Deck"); 436 436 + assert_eq!(deck.description, Some("A test deck".to_string())); 437 437 + assert_eq!(deck.tags, vec!["rust", "learning"]); 438 438 + assert_eq!(deck.card_refs.len(), 1); 439 439 + assert_eq!(deck.license, Some("CC-BY-4.0".to_string())); 440 440 + } 441 441 + 442 442 + #[test] 443 443 + fn test_parse_card_record() { 444 444 + let json = serde_json::json!({ 445 445 + "deckRef": "at://did:plc:abc/app.malfestio.deck/123", 446 446 + "front": "What is Rust?", 447 447 + "back": "A systems programming language", 448 448 + "cardType": "basic", 449 449 + "hints": ["Think about memory safety"], 450 450 + "createdAt": "2024-01-01T00:00:00Z" 451 451 + }); 452 452 + 453 453 + let card: CardRecord = serde_json::from_value(json).unwrap(); 454 454 + assert_eq!(card.deck_ref, "at://did:plc:abc/app.malfestio.deck/123"); 455 455 + assert_eq!(card.front, "What is Rust?"); 456 456 + assert_eq!(card.back, "A systems programming language"); 457 457 + assert_eq!(card.card_type, Some("basic".to_string())); 458 458 + assert_eq!(card.hints, vec!["Think about memory safety"]); 459 459 + } 460 460 + 461 461 + #[test] 462 462 + fn test_parse_note_record() { 463 463 + let json = serde_json::json!({ 464 464 + "title": "Study Notes", 465 465 + "body": "# Chapter 1\n\nSome content here.", 466 466 + "tags": ["chapter1"], 467 467 + "visibility": "public", 468 468 + "createdAt": "2024-01-01T00:00:00Z" 469 469 + }); 470 470 + 471 471 + let note: NoteRecord = serde_json::from_value(json).unwrap(); 472 472 + assert_eq!(note.title, "Study Notes"); 473 473 + assert_eq!(note.body, "# Chapter 1\n\nSome content here."); 474 474 + assert_eq!(note.tags, vec!["chapter1"]); 475 475 + assert_eq!(note.visibility, Some("public".to_string())); 476 476 + } 477 477 + 478 478 + #[test] 479 479 + fn test_parse_record_datetime() { 480 480 + let dt = parse_record_datetime("2024-01-15T10:30:00Z"); 481 481 + assert_eq!(dt.year(), 2024); 482 482 + assert_eq!(dt.month(), 1); 483 483 + assert_eq!(dt.day(), 15); 484 484 + } 485 485 + 486 486 + #[test] 487 487 + fn test_parse_record_datetime_invalid() { 488 488 + let dt = parse_record_datetime("invalid"); 489 489 + assert!(dt.year() >= 2024); 195 490 } 196 491 }

+60 -8

docs/at-notes.md

··· 48 48 49 49 ## Firehose / Jetstream 50 50 51 51 + ### Overview 52 52 + 53 53 + The AT Protocol provides two main options for consuming real-time repository events: 54 54 + 55 55 + 1. **Raw Firehose** (`com.atproto.sync.subscribeRepos`) - Full-fidelity, CBOR-encoded, cryptographically signed 56 56 + 2. **Jetstream** - Simplified JSON format, lower bandwidth, easier to consume 57 57 + 51 58 ### Raw Firehose 52 59 53 60 - **WebSocket**: Subscribe to `com.atproto.sync.subscribeRepos` from a Relay 54 54 - - **CBOR Decoding**: Parse incoming events 55 55 - - **Cursor Management**: Track position for reconnection 61 61 + - **CBOR Decoding**: Parse CAR files containing MST blocks 62 62 + - **Cryptographic Verification**: Validate commit signatures against DID signing keys 63 63 + - **Cursor Management**: Track `seq` position for reliable reconnection 56 64 57 57 - ### Jetstream (Recommended) 65 65 + **Event Types:** 58 66 59 59 - Bluesky's simplified JSON firehose: 67 67 + - `#commit` - Repository changes (record create/update/delete) 68 68 + - `#identity` - DID/handle updates 69 69 + - `#account` - Account status changes (active, deactivated, etc.) 60 70 61 61 - - JSON format (no CBOR decoding) 62 62 - - Reduced bandwidth (zstd compression) 63 63 - - Collection/repo filtering at source 64 64 - - Simpler reconnection with cursors 71 71 + ### Jetstream (Simplified) 72 72 + 73 73 + Bluesky's simplified JSON firehose - ideal for indexing and discovery: 74 74 + 75 75 + - **JSON format**: No CBOR decoding required 76 76 + - **zstd compression**: Reduced bandwidth (enable with `compress=true`) 77 77 + - **Collection filtering**: Subscribe to specific NSIDs 78 78 + - **DID filtering**: Watch specific accounts 79 79 + - **Cursor-based reconnection**: Microsecond timestamps 80 80 + 81 81 + **Public Endpoints:** 82 82 + 83 83 + - `wss://jetstream1.us-east.bsky.network/subscribe` 84 84 + - `wss://jetstream2.us-west.bsky.network/subscribe` 85 85 + 86 86 + **Tradeoffs:** 87 87 + 88 88 + - ⚠️ Events are NOT cryptographically signed (trust the Jetstream operator) 89 89 + - ⚠️ Not self-authenticating data 90 90 + - ✅ Much simpler to implement 91 91 + - ✅ Lower bandwidth and compute requirements 92 92 + 93 93 + ### Reliable Synchronization 94 94 + 95 95 + **Cursor Tracking:** 96 96 + 97 97 + - Store cursor position (microsecond timestamp) per endpoint 98 98 + - Resume from last processed cursor on reconnect 99 99 + - Handle gaps by fetching missing commits via `getRepo` if needed 100 100 + 101 101 + **Per-Repo Revision Tracking:** 102 102 + 103 103 + - Track latest `rev` (TID) for each DID 104 104 + - Compare incoming `rev` against stored value to detect gaps 105 105 + - Use `since` field to detect out-of-order events 106 106 + 107 107 + **Deletion Handling:** 108 108 + 109 109 + - Handle `operation: "delete"` in commit events 110 110 + - Mark records as deleted (soft or hard delete) 111 111 + 112 112 + **Best Practices:** 113 113 + 114 114 + - Process events sequentially per-DID (partition by DID) 115 115 + - Ignore events with `rev` ≤ stored latest rev 116 116 + - Validate records against Lexicon schema before indexing 65 117 66 118 ## Well-Known Endpoints 67 119

+3 -3

docs/todo.md

··· 38 38 39 39 **Firehose Enhancement:** 40 40 41 41 - - [ ] Upgrade firehose consumer to store full record content (not just metadata) 42 42 - - [ ] Add `indexed_decks`, `indexed_cards`, `indexed_notes` tables for remote content 43 43 - - [ ] Track latest processed revision per repo; handle deletions 41 41 + - [x] Upgrade firehose consumer to store full record content (not just metadata) 42 42 + - [x] Add `indexed_decks`, `indexed_cards`, `indexed_notes` tables for remote content 43 43 + - [x] Track latest processed revision per repo; handle deletions 44 44 45 45 **Search & Discovery:** 46 46

+75

migrations/009_2025_12_31_indexed_records.sql

··· 1 1 + -- Migration: Indexed tables for AT Protocol Firehose/Jetstream consumption 2 2 + 3 3 + CREATE TABLE repo_sync_state ( 4 4 + did TEXT PRIMARY KEY, 5 5 + latest_rev TEXT NOT NULL, -- TID of last processed commit 6 6 + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() 7 7 + ); 8 8 + 9 9 + CREATE INDEX idx_repo_sync_state_indexed_at ON repo_sync_state(indexed_at); 10 10 + 11 11 + -- Indexed decks from remote users 12 12 + CREATE TABLE indexed_decks ( 13 13 + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), 14 14 + at_uri TEXT NOT NULL UNIQUE, 15 15 + did TEXT NOT NULL, 16 16 + rkey TEXT NOT NULL, 17 17 + -- Full record content (denormalized for query performance) 18 18 + title TEXT NOT NULL, 19 19 + description TEXT, 20 20 + tags TEXT[] DEFAULT '{}', 21 21 + card_refs TEXT[] DEFAULT '{}', -- AT-URIs to cards in this deck 22 22 + source_refs TEXT[] DEFAULT '{}', -- AT-URIs to source materials 23 23 + license TEXT, 24 24 + record_created_at TIMESTAMPTZ NOT NULL, -- createdAt from the record 25 25 + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), 26 26 + deleted_at TIMESTAMPTZ -- Soft delete for tombstones 27 27 + ); 28 28 + 29 29 + CREATE INDEX idx_indexed_decks_did ON indexed_decks(did); 30 30 + CREATE INDEX idx_indexed_decks_at_uri ON indexed_decks(at_uri); 31 31 + CREATE INDEX idx_indexed_decks_indexed_at ON indexed_decks(indexed_at); 32 32 + CREATE INDEX idx_indexed_decks_tags ON indexed_decks USING GIN(tags); 33 33 + CREATE INDEX idx_indexed_decks_deleted ON indexed_decks(deleted_at) WHERE deleted_at IS NULL; 34 34 + 35 35 + -- Indexed cards from remote users 36 36 + CREATE TABLE indexed_cards ( 37 37 + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), 38 38 + at_uri TEXT NOT NULL UNIQUE, 39 39 + did TEXT NOT NULL, 40 40 + rkey TEXT NOT NULL, 41 41 + deck_ref TEXT NOT NULL, -- AT-URI to parent deck 42 42 + front TEXT NOT NULL, 43 43 + back TEXT NOT NULL, 44 44 + card_type TEXT DEFAULT 'basic', 45 45 + hints TEXT[] DEFAULT '{}', 46 46 + record_created_at TIMESTAMPTZ NOT NULL, 47 47 + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), 48 48 + deleted_at TIMESTAMPTZ 49 49 + ); 50 50 + 51 51 + CREATE INDEX idx_indexed_cards_did ON indexed_cards(did); 52 52 + CREATE INDEX idx_indexed_cards_at_uri ON indexed_cards(at_uri); 53 53 + CREATE INDEX idx_indexed_cards_deck_ref ON indexed_cards(deck_ref); 54 54 + CREATE INDEX idx_indexed_cards_deleted ON indexed_cards(deleted_at) WHERE deleted_at IS NULL; 55 55 + 56 56 + -- Indexed notes from remote users 57 57 + CREATE TABLE indexed_notes ( 58 58 + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), 59 59 + at_uri TEXT NOT NULL UNIQUE, 60 60 + did TEXT NOT NULL, 61 61 + rkey TEXT NOT NULL, 62 62 + title TEXT NOT NULL, 63 63 + body TEXT NOT NULL, 64 64 + tags TEXT[] DEFAULT '{}', 65 65 + visibility TEXT DEFAULT 'public', 66 66 + record_created_at TIMESTAMPTZ NOT NULL, 67 67 + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), 68 68 + deleted_at TIMESTAMPTZ 69 69 + ); 70 70 + 71 71 + CREATE INDEX idx_indexed_notes_did ON indexed_notes(did); 72 72 + CREATE INDEX idx_indexed_notes_at_uri ON indexed_notes(at_uri); 73 73 + CREATE INDEX idx_indexed_notes_tags ON indexed_notes USING GIN(tags); 74 74 + CREATE INDEX idx_indexed_notes_visibility ON indexed_notes(visibility); 75 75 + CREATE INDEX idx_indexed_notes_deleted ON indexed_notes(deleted_at) WHERE deleted_at IS NULL;