//! Keeps mirrors warm and the index fed from Jetstream v2
//! (`network.bsky.jetstream.subscribeEvents`), the firehose re-encoded as
//! JSON and filtered server side. Asking only for `dev.knotless.*` commits,
//! a server receives its own collections' writes and every account's
//! identity, account and sync events, with each record decoded inline, so
//! indexing it needs no `getRecord` round trip.
//!
//! Jetstream frames are not signed: the record is what the Jetstream
//! operator says the commit held. That is the same trust a server places in
//! the relay it follows, and the index is a cache either way.

use crate::App;
use crate::firehose::IDLE;
use crate::index::Index;
pub use knotless::jetstream::{COLLECTIONS, endpoint, parse};
use knotless::jetstream::{Event, Follower};
use std::sync::Arc;
use std::time::{Duration, Instant};

/// How often the index is told where the stream was read to.
const SAVE: Duration = Duration::from_secs(5);

/// Follows Jetstream until the process ends, reconnecting after the last
/// sequence number seen. While disconnected, mirrors fall back to the TTL.
pub fn watch(app: Arc<App>, base: String) -> impl Future<Output = ()> {
    follow(app, base, IDLE)
}

/// Follows Jetstream, giving a connection up after it stays silent through
/// a ping `idle` long. With an index, it starts where the index was last
/// fed from `base`, so a restart misses nothing Jetstream still keeps.
pub fn follow(app: Arc<App>, base: String, idle: Duration) -> impl Future<Output = ()> {
    app.follows("jetstream", &base);
    let feed = Feed {
        index: app.index.clone(),
        app,
        base: base.clone(),
        cursor: None,
        saved: None,
    };
    knotless::jetstream::follow(base, idle, feed)
}

/// The index's feed: each event observed, and where it was read to saved.
struct Feed {
    app: Arc<App>,
    index: Option<Arc<Index>>,
    base: String,
    /// Past the last event heard, as the next connection starts.
    cursor: Option<i64>,
    /// When the cursor was last saved.
    saved: Option<Instant>,
}

impl Feed {
    fn save(&self) {
        if let (Some(index), Some(seq)) = (&self.index, self.cursor) {
            let _ = index.set_stream_seq(&self.base, seq);
        }
    }
}

impl Follower for Feed {
    fn resumes(&self) -> bool {
        true
    }

    fn start(&mut self) -> Option<i64> {
        let index = self.index.as_ref()?;
        self.cursor = index.stream_seq(&self.base).ok().flatten();
        self.cursor
    }

    fn connected(&mut self, up: bool) {
        self.app.connected("jetstream", up);
    }

    fn event(&mut self, event: Event, frame: &str) {
        crate::telemetry::lagged("jetstream", frame);
        self.cursor = event.seq.map(|s| s + 1).or(self.cursor);
        self.app.heard("jetstream", event.seq);
        self.app.observe(event);
        if self.saved.is_none_or(|at| at.elapsed() >= SAVE) {
            self.save();
            self.saved = Some(Instant::now());
        }
    }

    fn ended(&mut self) {
        self.save();
        self.app.connected("jetstream", false);
    }
}