[types] use Did for did's in event types · ptr.pet/hydrant@85291e9

+3 -3

src/backfill/mod.rs

··· 4 4 use crate::types::{AccountEvt, BroadcastEvent, RepoState, RepoStatus, ResyncState, StoredEvent}; 5 5 use futures::TryFutureExt; 6 6 use jacquard::api::com_atproto::sync::get_repo::{GetRepo, GetRepoError}; 7 - use jacquard::prelude::*; 8 7 use jacquard::types::did::Did; 8 + use jacquard::{prelude::*, IntoStatic}; 9 9 use jacquard_common::xrpc::XrpcError; 10 10 use jacquard_repo::mst::Mst; 11 11 use jacquard_repo::MemoryBlockStore; ··· 237 237 238 238 let emit_identity = |status: &RepoStatus| { 239 239 let evt = AccountEvt { 240 - did: did.as_str().into(), 240 + did: did.clone(), 241 241 active: !matches!( 242 242 status, 243 243 RepoStatus::Deactivated | RepoStatus::Takendown | RepoStatus::Suspended ··· 403 403 app_state.db.next_event_id.fetch_add(1, Ordering::SeqCst); 404 404 let evt = StoredEvent::Record { 405 405 live: false, 406 - did: did.as_str().into(), 406 + did: did.clone().into_static(), 407 407 rev: rev.as_str().into(), 408 408 collection: collection.into(), 409 409 rkey: rkey.into(),

+5 -8

src/buffer/processor.rs

··· 40 40 let mut to_remove: Vec<Did<'static>> = Vec::new(); 41 41 42 42 loop { 43 - // receive new messages (non-blocking drain) 44 43 while let Ok(msg) = self.rx.try_recv() { 45 44 queues.entry(msg.did.clone()).or_default().push_back(msg); 46 45 } 47 46 48 - // process unblocked DIDs 49 47 for (did, queue) in &mut queues { 50 48 if self.state.blocked_dids.contains_sync(did) { 51 49 continue; ··· 98 96 debug!("processing buffered identity for {did}"); 99 97 let handle = identity.handle.as_ref().map(|h| h.as_str().to_smolstr()); 100 98 let evt = IdentityEvt { 101 - did: did.to_smolstr(), 99 + did: did.clone(), 102 100 handle, 103 101 }; 104 102 ops::emit_identity_event(&self.state.db, evt); ··· 106 104 SubscribeReposMessage::Account(account) => { 107 105 debug!("processing buffered account for {did}"); 108 106 let evt = AccountEvt { 109 - did: did.to_smolstr(), 107 + did: did.clone(), 110 108 active: account.active, 111 109 status: account.status.as_ref().map(|s| s.to_smolstr()), 112 110 }; 113 111 ops::emit_account_event(&self.state.db, evt); 114 112 113 + let did = did.clone(); 115 114 let state = self.state.clone(); 116 - let did = did.clone(); 117 - let account = account.clone(); // Account is 'static in BufferedMessage 115 + let account = account.clone(); 118 116 119 117 tokio::task::spawn_blocking(move || -> Result<()> { 120 - // handle status updates 121 118 if !account.active { 122 119 use jacquard::api::com_atproto::sync::subscribe_repos::AccountStatus; 123 120 if let Some(status) = &account.status { ··· 208 205 .flatten() 209 206 } 210 207 211 - async fn remove_from_db_buffer(&self, did: &str, buffered_at: i64) -> Result<()> { 208 + async fn remove_from_db_buffer(&self, did: &Did<'_>, buffered_at: i64) -> Result<()> { 212 209 let key = keys::buffer_key(did, buffered_at); 213 210 self.state.db.buffer.remove(key).into_diagnostic()?; 214 211 Ok(())

+2 -2

src/db/keys.rs

··· 66 66 } 67 67 68 68 // key format: {DID}\x00{timestamp} (for buffer entries) 69 - pub fn buffer_key(did: &str, timestamp: i64) -> Vec<u8> { 69 + pub fn buffer_key(did: &Did, timestamp: i64) -> Vec<u8> { 70 70 let mut key = Vec::with_capacity(did.len() + 1 + 8); 71 - key.extend_from_slice(did.as_bytes()); 71 + key.extend_from_slice(did_prefix(did).as_bytes()); 72 72 key.push(SEP); 73 73 key.extend_from_slice(&timestamp.to_be_bytes()); 74 74 key

+3 -2

src/ingest/mod.rs

··· 8 8 use n0_future::StreamExt; 9 9 use std::sync::atomic::Ordering; 10 10 use std::sync::Arc; 11 + use std::time::Duration; 11 12 use tracing::{debug, error, info}; 12 13 use url::Url; 13 14 ··· 62 63 Ok(s) => s, 63 64 Err(e) => { 64 65 error!("failed to connect to firehose: {e}, retrying in 5s..."); 65 - tokio::time::sleep(std::time::Duration::from_secs(5)).await; 66 + tokio::time::sleep(Duration::from_secs(5)).await; 66 67 continue; 67 68 } 68 69 }; ··· 83 84 } 84 85 85 86 error!("firehose disconnected, reconnecting in 5s..."); 86 - tokio::time::sleep(std::time::Duration::from_secs(5)).await; 87 + tokio::time::sleep(Duration::from_secs(5)).await; 87 88 } 88 89 } 89 90

+4 -3

src/ops.rs

··· 2 2 use crate::types::{AccountEvt, BroadcastEvent, IdentityEvt, MarshallableEvt, StoredEvent}; 3 3 use jacquard::api::com_atproto::sync::subscribe_repos::Commit; 4 4 use jacquard::cowstr::ToCowStr; 5 + use jacquard::IntoStatic; 5 6 use jacquard_repo::car::reader::parse_car_bytes; 6 7 use miette::{IntoDiagnostic, Result}; 7 8 use smol_str::{SmolStr, ToSmolStr}; ··· 12 13 13 14 // emitting identity is ephemeral 14 15 // we dont replay these, consumers can just fetch identity themselves if they need it 15 - pub fn emit_identity_event(db: &Db, evt: IdentityEvt) { 16 + pub fn emit_identity_event(db: &Db, evt: IdentityEvt<'static>) { 16 17 let event_id = db.next_event_id.fetch_add(1, Ordering::SeqCst); 17 18 let marshallable = MarshallableEvt { 18 19 id: event_id, ··· 24 25 let _ = db.event_tx.send(BroadcastEvent::Ephemeral(marshallable)); 25 26 } 26 27 27 - pub fn emit_account_event(db: &Db, evt: AccountEvt) { 28 + pub fn emit_account_event(db: &Db, evt: AccountEvt<'static>) { 28 29 let event_id = db.next_event_id.fetch_add(1, Ordering::SeqCst); 29 30 let marshallable = MarshallableEvt { 30 31 id: event_id, ··· 186 187 187 188 let evt = StoredEvent::Record { 188 189 live, 189 - did: did.as_str().into(), 190 + did: did.clone().into_static(), 190 191 rev: commit.rev.as_str().into(), 191 192 collection: collection.into(), 192 193 rkey: rkey.into(),

+24 -15

src/types.rs

··· 69 69 // from src/api/event.rs 70 70 71 71 #[derive(Debug, Serialize, Deserialize, Clone)] 72 - pub struct MarshallableEvt { 72 + pub struct MarshallableEvt<'i> { 73 73 pub id: u64, 74 74 #[serde(rename = "type")] 75 75 pub event_type: SmolStr, 76 + #[serde(borrow)] 76 77 #[serde(skip_serializing_if = "Option::is_none")] 77 - pub record: Option<RecordEvt>, 78 + pub record: Option<RecordEvt<'i>>, 79 + #[serde(borrow)] 78 80 #[serde(skip_serializing_if = "Option::is_none")] 79 - pub identity: Option<IdentityEvt>, 81 + pub identity: Option<IdentityEvt<'i>>, 82 + #[serde(borrow)] 80 83 #[serde(skip_serializing_if = "Option::is_none")] 81 - pub account: Option<AccountEvt>, 84 + pub account: Option<AccountEvt<'i>>, 82 85 } 83 86 84 87 #[derive(Clone, Debug)] 85 88 pub enum BroadcastEvent { 86 89 Persisted(u64), 87 - Ephemeral(MarshallableEvt), 90 + Ephemeral(MarshallableEvt<'static>), 88 91 } 89 92 90 93 #[derive(Debug, Serialize, Deserialize, Clone)] 91 - pub struct RecordEvt { 94 + pub struct RecordEvt<'i> { 92 95 pub live: bool, 93 - pub did: SmolStr, 96 + #[serde(borrow)] 97 + pub did: Did<'i>, 94 98 pub rev: SmolStr, 95 99 pub collection: SmolStr, 96 100 pub rkey: SmolStr, ··· 102 106 } 103 107 104 108 #[derive(Debug, Serialize, Deserialize, Clone)] 105 - pub struct IdentityEvt { 106 - pub did: SmolStr, 109 + pub struct IdentityEvt<'i> { 110 + #[serde(borrow)] 111 + pub did: Did<'i>, 107 112 #[serde(skip_serializing_if = "Option::is_none")] 108 113 pub handle: Option<SmolStr>, 109 114 } 110 115 111 116 #[derive(Debug, Serialize, Deserialize, Clone)] 112 - pub struct AccountEvt { 113 - pub did: SmolStr, 117 + pub struct AccountEvt<'i> { 118 + #[serde(borrow)] 119 + pub did: Did<'i>, 114 120 pub active: bool, 115 121 #[serde(skip_serializing_if = "Option::is_none")] 116 122 pub status: Option<SmolStr>, 117 123 } 118 124 119 125 #[derive(Debug, Serialize, Deserialize, Clone)] 120 - pub enum StoredEvent { 126 + pub enum StoredEvent<'i> { 121 127 Record { 122 128 live: bool, 123 - did: SmolStr, 129 + #[serde(borrow)] 130 + did: Did<'i>, 124 131 rev: SmolStr, 125 132 collection: SmolStr, 126 133 rkey: SmolStr, 127 134 action: SmolStr, 128 135 cid: Option<SmolStr>, 129 136 }, 130 - Identity(IdentityEvt), 131 - Account(AccountEvt), 137 + #[serde(borrow)] 138 + Identity(IdentityEvt<'i>), 139 + #[serde(borrow)] 140 + Account(AccountEvt<'i>), 132 141 }

Configure Feed

Configure Feed