EventSnap/backend/src/services/sse_tickets.rs

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use rand::Rng;

/// Short-lived single-use tickets that let `EventSource` clients open the SSE
/// stream without putting the JWT in the URL (where it would leak via access
/// logs / referer / browser history).
///
/// Flow: client `POST /api/v1/stream/ticket` with `Authorization: Bearer <jwt>`,
/// server returns an opaque ticket, client passes it as `?ticket=...` on the
/// stream open. Tickets are consumed on use and expire after `TTL`.
const TTL: Duration = Duration::from_secs(30);

/// Lifetime of a `Download` ticket, and it is deliberately far longer than [`TTL`].
///
/// A keepsake is up to ~1.4 GB over hotel or cellular wifi, so the download itself outlives a 30 s
/// window many times over — and a resumed transfer arrives minutes or hours after the ticket was
/// minted. A `Download` ticket is therefore a short-lived capability for ONE archive rather than a
/// single-shot nonce: [`SseTicketStore::redeem_download`] does not remove it, so a client may
/// resume with `Range` as many times as the transfer needs.
///
/// The abuse this does NOT open: the ticket is bound to a session (revoked with it), only mints at
/// `/export/ticket` where the 3/day limit is charged, and grants nothing but this event's own
/// keepsake — which every authenticated guest is entitled to download anyway. What it buys is that
/// one dropped connection no longer costs a guest a third of their daily allowance, at the
/// emotional payoff of the product.
const DOWNLOAD_TTL: Duration = Duration::from_secs(6 * 60 * 60);

/// The lifetime that applies to a given kind.
fn ttl_for(kind: TicketKind) -> Duration {
    match kind {
        TicketKind::Download(_) => DOWNLOAD_TTL,
        TicketKind::Sse => TTL,
    }
}

/// Ceiling on outstanding tickets across the whole process.
///
/// Not really about the bytes (~120 each) — about `issue` having had no bound of any kind.
/// Sized well above a real event: ~1000 concurrent clients each holding one live 30 s ticket.
const MAX_TICKETS: usize = 4096;

/// Live tickets one session may hold. Above 1 because two tabs sharing a token open their
/// EventSources concurrently; 4 absorbs that without letting a reconnect loop accumulate.
const MAX_TICKETS_PER_SESSION: usize = 4;

/// How many times one download ticket may be redeemed.
///
/// Making the ticket non-consuming is what lets a dropped transfer resume without spending another
/// of the guest's three daily downloads — but unbounded it also meant a single mint was an
/// unlimited download key for six hours, at BOTH archive endpoints, with the per-day limiter
/// (charged only at mint) never moving. On a 40 GB box serving ~1.4 GB archives that is the one
/// resource an ordinary guest could exhaust without doing anything obviously wrong.
///
/// 20 is far more than a resumed transfer needs (a browser retries a handful of times, not dozens)
/// and turns "unbounded until the ticket expires" into a bounded multiple. It does not make the
/// daily limit exact — that would mean charging per redemption, which would bill a client that
/// restarts from byte 0 instead of sending a `Range`, i.e. re-break the thing this exists to fix.
const MAX_DOWNLOAD_REDEMPTIONS: u32 = 20;

/// What a ticket may be redeemed for.
///
/// The store began life serving only SSE and stayed untyped when the export download started
/// reusing it, which silently made the two interchangeable. That is not a theoretical mixing
/// concern: `POST /stream/ticket` is rate-limited at 60/min per user and charges nothing, while
/// `POST /export/ticket` charges one of three PER-DAY downloads. An untyped ticket let any guest
/// mint at the cheap endpoint and redeem at the expensive one, so the daily export limit was
/// bypassable ~60×/minute — each redemption streaming the whole multi-GB keepsake, `no-store`,
/// off the same filesystem Postgres writes WAL to.
///
/// `consume` therefore requires the kind to MATCH. A ticket is only ever valid for the thing it
/// was minted for.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum TicketKind {
    /// Opens the SSE stream (`GET /stream`). Cheap, high volume.
    Sse,
    /// Downloads ONE export archive. Expensive, rate-limited per day.
    ///
    /// The archive is part of the ticket, not incidental to it. A bare `Download` ticket was
    /// accepted by BOTH `/export/zip` and `/export/html` — they share one authenticator — so with
    /// the redemption budget that makes a ticket resumable, a single mint authorised 20 transfers
    /// spread across both archives. Three mints a day therefore bought 60 full downloads of a
    /// ~1.4 GB keepsake, while the per-day limiter (charged only at mint) never moved.
    Download(ExportKind),
}

/// Which archive a [`TicketKind::Download`] is good for.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum ExportKind {
    /// `Gallery.<event>.<n>.zip` — the original media.
    Zip,
    /// `Memories.<event>.<n>.zip` — the offline HTML viewer.
    Html,
}

#[derive(Clone)]
pub struct SseTicketStore {
    inner: Arc<Mutex<HashMap<String, Entry>>>,
}

#[derive(Clone)]
struct Entry {
    token_hash: String,
    issued_at: Instant,
    kind: TicketKind,
    /// Times this ticket has been redeemed. Only meaningful for `Download` — see
    /// [`MAX_DOWNLOAD_REDEMPTIONS`].
    redemptions: u32,
}

impl SseTicketStore {
    pub fn new() -> Self {
        Self {
            inner: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// Drop every outstanding ticket. Used by the e2e TRUNCATE endpoint: tickets are bound to a
    /// session token hash, and TRUNCATE deletes the sessions out from under them, so anything left
    /// here is a dangling reference to a user that no longer exists.
    pub fn clear(&self) {
        self.inner.lock().unwrap().clear();
    }

    /// Mint a new ticket bound to the caller's session (identified by token hash).
    ///
    /// `None` when the store is at capacity — the caller should answer 503, not evict.
    ///
    /// Three bounds, because `issue` had none: no size cap, no per-caller cap, and no rate
    /// limit on the endpoint, while `prune` ran only hourly against a 30-second TTL. So any
    /// authenticated session could loop the endpoint and grow the map for an hour.
    pub fn issue(&self, token_hash: String, kind: TicketKind) -> Option<String> {
        let ticket = random_ticket();
        let mut map = self.inner.lock().unwrap();

        // Prune on issue rather than only hourly. This alone changes the bound from "tickets
        // minted since the last maintenance tick" to "tickets live at once", which is what the
        // 30 s TTL was always meant to express.
        map.retain(|_, e| e.issued_at.elapsed() <= ttl_for(e.kind));

        // Cap the caller's own outstanding tickets, evicting their oldest. NOT one-per-session:
        // two tabs sharing a token open their EventSources concurrently, and having tab B
        // invalidate tab A's unconsumed ticket looks exactly like a flaky SSE connection.
        //
        // SCOPED TO THE SAME KIND, which matters now that `Download` tickets live 6 h instead of
        // being consumed on first use. A long-lived download ticket is ALWAYS the oldest entry for
        // its session, so a kind-blind cap made it the first thing ordinary SSE churn threw away —
        // and `/export` itself opens an SSE connection on the very session that just minted it.
        // A couple of reconnects during the download (a wifi flap is enough; each attempt that
        // returns early abandons an unconsumed ticket) evicted the ticket out from under a running
        // transfer, so the next `Range` resume 401'd and the guest had to spend another of their
        // three daily downloads. Two flaps and they were locked out of their own keepsake for a day.
        //
        // Per-kind, an SSE reconnect storm can still only evict SSE tickets, which is what the cap
        // was written for; the guest's in-flight keepsake is no longer collateral.
        let mut mine: Vec<(String, Instant)> = map
            .iter()
            .filter(|(_, e)| e.token_hash == token_hash && e.kind == kind)
            .map(|(k, e)| (k.clone(), e.issued_at))
            .collect();
        if mine.len() >= MAX_TICKETS_PER_SESSION {
            mine.sort_by_key(|(_, issued)| *issued);
            for (key, _) in mine.iter().take(mine.len() - MAX_TICKETS_PER_SESSION + 1) {
                map.remove(key);
            }
        }

        // At capacity, REFUSE — never evict a stranger's ticket. Evicting would let one
        // misbehaving client deny SSE to the whole venue, which is worse than failing the
        // request that hit the ceiling.
        if map.len() >= MAX_TICKETS {
            tracing::warn!(
                outstanding = map.len(),
                "SSE ticket store at capacity; refusing to mint"
            );
            return None;
        }

        map.insert(
            ticket.clone(),
            Entry {
                token_hash,
                issued_at: Instant::now(),
                kind,
                redemptions: 0,
            },
        );
        Some(ticket)
    }

    /// Consume a ticket minted for `kind`. Returns `Some(token_hash)` if the ticket exists, is
    /// not expired, and was minted for this purpose. Single-use: the ticket is removed regardless
    /// of whether it was still fresh, so a replay can't slip through after expiry.
    ///
    /// A ticket of the WRONG kind is also removed. It was a valid ticket the caller legitimately
    /// held, so this is not punitive — but leaving it would let a redemption loop probe the store
    /// without ever spending anything, and the client has no legitimate reason to present a
    /// ticket at the wrong endpoint.
    /// Redeem a `Download` ticket WITHOUT consuming it.
    ///
    /// Downloads must be resumable — see [`DOWNLOAD_TTL`]. A browser resumes by re-issuing the same
    /// GET with a `Range` header, so a single-use ticket made `Accept-Ranges` a lie: the retry
    /// authenticated against a ticket that the interrupted attempt had already spent, 401'd, and
    /// the guest had to mint a new one, spending another of their three daily downloads. Two
    /// dropped connections and they were locked out of their own keepsake for ~24 hours.
    ///
    /// Still bound to a live session: the caller re-checks the session on every request, so
    /// revoking a session (logout, "sign out everywhere", a host PIN reset) kills the download too.
    pub fn redeem_download(&self, ticket: &str, want: ExportKind) -> Option<String> {
        let mut map = self.inner.lock().unwrap();
        let entry = map.get_mut(ticket)?;
        // The archive must match the one this ticket was minted for. Both download routes share
        // this authenticator, so without the payload check a ZIP ticket opened the HTML archive
        // too and the redemption budget was spent across both.
        if entry.kind != TicketKind::Download(want) {
            tracing::warn!(
                found = ?entry.kind,
                ?want,
                "ticket presented for the wrong archive (or wrong kind); rejected"
            );
            return None;
        }
        if entry.issued_at.elapsed() > DOWNLOAD_TTL {
            return None;
        }
        if entry.redemptions >= MAX_DOWNLOAD_REDEMPTIONS {
            tracing::warn!(
                redemptions = entry.redemptions,
                "download ticket exceeded its redemption budget; refusing"
            );
            return None;
        }
        entry.redemptions += 1;
        Some(entry.token_hash.clone())
    }

    pub fn consume(&self, ticket: &str, kind: TicketKind) -> Option<String> {
        let mut map = self.inner.lock().unwrap();
        let entry = map.remove(ticket)?;
        if entry.issued_at.elapsed() > ttl_for(entry.kind) {
            return None;
        }
        if entry.kind != kind {
            tracing::warn!(
                expected = ?kind,
                found = ?entry.kind,
                "ticket presented at the wrong endpoint; rejected"
            );
            return None;
        }
        Some(entry.token_hash)
    }

    /// Drop expired entries — called from the background maintenance task so a
    /// long-running process doesn't accumulate stale tickets.
    pub fn prune(&self) {
        let mut map = self.inner.lock().unwrap();
        map.retain(|_, e| e.issued_at.elapsed() <= ttl_for(e.kind));
    }
}

fn random_ticket() -> String {
    // 192 bits of randomness, base32-ish hex. Plenty of entropy and URL-safe.
    let mut rng = rand::rng();
    let mut bytes = [0u8; 24];
    rng.fill(&mut bytes);
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

#[cfg(test)]
mod tests {
    use super::*;

    /// `issue` now returns `Option`; in every test below the store is far from capacity, so an
    /// `expect` here documents that refusing is exceptional rather than routine.
    fn issue(store: &SseTicketStore, hash: &str) -> String {
        store
            .issue(hash.into(), TicketKind::Sse)
            .expect("store has capacity")
    }

    /// The store is shared by two endpoints with wildly different costs: `/stream/ticket` is
    /// 60/min per user and free, `/export/ticket` charges one of three PER-DAY downloads. While
    /// entries were untyped, a ticket minted at the cheap endpoint opened the expensive one — so
    /// the daily export limit could be bypassed ~60×/minute, each redemption streaming the whole
    /// multi-GB keepsake off the disk Postgres writes WAL to.
    ///
    /// Asserted in BOTH directions so this cannot be "fixed" by a check that only guards one.
    #[test]
    fn a_ticket_is_only_valid_for_the_purpose_it_was_minted_for() {
        let store = SseTicketStore::new();

        let sse = store.issue("h".into(), TicketKind::Sse).unwrap();
        assert_eq!(
            store.consume(&sse, TicketKind::Download(ExportKind::Zip)),
            None,
            "an SSE ticket must not open the export download"
        );

        let dl = store
            .issue("h".into(), TicketKind::Download(ExportKind::Zip))
            .unwrap();
        assert_eq!(
            store.consume(&dl, TicketKind::Sse),
            None,
            "a download ticket must not open the SSE stream"
        );

        // And the matching cases still work, so the guard is not simply rejecting everything.
        let sse = store.issue("h".into(), TicketKind::Sse).unwrap();
        assert_eq!(store.consume(&sse, TicketKind::Sse).as_deref(), Some("h"));
        let dl = store
            .issue("h".into(), TicketKind::Download(ExportKind::Zip))
            .unwrap();
        assert_eq!(
            store
                .consume(&dl, TicketKind::Download(ExportKind::Zip))
                .as_deref(),
            Some("h")
        );
    }

    #[test]
    fn issue_then_consume_returns_the_hash_exactly_once() {
        let store = SseTicketStore::new();
        let ticket = issue(&store, "hash-1");
        assert_eq!(
            store.consume(&ticket, TicketKind::Sse).as_deref(),
            Some("hash-1")
        );
        // Single-use: a replay of the same ticket is rejected.
        assert_eq!(
            store.consume(&ticket, TicketKind::Sse),
            None,
            "a consumed ticket must not be reusable"
        );
    }

    #[test]
    fn unknown_ticket_consumes_to_none() {
        let store = SseTicketStore::new();
        assert_eq!(store.consume("never-issued", TicketKind::Sse), None);
    }

    #[test]
    fn issued_tickets_are_unique_and_hex() {
        let store = SseTicketStore::new();
        let a = issue(&store, "h");
        let b = issue(&store, "h");
        assert_ne!(a, b, "each ticket must be unique");
        assert_eq!(a.len(), 48, "24 random bytes → 48 hex chars");
        assert!(a.chars().all(|c| c.is_ascii_hexdigit()));
    }

    #[test]
    fn fresh_ticket_survives_prune() {
        let store = SseTicketStore::new();
        let ticket = issue(&store, "h");
        store.prune(); // not expired → kept
        assert_eq!(
            store.consume(&ticket, TicketKind::Sse).as_deref(),
            Some("h")
        );
    }

    /// Build an entry that is already past the TTL.
    fn insert_stale(store: &SseTicketStore, key: &str, token_hash: &str) {
        store.inner.lock().unwrap().insert(
            key.to_string(),
            Entry {
                kind: TicketKind::Sse,
                token_hash: token_hash.into(),
                issued_at: Instant::now()
                    .checked_sub(TTL + Duration::from_secs(1))
                    .expect("host uptime should exceed the ticket TTL"),
                redemptions: 0,
            },
        );
    }

    #[test]
    fn expired_ticket_consumes_to_none() {
        let store = SseTicketStore::new();
        insert_stale(&store, "stale-ticket", "h");
        assert_eq!(
            store.consume("stale-ticket", TicketKind::Sse),
            None,
            "an expired ticket must not authenticate"
        );
    }

    /// The TTL is 30 s but `prune` only ran hourly, so the map was really bounded by "tickets
    /// minted in the last hour" — which is unbounded for a client in a loop.
    #[test]
    fn issuing_prunes_expired_entries() {
        let store = SseTicketStore::new();
        insert_stale(&store, "stale-a", "someone-else");
        insert_stale(&store, "stale-b", "someone-else");
        issue(&store, "h");
        assert_eq!(
            store.inner.lock().unwrap().len(),
            1,
            "issue must reclaim expired slots, not merely add to them"
        );
    }

    /// Two tabs sharing a token is normal, so the per-session cap must be above 1 — but a
    /// reconnect loop must not accumulate. The caller's OWN oldest is what gets evicted.
    #[test]
    fn a_session_is_capped_and_evicts_only_its_own_oldest() {
        let store = SseTicketStore::new();
        let stranger = issue(&store, "other-session");

        let mut mine: Vec<String> = Vec::new();
        for _ in 0..MAX_TICKETS_PER_SESSION + 2 {
            mine.push(issue(&store, "mine"));
        }

        let live = mine
            .iter()
            .filter(|t| store.inner.lock().unwrap().contains_key(*t))
            .count();
        assert_eq!(live, MAX_TICKETS_PER_SESSION, "one session, bounded");
        assert!(
            store
                .inner
                .lock()
                .unwrap()
                .contains_key(&mine[mine.len() - 1]),
            "the newest ticket is the one the caller is about to use"
        );
        assert_eq!(
            store.consume(&stranger, TicketKind::Sse).as_deref(),
            Some("other-session"),
            "another session's ticket must survive — evicting it would let one client deny \
             SSE to the venue"
        );
    }

    /// At capacity the store REFUSES rather than evicting a stranger. Refusing fails the one
    /// request that hit the ceiling; evicting would break an unrelated client's live stream.
    #[test]
    fn at_capacity_the_store_refuses_instead_of_evicting() {
        let store = SseTicketStore::new();
        {
            let mut map = store.inner.lock().unwrap();
            for i in 0..MAX_TICKETS {
                map.insert(
                    format!("filler-{i}"),
                    Entry {
                        kind: TicketKind::Sse,
                        token_hash: format!("session-{i}"),
                        issued_at: Instant::now(),
                        redemptions: 0,
                    },
                );
            }
        }
        assert_eq!(
            store.issue("newcomer".into(), TicketKind::Sse),
            None,
            "a full store must refuse, so the caller can answer 503"
        );
        assert!(
            store.inner.lock().unwrap().contains_key("filler-0"),
            "no existing ticket may be sacrificed to make room"
        );
    }

    /// A download ticket is deliberately non-consuming so a dropped 1.4 GB transfer can resume
    /// without spending one of the guest's three daily downloads. Unbounded, though, that made a
    /// single mint an unlimited download key for six hours while the per-day limiter — charged
    /// only at mint — never moved. This pins the bound without breaking resumption.
    #[test]
    fn a_download_ticket_resumes_freely_but_not_forever() {
        let store = SseTicketStore::new();
        let ticket = store
            .issue("session-a".into(), TicketKind::Download(ExportKind::Zip))
            .expect("fresh store should issue");

        // Every redemption inside the budget returns the session, so a resumed transfer works.
        for i in 0..MAX_DOWNLOAD_REDEMPTIONS {
            assert_eq!(
                store.redeem_download(&ticket, ExportKind::Zip).as_deref(),
                Some("session-a"),
                "redemption {i} should still be honoured"
            );
        }
        // Past it the ticket is spent: the guest re-mints (and is charged) rather than holding
        // an open-ended key.
        assert_eq!(store.redeem_download(&ticket, ExportKind::Zip), None);
    }

    #[test]
    fn a_download_ticket_opens_only_the_archive_it_was_minted_for() {
        // Both download routes share one authenticator, so without the archive in the ticket a
        // single mint was good for BOTH. Combined with the resume budget that made one mint worth
        // 2 x MAX_DOWNLOAD_REDEMPTIONS transfers of a multi-GB keepsake, while the per-day limit —
        // charged only at mint — never moved.
        let store = SseTicketStore::new();
        let zip = store
            .issue("s".into(), TicketKind::Download(ExportKind::Zip))
            .unwrap();
        assert_eq!(
            store.redeem_download(&zip, ExportKind::Html),
            None,
            "a ZIP ticket must not open the HTML archive"
        );
        assert_eq!(
            store.redeem_download(&zip, ExportKind::Zip).as_deref(),
            Some("s"),
            "...and the refusal above must be about the archive, not a spent ticket"
        );

        let html = store
            .issue("s".into(), TicketKind::Download(ExportKind::Html))
            .unwrap();
        assert_eq!(store.redeem_download(&html, ExportKind::Zip), None);
        assert_eq!(
            store.redeem_download(&html, ExportKind::Html).as_deref(),
            Some("s")
        );
    }

    #[test]
    fn sse_churn_cannot_evict_a_running_download() {
        // A `Download` ticket lives 6 h, so it is ALWAYS the oldest entry for its session — and a
        // kind-blind per-session cap therefore threw it away first. `/export` opens its own SSE
        // connection on the same session, so a couple of reconnects during the transfer evicted
        // the ticket out from under it: the next `Range` resume 401'd and the guest spent another
        // of their three daily downloads. Two wifi flaps and they lost their keepsake for a day.
        let store = SseTicketStore::new();
        let download = store
            .issue("one-session".into(), TicketKind::Download(ExportKind::Zip))
            .unwrap();

        // Far more SSE churn than the per-session cap, all on the same session.
        for _ in 0..(MAX_TICKETS_PER_SESSION * 3) {
            store.issue("one-session".into(), TicketKind::Sse).unwrap();
        }

        assert_eq!(
            store.redeem_download(&download, ExportKind::Zip).as_deref(),
            Some("one-session"),
            "an in-flight keepsake download must survive an SSE reconnect storm"
        );
    }

    /// The kind split is what stops a free SSE ticket from redeeming a rate-limited download.
    #[test]
    fn an_sse_ticket_is_never_redeemable_as_a_download() {
        let store = SseTicketStore::new();
        let sse = store
            .issue("session-b".into(), TicketKind::Sse)
            .expect("fresh store should issue");
        assert_eq!(store.redeem_download(&sse, ExportKind::Zip), None);
        // And it is still usable for what it IS, so the rejection above is about kind, not
        // the ticket having been quietly spent.
        assert_eq!(
            store.consume(&sse, TicketKind::Sse).as_deref(),
            Some("session-b")
        );
    }
}