obix 0.15.1

use crate::error::ObixFault;
use es_entity::errlanes::Laned;
use tokio::sync::{broadcast, mpsc, oneshot};

use std::sync::Arc;

use super::event::{EphemeralEventType, EphemeralOutboxEvent, RawPayload};
use crate::{
    handle::{OwnedTaskHandle, spawn_supervised},
    out::pg_notify::NotifyMessage,
};

pub(super) type CachedEvent = Arc<EphemeralOutboxEvent<RawPayload>>;
pub(super) type CachedEvents = im::HashMap<EphemeralEventType, CachedEvent>;

pub struct CacheHandle {
    ephemeral_event_receiver: Option<broadcast::Receiver<CachedEvent>>,
    backfill_request: mpsc::UnboundedSender<oneshot::Sender<CachedEvents>>,
}

impl CacheHandle {
    pub fn ephemeral_event_stream(&mut self) -> broadcast::Receiver<CachedEvent> {
        self.ephemeral_event_receiver
            .take()
            .expect("receiver already taken")
    }

    pub fn request_current_ephemeral_events(&self) -> oneshot::Receiver<CachedEvents> {
        let (tx, rx) = oneshot::channel();
        let _ = self.backfill_request.send(tx);
        rx
    }
}

#[derive(Debug)]
pub struct EphemeralOutboxEventCache<Tables> {
    ephemeral_event_sender: broadcast::Sender<CachedEvent>,
    backfill_request_send: mpsc::UnboundedSender<oneshot::Sender<CachedEvents>>,
    cache_fill_sender: broadcast::Sender<CachedEvent>,
    _cache_loop_handle: OwnedTaskHandle,
    _phantom: std::marker::PhantomData<Tables>,
}

impl<Tables> EphemeralOutboxEventCache<Tables>
where
    Tables: crate::tables::MailboxTables,
{
    pub fn handle(&self) -> CacheHandle {
        CacheHandle {
            ephemeral_event_receiver: Some(self.ephemeral_event_sender.subscribe()),
            backfill_request: self.backfill_request_send.clone(),
        }
    }

    pub fn cache_fill_sender(&self) -> &broadcast::Sender<CachedEvent> {
        &self.cache_fill_sender
    }

    pub async fn init(
        pool: &sqlx::PgPool,
        event_buffer_size: usize,
        ephemeral_notification_rx: mpsc::Receiver<NotifyMessage>,
    ) -> Result<Self, sqlx::Error> {
        let (backfill_send, backfill_recv) = mpsc::unbounded_channel();
        let (cache_fill_send, cache_fill_recv) = broadcast::channel(event_buffer_size);
        let (ephemeral_event_sender, _) = broadcast::channel(event_buffer_size);

        let cache_loop_handle = Self::spawn_cache_loop(
            pool,
            ephemeral_event_sender.clone(),
            backfill_recv,
            cache_fill_recv,
            cache_fill_send.clone(),
            ephemeral_notification_rx,
        )
        .await?;

        let ret = Self {
            backfill_request_send: backfill_send,
            ephemeral_event_sender,
            cache_fill_sender: cache_fill_send,
            _cache_loop_handle: cache_loop_handle,
            _phantom: std::marker::PhantomData,
        };
        Ok(ret)
    }

    fn insert_into_cache_and_broadcast(
        cache: CachedEvents,
        event: CachedEvent,
        ephemeral_event_sender: &broadcast::Sender<CachedEvent>,
    ) -> CachedEvents {
        // Last-write-wins by recorded_at: notification-triggered fetches for
        // the same event type can complete out of order — a strictly older
        // event must never overwrite (or be broadcast after) a newer one.
        if let Some(cached) = cache.get(&event.event_type)
            && cached.recorded_at > event.recorded_at
        {
            return cache;
        }
        let event_type = event.event_type.clone();
        let cache = cache.update(event_type, event.clone());
        let _ = ephemeral_event_sender.send(event);
        cache
    }

    async fn fetch_event_by_type(
        pool: sqlx::PgPool,
        event_type: EphemeralEventType,
        cache_fill_sender: broadcast::Sender<CachedEvent>,
    ) {
        if let Ok(events) =
            Tables::load_ephemeral_events::<RawPayload>(&pool, Some(event_type)).await
        {
            for event in events {
                let _ = cache_fill_sender.send(Arc::new(event));
            }
        }
    }

    /// Handle a `{event_type, recorded_at}` notification hint.
    ///
    /// The payload is NEVER taken from the notification body. PostgreSQL
    /// performs no authorization on LISTEN/NOTIFY channels — any role able
    /// to connect to this database can signal any channel — so a
    /// notification is only ever a hint that something changed; the event
    /// itself is always (re-)fetched from the table with this process's own
    /// credentials. `recorded_at` bounds that fetch: when the in-process
    /// cache already holds an event at least as recent (the publisher's
    /// post-commit broadcast beat the NOTIFY), the hint is a no-op.
    fn handle_notification(payload: &str, cache: &CachedEvents) -> Option<EphemeralEventType> {
        #[derive(serde::Deserialize)]
        struct NotificationHeader {
            event_type: EphemeralEventType,
            recorded_at: chrono::DateTime<chrono::Utc>,
        }

        let header: NotificationHeader = match serde_json::from_str(payload) {
            Ok(header) => header,
            Err(error) => {
                record_notification_undecodable(&error);
                return None;
            }
        };

        let cache_is_current = cache
            .get(&header.event_type)
            .is_some_and(|cached| cached.recorded_at >= header.recorded_at);
        if cache_is_current {
            None
        } else {
            Some(header.event_type)
        }
    }

    fn process_notification(
        notification: sqlx::postgres::PgNotification,
        cache: &CachedEvents,
    ) -> Option<EphemeralEventType> {
        if notification.channel() == Tables::ephemeral_outbox_events_channel() {
            Self::handle_notification(notification.payload(), cache)
        } else {
            None
        }
    }

    async fn spawn_cache_loop(
        pool: &sqlx::PgPool,
        ephemeral_event_sender: broadcast::Sender<CachedEvent>,
        mut backfill_request: mpsc::UnboundedReceiver<oneshot::Sender<CachedEvents>>,
        mut cache_fill_receiver: broadcast::Receiver<CachedEvent>,
        cache_fill_sender: broadcast::Sender<CachedEvent>,
        mut ephemeral_notification_rx: mpsc::Receiver<NotifyMessage>,
    ) -> Result<OwnedTaskHandle, sqlx::Error> {
        let pool = pool.clone();

        let handle = spawn_supervised("obix::ephemeral_cache_loop", async move {
            let mut ephemeral_cache: CachedEvents = im::HashMap::new();

            loop {
                tokio::select! {
                    biased;

                    result = backfill_request.recv() => {
                        match result {
                            Some(sender) => {
                                let _ = sender.send(ephemeral_cache.clone());
                            }
                            None => {
                                record_backfill_channel_closed();
                                break;
                            }
                        }
                        continue;
                    }

                    result = cache_fill_receiver.recv() => {
                        match result {
                            Ok(event) => {
                                ephemeral_cache = Self::insert_into_cache_and_broadcast(
                                    ephemeral_cache,
                                    event,
                                    &ephemeral_event_sender,
                                );

                                while let Ok(event) = cache_fill_receiver.try_recv() {
                                    ephemeral_cache = Self::insert_into_cache_and_broadcast(
                                        ephemeral_cache,
                                        event,
                                        &ephemeral_event_sender,
                                    );
                                }
                            }
                            Err(broadcast::error::RecvError::Lagged(n)) => {
                                record_cache_fill_lagged(n);
                                continue;
                            }
                            Err(broadcast::error::RecvError::Closed) => {
                                record_cache_fill_closed();
                                break;
                            }
                        }
                    }

                    result = ephemeral_notification_rx.recv() => {
                        match result {
                            Some(message) => {
                                let mut resync_needed = false;
                                let mut messages = vec![message];
                                // Process any additional buffered notifications
                                while let Ok(message) = ephemeral_notification_rx.try_recv() {
                                    messages.push(message);
                                }
                                // Collect the unique event types that need a
                                // fetch. A burst of N notifications for the
                                // same type is drained against one cache
                                // snapshot (fetches are async, the cache is
                                // unchanged during the drain), so deduping
                                // here collapses them into a single fetch
                                // rather than N redundant broadcasts of the
                                // same latest event.
                                let mut types_to_fetch =
                                    std::collections::HashSet::<EphemeralEventType>::new();
                                for message in messages {
                                    match message {
                                        NotifyMessage::Notification(notification) => {
                                            if let Some(event_type) =
                                                Self::process_notification(
                                                    notification,
                                                    &ephemeral_cache,
                                                )
                                            {
                                                types_to_fetch.insert(event_type);
                                            }
                                        }
                                        NotifyMessage::Resync => {
                                            resync_needed = true;
                                        }
                                    }
                                }
                                for event_type in types_to_fetch {
                                    tokio::spawn(Self::fetch_event_by_type(
                                        pool.clone(),
                                        event_type,
                                        cache_fill_sender.clone(),
                                    ));
                                }

                                if resync_needed {
                                    match Tables::load_ephemeral_events::<RawPayload>(&pool, None).await {
                                        Ok(events) => {
                                            for event in events {
                                                let stale = ephemeral_cache
                                                    .get(&event.event_type)
                                                    .is_some_and(|cached| {
                                                        cached.recorded_at >= event.recorded_at
                                                    });
                                                if !stale {
                                                    ephemeral_cache =
                                                        Self::insert_into_cache_and_broadcast(
                                                            ephemeral_cache,
                                                            Arc::new(event),
                                                            &ephemeral_event_sender,
                                                        );
                                                }
                                            }
                                        }
                                        Err(e) => {
                                            record_resync_failed(e)
                                        }
                                    }
                                }
                            }
                            None => {
                                record_notification_channel_closed();
                                break;
                            }
                        }
                    }
                }
            }
        });
        Ok(OwnedTaskHandle::new(handle))
    }
}

#[tracing::instrument(
    name = "obix.ephemeral_cache.backfill_channel_closed",
    level = "error",
    fields(otel.status_code = "ERROR"),
)]
fn record_backfill_channel_closed() {}

#[tracing::instrument(
    name = "obix.ephemeral_cache.cache_fill_lagged",
    level = "error",
    fields(otel.status_code = "ERROR"),
)]
fn record_cache_fill_lagged(dropped: u64) {}

#[tracing::instrument(
    name = "obix.ephemeral_cache.cache_fill_closed",
    level = "error",
    fields(otel.status_code = "ERROR"),
)]
fn record_cache_fill_closed() {}

#[tracing::instrument(
    name = "obix.ephemeral_cache.notification_channel_closed",
    level = "error",
    fields(otel.status_code = "ERROR"),
)]
fn record_notification_channel_closed() {}

#[tracing::instrument(
    name = "obix.ephemeral_cache.resync_failed",
    level = "error",
    skip_all,
    fields(
        otel.status_code = "ERROR",
        error = tracing::field::Empty,
        error.lane = tracing::field::Empty,
        error.code = tracing::field::Empty,
        error.level = tracing::field::Empty,
        exception.message = tracing::field::Empty,
        exception.type = tracing::field::Empty,
    ),
)]
fn record_resync_failed(fault: impl Into<ObixFault>) {
    fault.into().record(&tracing::Span::current());
}

#[tracing::instrument(
    name = "obix.ephemeral_cache.notification_undecodable",
    level = "warn",
    skip_all,
    fields(error = %error),
)]
fn record_notification_undecodable(error: &serde_json::Error) {}