Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754//! best-effort posthog analytics for live firehose events.//! events are queued in memory and flushed in batches so posthog outages never slow down ingest.use std::sync::Arc;use std::sync::atomic::{AtomicU64, Ordering};use std::time::Duration;
use bobbin_edge_index::RepoTransition;use bobbin_record_lru::RecordStore;use bobbin_types::edges::Record;use bobbin_types::ids::owner_did_from_aturi;use bobbin_types::org_tangled::feed::star::StarSubject;use jacquard_common::DefaultStr;use jacquard_common::types::nsid::Nsid;use jacquard_common::types::string::{AtUri, Cid};use jacquard_common::types::tid::Tid;use serde_json::{Map, Value, json};use tokio::sync::mpsc;use tokio_util::sync::CancellationToken;use tracing::{debug, instrument, warn};use url::Url;use uuid::Uuid;
use crate::metrics::{self, AnalyticsResult};
const QUEUE_CAPACITY: usize = 4096;const BATCH_MAX: usize = 100;const FLUSH_INTERVAL: Duration = Duration::from_secs(5);const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);const EVENT_NAMESPACE: Uuid = Uuid::from_u128(0x88db3b6535414734b6f62ed4a5ae1e54);
pub struct PosthogConfig { pub api_key: String, pub host: Url, pub environment: String,}
pub struct Analytics { tx: mpsc::Sender<Queued>, environment: Arc<str>, dropped: Arc<AtomicU64>,}
struct Queued { event: &'static str, distinct_id: String, properties: Map<String, Value>, uuid: Uuid, at: chrono::DateTime<chrono::Utc>,}
impl Analytics { pub fn start(config: PosthogConfig, cancel: CancellationToken) -> Arc<Self> { let (tx, rx) = mpsc::channel(QUEUE_CAPACITY); let dropped = Arc::new(AtomicU64::new(0)); let handle = Arc::new(Self { tx, environment: Arc::from(config.environment.as_str()), dropped: dropped.clone(), }); tokio::spawn(flusher(config, rx, dropped, cancel)); handle }
fn capture( &self, event: &'static str, distinct_id: &str, properties: Map<String, Value>, version: &str, rev: &Tid, ) { let mut properties = properties; properties.insert( "environment".to_owned(), Value::String((*self.environment).to_owned()), ); match self.tx.try_send(Queued { event, distinct_id: distinct_id.to_owned(), properties, uuid: event_uuid(event, version), // the commit time rather than ours, so a resent event lands on the same // timestamp and posthog folds it into the first at: crate::tid_time(rev).unwrap_or_else(chrono::Utc::now), }) { Ok(()) => {} Err(mpsc::error::TrySendError::Full(_)) => { metrics::analytics_events(AnalyticsResult::QueueFull, 1); let dropped = self.dropped.fetch_add(1, Ordering::Relaxed) + 1; // log on powers of two so a stuck flusher does not flood the logs if dropped.is_power_of_two() { warn!( target: "bobbin_ingest::posthog", dropped, capacity = QUEUE_CAPACITY, "analytics queue full, dropping events", ); } } Err(mpsc::error::TrySendError::Closed(_)) => { metrics::analytics_events(AnalyticsResult::FlusherGone, 1); debug!(target: "bobbin_ingest::posthog", "flusher gone, event dropped"); } } }
pub(crate) fn observe_upsert( &self, source: &AtUri, parsed: &Record, cid: Option<&Cid>, rev: &Tid, records: &dyn RecordStore, ) { let Some(distinct_id) = did_str(source) else { debug!( target: "bobbin_ingest::posthog", %source, "record author is not a did, skipping event", ); return; }; if let Some(mapped) = upsert_event(source, parsed, records) { let version = cid.map_or(source.as_ref(), |cid| cid.as_ref()); self.capture(mapped.name, &distinct_id, mapped.properties, version, rev); } else { debug!( target: "bobbin_ingest::posthog", %source, "upsert maps to no event, skipping", ); } }
/// `version` is the triggering record's cid, which keys the event uuid pub(crate) fn observe_repo(&self, transition: &RepoTransition, version: &str, rev: &Tid) { if let Some(mapped) = repo_event(transition) { self.capture( mapped.name, &transition.owner, mapped.properties, version, rev, ); } }
pub(crate) fn observe_delete( &self, source: &AtUri, nsid: &Nsid<DefaultStr>, rev: &Tid, records: &dyn RecordStore, ) { let Some(distinct_id) = did_str(source) else { debug!( target: "bobbin_ingest::posthog", %source, "deleted record author is not a did, skipping event", ); return; }; if let Some(mapped) = delete_event(source, nsid.as_ref(), records) { let cached = records.get(source); let version = cached .as_ref() .map_or(source.as_ref(), |body| body.cid.as_ref()); self.capture(mapped.name, &distinct_id, mapped.properties, version, rev); } }}
// the record cid keys the uuid, so each record version keeps one stable event id// across resendsfn event_uuid(event: &str, version: &str) -> Uuid { Uuid::new_v5(&EVENT_NAMESPACE, format!("{event} {version}").as_bytes())}
struct Mapped { name: &'static str, properties: Map<String, Value>,}
fn did_str(source: &AtUri) -> Option<String> { owner_did_from_aturi(source).map(|did| did.as_ref().to_owned())}
fn prop(key: &str, value: impl Into<Value>) -> Option<(String, Value)> { Some((key.to_owned(), value.into()))}
fn props(entries: impl IntoIterator<Item = Option<(String, Value)>>) -> Map<String, Value> { entries.into_iter().flatten().collect()}
// a repo exists once claimed, so new/renamed come from claim transitions rather than the manifest framefn repo_event(transition: &RepoTransition) -> Option<Mapped> { let repo_at = format!("at://{}/org.tangled.repo.manifest/self", transition.repo); match &transition.previous { None => Some(Mapped { name: "new_repo", properties: props([ prop("repo", transition.slug.as_str()), prop("repo_at", repo_at), ]), }), Some((_, old_slug)) if *old_slug != transition.slug => Some(Mapped { name: "repo_renamed", properties: props([ prop("repo_at", repo_at), prop("owner", transition.owner.as_str()), prop("new_name", transition.slug.as_str()), prop("old_name", old_slug.as_str()), ]), }), Some(_) => None, }}
fn upsert_event(source: &AtUri, record: &Record, records: &dyn RecordStore) -> Option<Mapped> { // observation runs before the commit, so presence in the cache distinguishes edits from creates let existed = |source: &AtUri| records.get(source).is_some(); match record { Record::Star(star) => { let (kind, subject) = star_subject(&star.subject); Some(Mapped { name: "star", properties: props([prop("subject_type", kind), prop("subject", subject)]), }) } Record::Follow(_) => Some(Mapped { name: "follow", properties: props([prop("subject", rkey_str(source))]), }), Record::SnippetString(_) => Some(Mapped { name: if existed(source) { "edit_string" } else { "new_string" }, properties: props([prop("rkey", rkey_str(source))]), }), Record::Profile(_) if existed(source) => Some(Mapped { name: "edit_profile", properties: Map::new(), }), Record::Ticket(ticket) => { let kind = if ticket.target_branch.is_some() { "pull" } else { "issue" }; Some(Mapped { name: if existed(source) { "update_ticket" } else { "new_ticket" }, properties: props([ prop("repo", source.authority().as_ref()), prop("uri", source.as_ref()), prop("kind", kind), ]), }) } Record::FeedComment(comment) => Some(Mapped { name: "new_comment", properties: props([ prop("subject_at", comment.subject.uri.as_ref()), comment .extra_data .as_ref() .and_then(|extra| extra.get("mentions")) .map(|mentions| ("mentions".to_owned(), json!(mentions))), ]), }), _ => None, }}
fn star_subject(subject: &StarSubject) -> (&'static str, String) { match subject { StarSubject::Repo(repo) => ("repo", repo.identity.as_ref().to_owned()), StarSubject::Record(record) => ("record", record.record.uri.as_ref().to_owned()), }}
// firehose deletes only carry the uri, so unstar needs the cached body to know the subjectfn delete_event(source: &AtUri, nsid: &str, records: &dyn RecordStore) -> Option<Mapped> { match nsid { "org.tangled.snippet.string" => Some(Mapped { name: "delete_string", properties: props([prop("rkey", rkey_str(source))]), }), "org.tangled.feed.star" => { let Record::Star(star) = cached_record(source, nsid, records)? else { return None; }; let (kind, subject) = star_subject(&star.subject); Some(Mapped { name: "unstar", properties: props([prop("subject_type", kind), prop("subject", subject)]), }) } "org.tangled.graph.follow" => Some(Mapped { name: "unfollow", properties: props([prop("subject", rkey_str(source))]), }), _ => None, }}
fn cached_record(source: &AtUri, nsid: &str, records: &dyn RecordStore) -> Option<Record> { let body = records.get(source)?; let decoded = Record::from_json_bytes(&Nsid::<DefaultStr>::new_owned(nsid).ok()?, &body.value) .map_err(|error| { debug!( target: "bobbin_ingest::posthog", %source, %error, "cached record body no longer decodes, skipping delete event", ) }) .ok()?; Some(decoded)}
fn rkey_str(source: &AtUri) -> String { source .rkey() .map(|rkey| rkey.as_ref().to_owned()) .unwrap_or_default()}
#[instrument(name = "bobbin_ingest::posthog_flush", skip_all, fields(host = %config.host))]async fn flusher( config: PosthogConfig, mut rx: mpsc::Receiver<Queued>, dropped: Arc<AtomicU64>, cancel: CancellationToken,) { let client = match reqwest::Client::builder().timeout(REQUEST_TIMEOUT).build() { Ok(client) => client, Err(error) => { warn!(target: "bobbin_ingest::posthog", %error, "could not build posthog client, analytics disabled"); return; } }; let url = { let mut url = config.host.clone(); url.set_path("/batch"); url }; let mut batch: Vec<Queued> = Vec::with_capacity(BATCH_MAX); let mut tick = tokio::time::interval(FLUSH_INTERVAL); loop { let queued = tokio::select! { biased; _ = cancel.cancelled() => { while let Ok(queued) = rx.try_recv() { batch.push(queued); if batch.len() >= BATCH_MAX { flush(&client, &url, &config.api_key, &mut batch).await; } } flush(&client, &url, &config.api_key, &mut batch).await; let total_dropped = dropped.load(Ordering::Relaxed); if total_dropped > 0 { warn!(target: "bobbin_ingest::posthog", total_dropped, "events dropped over analytics lifetime"); } debug!(target: "bobbin_ingest::posthog", "posthog flusher stopped"); break; } queued = rx.recv() => match queued { Some(queued) => queued, None => { flush(&client, &url, &config.api_key, &mut batch).await; break; } }, _ = tick.tick() => { flush(&client, &url, &config.api_key, &mut batch).await; continue; } }; batch.push(queued); if batch.len() >= BATCH_MAX { flush(&client, &url, &config.api_key, &mut batch).await; } }}
#[instrument(level = "debug", skip_all, fields(count = batch.len()))]async fn flush(client: &reqwest::Client, url: &Url, api_key: &str, batch: &mut Vec<Queued>) { if batch.is_empty() { return; } let batch_events: Vec<Value> = batch .iter() .map(|queued| { json!({ "event": queued.event, "distinct_id": queued.distinct_id, "properties": queued.properties, "uuid": queued.uuid.to_string(), "timestamp": queued.at.to_rfc3339(), }) }) .collect(); let count = batch_events.len(); match client .post(url.clone()) .json(&json!({"api_key": api_key, "batch": batch_events})) .send() .await { Ok(response) if response.status().is_success() => { // drain the body so hyper can reuse the connection let _ = response.bytes().await; metrics::analytics_events(AnalyticsResult::Sent, count); debug!(target: "bobbin_ingest::posthog", count, "posthog batch accepted"); } Ok(response) => { metrics::analytics_events(AnalyticsResult::Rejected, count); warn!(target: "bobbin_ingest::posthog", count, status = %response.status(), "posthog rejected batch, dropping it"); } Err(error) => { metrics::analytics_events(AnalyticsResult::Unreachable, count); warn!(target: "bobbin_ingest::posthog", count, %error, "posthog unreachable, dropping batch"); } } // drop failed batches instead of retrying so a dead endpoint cannot back up memory batch.clear();}
#[cfg(test)]mod tests { use super::*; use bobbin_types::record::RecordBody; use bytes::Bytes; use std::collections::HashMap; use std::sync::Mutex; use std::time::Duration; use tokio_util::sync::CancellationToken; use wiremock::matchers::{body_partial_json, method, path}; use wiremock::{Mock, MockServer, ResponseTemplate};
const DAWN: &str = "did:plc:dawn";
#[derive(Default)] struct MapStore(Mutex<HashMap<AtUri<DefaultStr>, Arc<RecordBody>>>);
impl RecordStore for MapStore { fn get(&self, uri: &AtUri<DefaultStr>) -> Option<Arc<RecordBody>> { self.0.lock().unwrap().get(uri).cloned() } fn put(&self, uri: AtUri<DefaultStr>, body: Arc<RecordBody>) { self.0.lock().unwrap().insert(uri, body); } fn remove(&self, uri: &AtUri<DefaultStr>) { self.0.lock().unwrap().remove(uri); } }
fn uri(s: &str) -> AtUri<DefaultStr> { AtUri::new_owned(s).unwrap() }
fn seed(store: &MapStore, source: &str, body: serde_json::Value) { let bytes = Bytes::from(serde_json::to_vec(&body).unwrap()); store.put( uri(source), Arc::new(RecordBody { uri: uri(source), cid: "bafyreibjo4hv5t3sx3bh3j3sqfcd5v2v2eirpdcv3gyp4a3ukfa2cdvwla" .parse() .expect("literal cid"), value: bytes, }), ); }
struct UpsertFixture { source: AtUri<DefaultStr>, parsed: Record, }
fn upsert(source: &str, nsid: &str, body: serde_json::Value) -> UpsertFixture { let bytes = serde_json::to_vec(&body).unwrap(); let collection = Nsid::<DefaultStr>::new_owned(nsid).unwrap(); UpsertFixture { source: uri(source), parsed: Record::from_json_bytes(&collection, &bytes).unwrap(), } }
fn mapped(f: &UpsertFixture, store: &MapStore) -> Option<(&'static str, Map<String, Value>)> { upsert_event(&f.source, &f.parsed, store).map(|m| (m.name, m.properties)) }
fn mapped_delete( source: &str, nsid: &str, store: &MapStore, ) -> Option<(&'static str, Map<String, Value>)> { delete_event(&uri(source), nsid, store).map(|m| (m.name, m.properties)) }
fn prop_str(m: &Map<String, Value>, key: &str) -> String { m.get(key) .and_then(Value::as_str) .unwrap_or_default() .to_owned() }
fn transition(previous: Option<(&str, &str)>, owner: &str, slug: &str) -> RepoTransition { RepoTransition { repo: "did:plc:abalone".to_owned(), owner: owner.to_owned(), slug: slug.to_owned(), previous: previous.map(|(o, s)| (o.to_owned(), s.to_owned())), } }
#[test] fn repo_events_follow_claim_transitions() { let Some(m) = repo_event(&transition(None, DAWN, "abalone")) else { panic!() }; assert_eq!(m.name, "new_repo"); assert_eq!(prop_str(&m.properties, "repo"), "abalone"); assert_eq!( prop_str(&m.properties, "repo_at"), "at://did:plc:abalone/org.tangled.repo.manifest/self" );
let Some(m) = repo_event(&transition( Some(("did:plc:old", "old-name")), DAWN, "new-name", )) else { panic!() }; assert_eq!(m.name, "repo_renamed"); assert_eq!(prop_str(&m.properties, "old_name"), "old-name"); assert_eq!(prop_str(&m.properties, "new_name"), "new-name"); assert_eq!(prop_str(&m.properties, "owner"), DAWN);
// a transfer that keeps the slug is not a rename assert!( repo_event(&transition( Some(("did:plc:old", "abalone")), DAWN, "abalone" )) .is_none() ); }
#[test] fn manifest_frames_map_to_no_event() { let op = upsert( "at://did:plc:abalone/org.tangled.repo.manifest/self", "org.tangled.repo.manifest", json!({ "$type": "org.tangled.repo.manifest", "createdAt": "2026-05-01T00:00:00Z", "declaration": {"owner": DAWN, "name": "abalone"}, }), ); assert!(mapped(&op, &MapStore::default()).is_none()); }
#[test] fn delete_events_split_by_cache_need() { let store = MapStore::default(); seed( &store, "at://did:plc:dawn/org.tangled.feed.star/3kzzzzzzzzzzz", json!({ "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:other"}, }), ); let Some((name, props)) = mapped_delete( "at://did:plc:dawn/org.tangled.feed.star/3kzzzzzzzzzzz", "org.tangled.feed.star", &store, ) else { panic!() }; assert_eq!(name, "unstar"); assert_eq!(prop_str(&props, "subject_type"), "repo"); assert_eq!(prop_str(&props, "subject"), "did:plc:other");
let store = MapStore::default(); assert!( mapped_delete( "at://did:plc:dawn/org.tangled.feed.star/3kzzzzzzzzzzz", "org.tangled.feed.star", &store, ) .is_none() );
let Some((name, props)) = mapped_delete( "at://did:plc:dawn/org.tangled.snippet.string/3lzzzzzzzzzzz", "org.tangled.snippet.string", &store, ) else { panic!() }; assert_eq!(name, "delete_string"); assert_eq!(prop_str(&props, "rkey"), "3lzzzzzzzzzzz"); }
#[tokio::test] async fn observe_attributes_events_to_the_record_author() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/batch")) .and(body_partial_json(json!({ "batch": [ {"event": "star", "distinct_id": DAWN, "properties": {"environment": "dev", "subject_type": "repo", "subject": "did:plc:other"}}, ], }))) .respond_with(ResponseTemplate::new(200)) .expect(1) .mount(&server) .await;
let cancel = CancellationToken::new(); let analytics = Analytics::start( PosthogConfig { api_key: "phc_test".to_owned(), host: Url::parse(&server.uri()).unwrap(), environment: "dev".to_owned(), }, cancel.clone(), ); let store = MapStore::default(); let star = upsert( "at://did:plc:dawn/org.tangled.feed.star/3kzzzzzzzzzzz", "org.tangled.feed.star", json!({ "$type": "org.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": {"$type": "org.tangled.feed.star#repo", "identity": "did:plc:other"}, }), ); analytics.observe_upsert(&star.source, &star.parsed, None, &rev(), &store); cancel.cancel(); for _ in 0..100 { if !server.received_requests().await.unwrap().is_empty() { break; } tokio::time::sleep(Duration::from_millis(20)).await; } assert_eq!(server.received_requests().await.unwrap().len(), 1); }
#[tokio::test] async fn flusher_batches_and_drains_on_cancel() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/batch")) .and(body_partial_json(json!({ "api_key": "phc_test", "batch": [{ "event": "follow", "distinct_id": DAWN, "properties": {"environment": "dev"}, "uuid": event_uuid("follow", FOLLOW_CID).to_string(), "timestamp": "2026-05-01T00:00:00+00:00", }], }))) .respond_with(ResponseTemplate::new(200)) .expect(1) .mount(&server) .await;
let cancel = CancellationToken::new(); let analytics = Analytics::start( PosthogConfig { api_key: "phc_test".to_owned(), host: Url::parse(&server.uri()).unwrap(), environment: "dev".to_owned(), }, cancel.clone(), ); let store = MapStore::default(); let follow = upsert( "at://did:plc:dawn/org.tangled.graph.follow/did:plc:other", "org.tangled.graph.follow", json!({ "$type": "org.tangled.graph.follow", "createdAt": "2026-05-01T00:00:00Z", }), ); let cid: Cid = FOLLOW_CID.parse().unwrap(); analytics.observe_upsert(&follow.source, &follow.parsed, Some(&cid), &rev(), &store); cancel.cancel(); for _ in 0..100 { if !server.received_requests().await.unwrap().is_empty() { break; } tokio::time::sleep(Duration::from_millis(20)).await; } assert_eq!(server.received_requests().await.unwrap().len(), 1); }
const FOLLOW_CID: &str = "bafyreibjo4hv5t3sx3bh3j3sqfcd5v2v2eirpdcv3gyp4a3ukfa2cdvwla";
fn rev() -> Tid { Tid::from_time(1_777_593_600_000_000, 0) }
#[tokio::test] async fn rejected_batch_is_dropped_not_retried() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/batch")) .respond_with(ResponseTemplate::new(500)) .expect(1) .mount(&server) .await;
let cancel = CancellationToken::new(); let analytics = Analytics::start( PosthogConfig { api_key: "phc_test".to_owned(), host: Url::parse(&server.uri()).unwrap(), environment: "dev".to_owned(), }, cancel.clone(), ); let store = MapStore::default(); let follow = upsert( "at://did:plc:dawn/org.tangled.graph.follow/did:plc:other", "org.tangled.graph.follow", json!({ "$type": "org.tangled.graph.follow", "createdAt": "2026-05-01T00:00:00Z", }), ); analytics.observe_upsert(&follow.source, &follow.parsed, None, &rev(), &store); cancel.cancel(); for _ in 0..100 { if !server.received_requests().await.unwrap().is_empty() { break; } tokio::time::sleep(Duration::from_millis(20)).await; } tokio::time::sleep(Duration::from_millis(100)).await; assert_eq!(server.received_requests().await.unwrap().len(), 1); }}