cf-chat-engine 0.2.0

//! Repository ports (domain-owned abstractions).
//!
//! These traits are the persistence *ports* the domain services depend on.
//! They are defined here — in the domain layer — and return domain types, so
//! the domain never imports infrastructure (DE0301). The concrete Sea-ORM
//! implementations live in `infra::db::repo` and are the *adapters*; unit tests
//! swap in in-memory mocks. This mirrors the existing `VariantRepo` port in
//! `domain::service::variant_service`.
//
// @cpt-cf-chat-engine-infra-repo-root:p3

use async_trait::async_trait;
use serde_json::Value as JsonValue;
use time::OffsetDateTime;
use toolkit_macros::domain_model;
use toolkit_odata::{ODataQuery, Page};
use uuid::Uuid;

use chat_engine_sdk::models::{
    FileCitation, LifecycleState, LinkCitation, LinkReference, Message, MessagePartInput, Session,
    SessionType,
};

use crate::domain::error::ChatEngineError;
use crate::domain::reaction::{MessageReaction, ReactionType};

// ===========================================================================
// messages
// ===========================================================================

/// Inputs for [`MessageRepo::insert_user_and_assistant_stub`].
#[domain_model]
#[derive(Debug, Clone)]
pub struct NewUserMessage {
    pub session_id: Uuid,
    /// Owning tenant, denormalized from the session (sourced from the JWT
    /// `tenant_id` claim, never the request body). Stamped onto both the user
    /// message and the assistant stub. `None` keeps the column NULL.
    pub tenant_id: Option<String>,
    /// Author of the user message — the authenticated user (JWT `user_id`
    /// claim). Applied to the user message only; the assistant stub has no
    /// human author and is always persisted with `user_id = NULL`.
    pub user_id: Option<String>,
    /// Parent in the message tree (`None` only for the very first message in
    /// a session). When `Some`, the caller MUST have already verified the
    /// parent belongs to `session_id` via
    /// [`MessageRepo::find_message_in_session`] — the repo only enforces FK
    /// integrity at the DB layer.
    pub parent_message_id: Option<Uuid>,
    /// User-supplied message body as an ordered list of typed parts. Persisted
    /// into `message_parts` numbered `0..n` in list order. Must be non-empty
    /// (the service rejects an empty body before reaching the repo).
    pub parts: Vec<MessagePartInput>,
    /// Opaque external file UUIDs (Chat Engine never fetches the bytes).
    /// Stored as `Some(json-array)` when non-empty; `None` otherwise.
    pub file_ids: Option<Vec<Uuid>>,
    /// Optional message-level metadata (request_id, capability snapshot, …).
    pub metadata: Option<JsonValue>,
}

/// Outcome of [`MessageRepo::insert_user_and_assistant_stub`] — both IDs are
/// generated by the repo so the caller can immediately reference them in the
/// outgoing NDJSON wire (`StreamingStartEvent { message_id }` uses the
/// assistant id).
#[domain_model]
#[derive(Debug, Clone, Copy)]
pub struct InsertedPair {
    pub user_message_id: Uuid,
    pub assistant_message_id: Uuid,
    /// `variant_index` assigned to the user message. The assistant stub
    /// always starts at `variant_index=0` because it has no siblings yet
    /// (variants are a recreate-time concern, Phase 6).
    pub user_variant_index: i32,
}

/// Citations and references the plugin attached to its terminal `text` part
/// (FR-023). Persisted into the child tables keyed by the finalized part id.
#[domain_model]
#[derive(Debug, Clone, Default)]
pub struct PartCitations {
    pub file_citations: Vec<FileCitation>,
    pub link_citations: Vec<LinkCitation>,
    pub references: Vec<LinkReference>,
}

impl PartCitations {
    /// True when there is nothing to persist.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.file_citations.is_empty()
            && self.link_citations.is_empty()
            && self.references.is_empty()
    }
}

/// Three-way outcome the finalize step must persist. Mirrors the streaming
/// state machine in ADR-0003.
#[domain_model]
#[derive(Debug, Clone)]
pub enum FinalizeOutcome {
    /// Plugin streamed `Complete` and the response body is ready.
    Complete {
        /// Accumulated assistant text (concatenation of all `Chunk.chunk`
        /// payloads). Stored under `content.text` per the SDK convention as
        /// the primary text part (`number = 0`).
        text: String,
        /// Plugin-defined metadata (model, finish_reason, usage). When
        /// `Some` it overrides whatever the stub carried.
        metadata: Option<JsonValue>,
        /// Citations/references attached to the completed text part (FR-023).
        citations: PartCitations,
        /// Additional typed parts streamed via `StreamingEvent::Part`
        /// (images/videos/links/code). Persisted in arrival order after the
        /// text part; each carries its own citations (FR-024 Phase B).
        extra_parts: Vec<MessagePartInput>,
    },
    /// Client disconnected (or deadline elapsed) mid-stream. Persist the
    /// partial response with `cancelled=true, partial=true` markers.
    Cancelled {
        /// Accumulated text up to the cancellation point.
        text: String,
    },
    /// Plugin yielded an `Err`/`StreamingErrorEvent` mid-stream. Persist
    /// the partial response with `finish_reason` recorded.
    Errored {
        /// Accumulated text up to the failure point.
        text: String,
        /// Human-readable error message (already trimmed of secrets; safe
        /// to surface to operators).
        error: String,
        /// Canonical finish_reason label (`"error"`, `"timeout"`,
        /// `"interrupted"`). Stored under `metadata.finish_reason`.
        finish_reason: &'static str,
    },
}

/// Repository surface for the `messages` table.
#[async_trait]
pub trait MessageRepo: Send + Sync {
    /// Insert the user message + a pre-allocated assistant stub in a single
    /// SERIALIZABLE transaction.
    async fn insert_user_and_assistant_stub(
        &self,
        req: NewUserMessage,
    ) -> Result<InsertedPair, ChatEngineError>;

    /// Atomically finalize the assistant message: updates `content`,
    /// `is_complete`, and `metadata` in one statement.
    async fn finalize_assistant(
        &self,
        session_id: Uuid,
        assistant_message_id: Uuid,
        outcome: FinalizeOutcome,
    ) -> Result<(), ChatEngineError>;

    /// Fetch the active visible history for a session, in chronological
    /// order. Filters `is_active=true AND is_hidden_from_backend=false`.
    /// Applies `LIMIT depth` only when `depth.is_some()`.
    async fn fetch_active_history(
        &self,
        session_id: Uuid,
        depth: Option<u32>,
    ) -> Result<Vec<Message>, ChatEngineError>;

    /// Look up a single message scoped to a session — used by the service
    /// to validate `parent_message_id` before any write.
    async fn find_message_in_session(
        &self,
        session_id: Uuid,
        message_id: Uuid,
    ) -> Result<Option<Message>, ChatEngineError>;

    /// Look up a single message by id alone (no session scope) — used by the
    /// REST routes keyed on `message_id` only to resolve the owning
    /// `session_id` before delegating to session-scoped service methods. The
    /// caller is responsible for the subsequent ownership check.
    ///
    /// Default impl returns `Internal` so existing test mocks compile; the
    /// Sea-ORM impl overrides this.
    async fn find_message_by_id(
        &self,
        message_id: Uuid,
    ) -> Result<Option<Message>, ChatEngineError> {
        let _ = message_id;
        Err(ChatEngineError::internal(
            "find_message_by_id not implemented for this repository",
        ))
    }

    /// Phase 7 hook (context management). Return the *raw* active path for a
    /// session — every message with `is_active=true AND is_complete=true`
    /// ordered by `created_at ASC`. Unlike [`Self::fetch_active_history`],
    /// this method does NOT filter out `is_hidden_from_backend=true` rows.
    async fn list_active_path(&self, session_id: Uuid) -> Result<Vec<Message>, ChatEngineError> {
        self.fetch_active_history(session_id, None).await
    }

    /// Phase 6 hook (recreate). Insert a fresh assistant sibling under
    /// `parent_message_id` with `variant_index = MAX+1`.
    ///
    /// Default impl returns `Internal` so existing test mocks compile;
    /// the Sea-ORM impl overrides this.
    async fn insert_assistant_variant_stub(
        &self,
        session_id: Uuid,
        parent_message_id: Uuid,
        tenant_id: Option<String>,
    ) -> Result<InsertedPair, ChatEngineError> {
        let _ = (session_id, parent_message_id, tenant_id);
        Err(ChatEngineError::internal(
            "insert_assistant_variant_stub not implemented for this repository",
        ))
    }

    /// Phase 8 hook (session intelligence). Persist an AI-generated summary
    /// as a `role=system` message and atomically flip the listed messages'
    /// `is_hidden_from_backend` flag to `true`.
    async fn insert_summary_message(
        &self,
        session_id: Uuid,
        text: String,
        metadata: Option<JsonValue>,
        summarized_ids: Vec<Uuid>,
        tenant_id: Option<String>,
    ) -> Result<Uuid, ChatEngineError> {
        let _ = (session_id, text, metadata, summarized_ids, tenant_id);
        Err(ChatEngineError::internal(
            "insert_summary_message not implemented for this repository",
        ))
    }

    /// Phase 8 hook (retention cleanup). List every non-root message in the
    /// session ordered by `created_at ASC`.
    async fn list_non_root_messages_chrono(
        &self,
        session_id: Uuid,
    ) -> Result<Vec<Message>, ChatEngineError> {
        let _ = session_id;
        Ok(Vec::new())
    }

    /// Phase 8 hook (retention cleanup). List every non-root message in
    /// the session whose `created_at` is older than `older_than`.
    async fn list_non_root_messages_older_than(
        &self,
        session_id: Uuid,
        older_than: OffsetDateTime,
    ) -> Result<Vec<Message>, ChatEngineError> {
        let _ = (session_id, older_than);
        Ok(Vec::new())
    }

    /// Phase 8 hook (retention cleanup, bounded). Count non-root
    /// messages in the session.
    async fn count_non_root_messages(&self, session_id: Uuid) -> Result<u64, ChatEngineError> {
        let _ = session_id;
        Ok(0)
    }

    /// Phase 8 hook (retention cleanup, bounded). Return the IDs of
    /// the OLDEST non-root messages in the session — ordered by
    /// `created_at ASC` — capped at `limit`.
    async fn list_oldest_non_root_message_ids(
        &self,
        session_id: Uuid,
        limit: u32,
    ) -> Result<Vec<Uuid>, ChatEngineError> {
        let _ = (session_id, limit);
        Ok(Vec::new())
    }

    /// Phase 8 hook (retention cleanup, bounded). Return the IDs of
    /// non-root messages with `created_at < older_than`.
    async fn list_non_root_message_ids_older_than(
        &self,
        session_id: Uuid,
        older_than: OffsetDateTime,
        limit: u32,
    ) -> Result<Vec<Uuid>, ChatEngineError> {
        let _ = (session_id, older_than, limit);
        Ok(Vec::new())
    }

    /// Phase 8 hook (retention cleanup). Atomically delete the message
    /// identified by `root_id` and every descendant reachable through
    /// `parent_message_id` within `session_id`.
    async fn delete_message_subtree(
        &self,
        session_id: Uuid,
        root_id: Uuid,
    ) -> Result<u64, ChatEngineError> {
        let _ = (session_id, root_id);
        Err(ChatEngineError::internal(
            "delete_message_subtree not implemented for this repository",
        ))
    }
}

// ===========================================================================
// message reactions
// ===========================================================================

/// Outcome of [`ReactionRepo::upsert`]. Carries the persisted row plus the
/// `previous_reaction_type` captured before the write so the service can
/// populate the plugin event without an extra round-trip.
#[domain_model]
#[derive(Debug, Clone)]
pub struct ReactionUpsertOutcome {
    /// Stored reaction after the upsert (always `Like` or `Dislike` —
    /// `None` is handled by [`ReactionRepo::delete`], not this method).
    pub reaction: MessageReaction,
    /// Prior reaction value for this `(message_id, user_id)` pair.
    /// `None` when the user had no reaction on this message before.
    pub previous_reaction_type: Option<ReactionType>,
}

/// Outcome of [`ReactionRepo::delete`]. The service uses `applied` to
/// determine the HTTP response shape (200 with `applied: false` when no
/// row was present, 200 with `applied: true` when a row was removed) and
/// `previous_reaction_type` to populate the plugin event.
#[domain_model]
#[derive(Debug, Clone)]
pub struct ReactionDeleteOutcome {
    /// True when a row was removed (i.e. the user had a prior reaction).
    /// False when no row existed (idempotent no-op).
    pub applied: bool,
    /// Prior reaction value when `applied = true`; `None` otherwise.
    pub previous_reaction_type: Option<ReactionType>,
}

/// Repository surface for the `message_reactions` table.
#[async_trait]
pub trait ReactionRepo: Send + Sync {
    /// Fetch the stored reaction for `(message_id, user_id)`. Returns
    /// `Ok(None)` when no row exists (the user has not reacted).
    async fn get_by_pk(
        &self,
        message_id: Uuid,
        user_id: &str,
    ) -> Result<Option<MessageReaction>, ChatEngineError>;

    /// UPSERT the user's reaction on the message. The caller MUST have
    /// already validated `reaction_type` is `Like` or `Dislike` —
    /// `ReactionType::None` is a DELETE marker handled by
    /// [`Self::delete`].
    ///
    /// Pre-update value is captured atomically with the write so the
    /// service can populate `MessageReactionEvent.previous_reaction_type`
    /// without a second round-trip.
    async fn upsert(
        &self,
        message_id: Uuid,
        user_id: &str,
        reaction_type: ReactionType,
    ) -> Result<ReactionUpsertOutcome, ChatEngineError>;

    /// Remove the user's reaction (idempotent). Returns `applied = false`
    /// when no row existed.
    async fn delete(
        &self,
        message_id: Uuid,
        user_id: &str,
    ) -> Result<ReactionDeleteOutcome, ChatEngineError>;

    /// Enumerate every reaction on the given message. Ordering is left
    /// unspecified — callers that need deterministic order should sort
    /// client-side.
    async fn list_by_message(
        &self,
        message_id: Uuid,
    ) -> Result<Vec<MessageReaction>, ChatEngineError>;
}

// ===========================================================================
// plugin configs
// ===========================================================================

/// Repository surface for `plugin_configs`.
///
/// The trait is object-safe — services hold `Arc<dyn PluginConfigRepo>` so the
/// concrete backend (Sea-ORM today, an in-memory mock in tests) is swappable.
#[async_trait]
pub trait PluginConfigRepo: Send + Sync {
    /// Look up the JSONB config for a `(plugin_instance_id, session_type_id)`
    /// pair. Returns `Ok(None)` when the row is absent.
    async fn find(
        &self,
        plugin_instance_id: &str,
        session_type_id: Uuid,
    ) -> Result<Option<JsonValue>, ChatEngineError>;

    /// Insert or update the JSONB config for a pair. The `updated_at` column
    /// MUST be refreshed on both insert and update.
    async fn upsert(
        &self,
        plugin_instance_id: &str,
        session_type_id: Uuid,
        config: JsonValue,
    ) -> Result<(), ChatEngineError>;

    /// Remove the row keyed by `(plugin_instance_id, session_type_id)`. A
    /// missing row is NOT an error — this method is idempotent.
    async fn delete(
        &self,
        plugin_instance_id: &str,
        session_type_id: Uuid,
    ) -> Result<(), ChatEngineError>;
}

// ===========================================================================
// sessions
// ===========================================================================

/// Default soft-delete grace period applied when the session-type doesn't
/// declare an explicit [`crate::domain::retention::RetentionPolicy`]. Mirrors
/// the PRD default (30 days).
pub const DEFAULT_SOFT_DELETE_RETENTION_DAYS: i64 = 30;

/// Inputs for [`SessionRepo::insert`]. The repo owns the persistence shape
/// (building the `ActiveModel` and defaulting `lifecycle_state`, `share_token`
/// and the soft-delete columns); callers pass domain values only.
#[domain_model]
#[derive(Debug, Clone)]
pub struct NewSession {
    pub session_id: Uuid,
    pub tenant_id: String,
    pub user_id: String,
    pub client_id: Option<String>,
    pub session_type_id: Option<Uuid>,
    pub metadata: Option<JsonValue>,
    pub created_at: OffsetDateTime,
    pub updated_at: OffsetDateTime,
}

/// Typed answer returned by [`SessionRepo::check_session_scope`] — the only
/// API service code should use to distinguish cross-tenant (403) from
/// cross-user / missing (404) access. The repo only exposes the session when
/// the caller proved ownership.
#[allow(clippy::large_enum_variant)]
#[domain_model]
#[derive(Debug, Clone)]
pub enum SessionScopeCheck {
    /// Session exists AND is owned by `(tenant_id, user_id)`.
    Owned(Session),
    /// Session exists but lives in a different tenant. Maps to HTTP 403.
    WrongTenant,
    /// Session exists in this tenant but belongs to a different user.
    /// Maps to HTTP 404 (anti-enumeration, ADR-0021).
    WrongUser,
    /// Session id does not resolve to any row, or the row is `HardDeleted`.
    NotFound,
}

/// Repository surface for the `sessions` table. Returns domain [`Session`]
/// values; the entity ↔ domain conversion lives in the infra adapter.
#[async_trait]
pub trait SessionRepo: Send + Sync {
    /// Insert a new session row and return the persisted domain value.
    async fn insert(&self, new: NewSession) -> Result<Session, ChatEngineError>;

    /// Scoped read by `(tenant_id, user_id, session_id)`. `HardDeleted` rows
    /// are treated as absent.
    async fn find_by_id(
        &self,
        tenant_id: &str,
        user_id: &str,
        session_id: Uuid,
    ) -> Result<Option<Session>, ChatEngineError>;

    /// Cursor/OData-paginated list scoped to `(tenant_id, user_id)`.
    async fn list_paginated(
        &self,
        tenant_id: &str,
        user_id: &str,
        query: &ODataQuery,
    ) -> Result<Page<Session>, ChatEngineError>;

    /// Replace the session metadata JSONB (scoped write).
    async fn update_metadata(
        &self,
        tenant_id: &str,
        user_id: &str,
        session_id: Uuid,
        metadata: Option<JsonValue>,
    ) -> Result<Session, ChatEngineError>;

    /// Replace the enabled-capabilities JSONB (scoped write).
    async fn update_capabilities(
        &self,
        tenant_id: &str,
        user_id: &str,
        session_id: Uuid,
        capabilities: Option<JsonValue>,
    ) -> Result<Session, ChatEngineError>;

    /// Transition the lifecycle state (scoped write). Restoring to `Active`
    /// clears the soft-delete bookkeeping columns.
    async fn update_lifecycle_state(
        &self,
        tenant_id: &str,
        user_id: &str,
        session_id: Uuid,
        new_state: LifecycleState,
    ) -> Result<Session, ChatEngineError>;

    /// Soft-delete: flip to `soft_deleted`, stamp `deleted_at` and
    /// `scheduled_hard_delete_at = now + retention_days`.
    async fn soft_delete(
        &self,
        tenant_id: &str,
        user_id: &str,
        session_id: Uuid,
        retention_days: i64,
    ) -> Result<Session, ChatEngineError>;

    /// Hard-delete the session and cascade messages + reactions. Returns
    /// `false` when the row was already absent.
    async fn hard_delete(
        &self,
        tenant_id: &str,
        user_id: &str,
        session_id: Uuid,
    ) -> Result<bool, ChatEngineError>;

    /// Return the `scheduled_hard_delete_at` timestamp for a scoped session,
    /// if any. Used by the restore flow to reject sessions past the grace
    /// window. `None` when absent or unset.
    ///
    /// Default impl returns `None` so test mocks compile.
    async fn scheduled_hard_delete_at(
        &self,
        tenant_id: &str,
        user_id: &str,
        session_id: Uuid,
    ) -> Result<Option<OffsetDateTime>, ChatEngineError> {
        let _ = (tenant_id, user_id, session_id);
        Ok(None)
    }

    /// Phase 8 (retention). Active sessions in a tenant, keyset-paged by
    /// `session_id`. Default impl returns empty so mocks compile.
    async fn list_active_sessions_for_tenant(
        &self,
        tenant_id: &str,
        after: Option<Uuid>,
        limit: u32,
    ) -> Result<Vec<Session>, ChatEngineError> {
        let _ = (tenant_id, after, limit);
        Ok(Vec::new())
    }

    /// Phase 8 (retention). Distinct tenants owning ≥1 active session.
    async fn list_tenants_with_active_sessions(&self) -> Result<Vec<String>, ChatEngineError> {
        Ok(Vec::new())
    }

    /// Internal building block for [`Self::check_session_scope`] — unscoped
    /// lookup by primary key. Service code MUST NOT call it directly.
    /// Default impl returns `None` so mocks compile.
    async fn find_by_session_id_unscoped(
        &self,
        session_id: Uuid,
    ) -> Result<Option<Session>, ChatEngineError> {
        let _ = session_id;
        Ok(None)
    }

    /// Resolve a session id under the caller's scope and report ownership as
    /// a typed discriminant. The default issues an unscoped lookup and
    /// discriminates in-process.
    async fn check_session_scope(
        &self,
        tenant_id: &str,
        user_id: &str,
        session_id: Uuid,
    ) -> Result<SessionScopeCheck, ChatEngineError> {
        let Some(row) = self.find_by_session_id_unscoped(session_id).await? else {
            return Ok(SessionScopeCheck::NotFound);
        };
        if row.tenant_id.as_str() != tenant_id {
            return Ok(SessionScopeCheck::WrongTenant);
        }
        if row.user_id.as_str() != user_id {
            return Ok(SessionScopeCheck::WrongUser);
        }
        Ok(SessionScopeCheck::Owned(row))
    }

    /// Phase 10 (sharing). Unauthenticated lookup by `share_token`, excluding
    /// `HardDeleted`. Default impl returns `None` so mocks compile.
    async fn find_by_share_token(
        &self,
        share_token: &str,
    ) -> Result<Option<Session>, ChatEngineError> {
        let _ = share_token;
        Ok(None)
    }

    /// Phase 10 (sharing). Atomically replace `share_token` AND metadata,
    /// bumping `updated_at`. Default impl errors so mocks that don't share
    /// compile.
    async fn update_share_token(
        &self,
        tenant_id: &str,
        user_id: &str,
        session_id: Uuid,
        share_token: Option<String>,
        metadata: Option<JsonValue>,
    ) -> Result<Session, ChatEngineError> {
        let _ = (tenant_id, user_id, session_id, share_token, metadata);
        Err(ChatEngineError::internal(
            "update_share_token not implemented for this repository",
        ))
    }
}

// ===========================================================================
// session types
// ===========================================================================

/// Inputs for [`SessionTypeRepo::insert`]. The repo owns the persistence shape
/// (building the `ActiveModel`); callers pass domain values only.
#[domain_model]
#[derive(Debug, Clone)]
pub struct NewSessionType {
    pub session_type_id: Uuid,
    pub name: String,
    pub plugin_instance_id: Option<String>,
    pub created_at: OffsetDateTime,
    pub updated_at: OffsetDateTime,
}

/// Repository surface for the `session_types` table.
#[async_trait]
pub trait SessionTypeRepo: Send + Sync {
    /// Persist a new session type and return the stored domain value.
    async fn insert(&self, new: NewSessionType) -> Result<SessionType, ChatEngineError>;

    /// Lookup by surrogate primary key.
    async fn find_by_id(
        &self,
        session_type_id: Uuid,
    ) -> Result<Option<SessionType>, ChatEngineError>;

    /// List all session types ordered by `created_at DESC`. Session types are
    /// operator-managed and small, so this surface is not paginated.
    async fn list(&self) -> Result<Vec<SessionType>, ChatEngineError>;
}

// ===========================================================================
// stream event resume buffer
// ===========================================================================

/// One buffered wire event, returned by [`StreamEventBuffer::read_since`].
#[domain_model]
#[derive(Debug, Clone)]
pub struct BufferedEvent {
    /// Per-message sequence number (the SSE `id:`).
    pub seq: u64,
    /// Serialized wire event, replayed verbatim.
    pub event: JsonValue,
}

/// Short-TTL append-only buffer that bridges SSE reconnects (`Last-Event-ID`).
/// Not durable history — the persisted message is the durable record.
#[async_trait]
pub trait StreamEventBuffer: Send + Sync {
    /// Append `event` at `(message_id, seq)` with the given TTL deadline.
    /// Idempotent on the PK: a re-append of the same `(message_id, seq)` is a
    /// no-op (so a retried write never errors).
    async fn append(
        &self,
        message_id: Uuid,
        seq: u64,
        event: JsonValue,
        expires_at: OffsetDateTime,
    ) -> Result<(), ChatEngineError>;

    /// Return buffered events for `message_id` with `seq > after_seq` (or all
    /// when `after_seq` is `None`), ordered by `seq` ascending.
    async fn read_since(
        &self,
        message_id: Uuid,
        after_seq: Option<u64>,
    ) -> Result<Vec<BufferedEvent>, ChatEngineError>;

    /// Delete all rows whose `expires_at` is at or before `now`. Returns the
    /// number of rows removed. Called by the periodic TTL sweep.
    async fn delete_expired(&self, now: OffsetDateTime) -> Result<u64, ChatEngineError>;
}