//! `/stream` events staged by a transaction.

use std::sync::Arc;

use bytes::Bytes;
use fjall::OwnedWriteBatch;
use miette::Result;

use crate::db::keys;
use crate::db::keyspaces::StreamDb;
use crate::db::sequencer::{Assign, publish_in_order};
use crate::types::{
    AccountEvt, BroadcastEvent, EventType, IdentityEvt, LiveRecordEvent, MarshallableEvt,
    StoredEvent,
};

/// a staged `/stream` event, known by its place in the outbox until the
/// transaction commits and gives it an id.
#[derive(Clone, Copy)]
pub(crate) struct StreamEventRef(usize);

impl StreamEventRef {
    pub(super) fn index(self) -> usize {
        self.0
    }
}

#[derive(Default)]
pub(crate) struct StreamOutbox {
    events: Vec<Pending>,
}

enum Pending {
    Record {
        row: Vec<u8>,
        live: Option<(StoredEvent<'static>, Option<Bytes>)>,
    },
    Identity(IdentityEvt),
    Account(AccountEvt<'static>),
}

impl StreamOutbox {
    pub(super) fn is_empty(&self) -> bool {
        self.events.is_empty()
    }

    /// a persisted record event. `live` is the copy to broadcast, with the
    /// record body when live subscribers may use it instead of a db read;
    /// without it, subscribers read the row once a persisted marker covers it.
    pub(crate) fn push_record(
        &mut self,
        row: Vec<u8>,
        live: Option<(StoredEvent<'static>, Option<Bytes>)>,
    ) -> StreamEventRef {
        self.events.push(Pending::Record { row, live });
        StreamEventRef(self.events.len() - 1)
    }

    /// an identity event. broadcast only, never persisted or replayed.
    pub(crate) fn push_identity(&mut self, identity: IdentityEvt) {
        self.events.push(Pending::Identity(identity));
    }

    /// an account event. broadcast only, never persisted or replayed.
    pub(crate) fn push_account(&mut self, account: AccountEvt<'static>) {
        self.events.push(Pending::Account(account));
    }

    pub(super) fn stage<'a>(
        self,
        db: &'a StreamDb,
        assign: &mut Assign<'a>,
        batch: &mut OwnedWriteBatch,
    ) -> Result<StagedStream> {
        let events = self
            .events
            .into_iter()
            .map(|event| {
                let id = db.ids.take(assign)?;
                Ok((id, event.stage(id, db, batch)))
            })
            .collect::<Result<_>>()?;
        Ok(StagedStream { events })
    }
}

impl Pending {
    /// write this event's row, if it has one, and build its broadcast.
    fn stage(self, id: u64, db: &StreamDb, batch: &mut OwnedWriteBatch) -> Option<BroadcastEvent> {
        let (kind, identity, account) = match self {
            Self::Record { row, live } => {
                batch.insert(&db.events, keys::event_key(id), row);
                return live.map(|(stored, inline_block)| {
                    BroadcastEvent::LiveRecord(Arc::new(LiveRecordEvent::new(
                        id,
                        stored,
                        inline_block,
                    )))
                });
            }
            Self::Identity(identity) => (EventType::Identity, Some(identity), None),
            Self::Account(account) => (EventType::Account, None, Some(account)),
        };
        Some(BroadcastEvent::Ephemeral(Box::new(MarshallableEvt {
            id,
            kind,
            record: None,
            identity,
            account,
        })))
    }
}

/// staged events with their ids, in outbox order.
pub(super) struct StagedStream {
    events: Vec<(u64, Option<BroadcastEvent>)>,
}

impl StagedStream {
    #[cfg(feature = "jetstream")]
    pub(super) fn id(&self, event: StreamEventRef) -> u64 {
        self.events[event.0].0
    }

    /// broadcast the staged events, returning the first one's id. the rest
    /// follow it without gaps.
    pub(super) fn publish(self, db: &StreamDb) -> Option<u64> {
        let first = self.events.first().map(|(id, _)| *id);
        publish_in_order(&db.event_tx, self.events, BroadcastEvent::Persisted);
        first
    }
}