//! 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 resends
fn 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 frame
fn 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 subject
fn 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);
    }
}