PiCloud/crates/manager-core/src/outbox_event_emitter.rs

//! `OutboxEventEmitter` — the real `ServiceEventEmitter` that replaces
//! v1.1.0's `NoopEventEmitter` once the triggers framework lands.
//!
//! On each `emit` (a KV mutation, future doc/file/pubsub event, etc.):
//!   1. Look up matching triggers for the event's (app_id, source, op,
//!      collection) tuple via `TriggerRepo::list_matching_*`.
//!   2. For each match, write one outbox row carrying the event payload
//!      serialized as a `TriggerEvent`.
//!
//! Defaults applied at write time so `OutboxRow.payload` carries
//! everything the dispatcher needs to reconstruct the executor
//! invocation without joining back to the trigger row.
//!
//! Non-KV `ServiceEvent` sources are silently dropped in v1.1.1 — the
//! dispatcher only knows how to fire KV triggers this release. Future
//! sources (docs/files/pubsub) add their own dispatch arm.

use std::sync::Arc;

use async_trait::async_trait;
use picloud_shared::{
    DocsEventOp, EmitError, FileMeta, FilesEventOp, GroupId, KvEventOp, SdkCallCx, ServiceEvent,
    ServiceEventEmitter, TriggerEvent,
};

use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxSourceKind};
use crate::trigger_repo::TriggerRepo;

pub struct OutboxEventEmitter {
    triggers: Arc<dyn TriggerRepo>,
    outbox: Arc<dyn OutboxRepo>,
}

impl OutboxEventEmitter {
    #[must_use]
    pub fn new(triggers: Arc<dyn TriggerRepo>, outbox: Arc<dyn OutboxRepo>) -> Self {
        Self { triggers, outbox }
    }
}

#[async_trait]
impl ServiceEventEmitter for OutboxEventEmitter {
    async fn emit(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> {
        match event.source {
            "kv" => self.emit_kv(cx, event).await,
            "docs" => self.emit_docs(cx, event).await,
            "files" => self.emit_files(cx, event).await,
            // Future sources land here. For now, silently drop — the
            // SDK calls `events.emit(...)` unconditionally for forward
            // compat, so swallowing without an error is correct.
            _ => Ok(()),
        }
    }

    #[allow(clippy::too_many_lines)] // one match arm per source kind (kv/docs/files)
    async fn emit_shared(
        &self,
        cx: &SdkCallCx,
        owning_group: GroupId,
        event: ServiceEvent,
    ) -> Result<(), EmitError> {
        // §11.6: a shared-collection write. Match `shared = true` triggers on
        // the OWNING group; each fires under the WRITER app (`cx.app_id`), the
        // same "group template runs under the firing app" model.
        let source_kind = match event.source {
            "kv" => OutboxSourceKind::Kv,
            "docs" => OutboxSourceKind::Docs,
            "files" => OutboxSourceKind::Files,
            _ => return Ok(()),
        };
        let Some(collection) = event.collection.clone() else {
            return Ok(());
        };
        let (matches, trigger_event) = match event.source {
            "kv" => {
                let Some(op) = KvEventOp::from_wire(event.op) else {
                    return Ok(());
                };
                let m = self
                    .triggers
                    .list_matching_shared_kv(owning_group, &collection, op)
                    .await
                    .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
                let ev = TriggerEvent::Kv {
                    op,
                    collection,
                    key: event.key.clone().unwrap_or_default(),
                    value: event.payload.clone(),
                };
                (
                    m.into_iter()
                        .map(|x| (x.trigger_id, x.script_id))
                        .collect::<Vec<_>>(),
                    ev,
                )
            }
            "docs" => {
                let Some(op) = DocsEventOp::from_wire(event.op) else {
                    return Ok(());
                };
                let m = self
                    .triggers
                    .list_matching_shared_docs(owning_group, &collection, op)
                    .await
                    .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
                let ev = TriggerEvent::Docs {
                    op,
                    collection,
                    id: event.key.clone().unwrap_or_default(),
                    data: event.payload.clone(),
                    prev_data: event.old_payload.clone(),
                };
                (
                    m.into_iter()
                        .map(|x| (x.trigger_id, x.script_id))
                        .collect::<Vec<_>>(),
                    ev,
                )
            }
            "files" => {
                let Some(op) = FilesEventOp::from_wire(event.op) else {
                    return Ok(());
                };
                let Some(meta) = event
                    .payload
                    .clone()
                    .and_then(|v| serde_json::from_value::<FileMeta>(v).ok())
                else {
                    return Ok(());
                };
                let m = self
                    .triggers
                    .list_matching_shared_files(owning_group, &collection, op)
                    .await
                    .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;
                let ev = TriggerEvent::Files {
                    op,
                    collection,
                    id: meta.id.to_string(),
                    name: meta.name,
                    content_type: meta.content_type,
                    size: meta.size,
                    checksum: meta.checksum,
                    prev: event.old_payload.clone(),
                };
                (
                    m.into_iter()
                        .map(|x| (x.trigger_id, x.script_id))
                        .collect::<Vec<_>>(),
                    ev,
                )
            }
            _ => return Ok(()),
        };
        if matches.is_empty() {
            return Ok(());
        }
        let payload = serde_json::to_value(&trigger_event)
            .map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?;
        for (trigger_id, script_id) in matches {
            self.outbox
                .insert(NewOutboxRow {
                    app_id: cx.app_id,
                    source_kind,
                    trigger_id: Some(trigger_id),
                    script_id: Some(script_id),
                    reply_to: None,
                    payload: payload.clone(),
                    origin_principal: cx.principal.as_ref().map(|p| p.user_id),
                    trigger_depth: cx.trigger_depth.saturating_add(1),
                    root_execution_id: Some(cx.root_execution_id),
                })
                .await
                .map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?;
        }
        Ok(())
    }
}

impl OutboxEventEmitter {
    async fn emit_kv(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> {
        let Some(op) = KvEventOp::from_wire(event.op) else {
            return Ok(()); // unknown op — drop quietly
        };
        let Some(collection) = event.collection.clone() else {
            return Ok(()); // KV events always carry a collection — defensively skip
        };
        let key = event.key.clone().unwrap_or_default();

        let matches = self
            .triggers
            .list_matching_kv(cx.app_id, &collection, op)
            .await
            .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;

        if matches.is_empty() {
            return Ok(());
        }

        // Serialize the originating event as a TriggerEvent so the
        // dispatcher can hand it to the script as `ctx.event` without
        // round-tripping back to the trigger row.
        let trigger_event = TriggerEvent::Kv {
            op,
            collection,
            key,
            value: event.payload.clone(),
        };
        let payload = serde_json::to_value(&trigger_event)
            .map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?;

        for m in matches {
            self.outbox
                .insert(NewOutboxRow {
                    app_id: cx.app_id,
                    source_kind: OutboxSourceKind::Kv,
                    trigger_id: Some(m.trigger_id),
                    script_id: Some(m.script_id),
                    reply_to: None,
                    payload: payload.clone(),
                    origin_principal: cx.principal.as_ref().map(|p| p.user_id),
                    trigger_depth: cx.trigger_depth.saturating_add(1),
                    root_execution_id: Some(cx.root_execution_id),
                })
                .await
                .map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?;
        }
        Ok(())
    }

    /// v1.1.2. Mirrors `emit_kv` — fan out a docs mutation across
    /// matching docs triggers + write one outbox row each. The
    /// `prev_data` change-data-capture surface is preserved from the
    /// `ServiceEvent.old_payload` field (set by `DocsServiceImpl` on
    /// update and delete; `None` for create).
    async fn emit_docs(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> {
        let Some(op) = DocsEventOp::from_wire(event.op) else {
            return Ok(());
        };
        let Some(collection) = event.collection.clone() else {
            return Ok(());
        };
        let id = event.key.clone().unwrap_or_default();

        let matches = self
            .triggers
            .list_matching_docs(cx.app_id, &collection, op)
            .await
            .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;

        if matches.is_empty() {
            return Ok(());
        }

        let trigger_event = TriggerEvent::Docs {
            op,
            collection,
            id,
            data: event.payload.clone(),
            prev_data: event.old_payload.clone(),
        };
        let payload = serde_json::to_value(&trigger_event)
            .map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?;

        for m in matches {
            self.outbox
                .insert(NewOutboxRow {
                    app_id: cx.app_id,
                    source_kind: OutboxSourceKind::Docs,
                    trigger_id: Some(m.trigger_id),
                    script_id: Some(m.script_id),
                    reply_to: None,
                    payload: payload.clone(),
                    origin_principal: cx.principal.as_ref().map(|p| p.user_id),
                    trigger_depth: cx.trigger_depth.saturating_add(1),
                    root_execution_id: Some(cx.root_execution_id),
                })
                .await
                .map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?;
        }
        Ok(())
    }

    /// v1.1.5. Fan out a files mutation across matching files triggers.
    /// The `ServiceEvent.payload` is the file **metadata** (never the
    /// blob bytes); `old_payload` is the prior metadata (the deleted
    /// row's metadata on delete). The `TriggerEvent::Files` carries the
    /// metadata fields explicitly + `prev` for the change-data-capture
    /// surface.
    async fn emit_files(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> {
        let Some(op) = FilesEventOp::from_wire(event.op) else {
            return Ok(());
        };
        let Some(collection) = event.collection.clone() else {
            return Ok(());
        };
        // The payload is the FileMeta JSON the FilesServiceImpl emitted.
        let Some(meta) = event
            .payload
            .clone()
            .and_then(|v| serde_json::from_value::<FileMeta>(v).ok())
        else {
            return Ok(());
        };

        let matches = self
            .triggers
            .list_matching_files(cx.app_id, &collection, op)
            .await
            .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?;

        if matches.is_empty() {
            return Ok(());
        }

        let trigger_event = TriggerEvent::Files {
            op,
            collection,
            id: meta.id.to_string(),
            name: meta.name,
            content_type: meta.content_type,
            size: meta.size,
            checksum: meta.checksum,
            prev: event.old_payload.clone(),
        };
        let payload = serde_json::to_value(&trigger_event)
            .map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?;

        for m in matches {
            self.outbox
                .insert(NewOutboxRow {
                    app_id: cx.app_id,
                    source_kind: OutboxSourceKind::Files,
                    trigger_id: Some(m.trigger_id),
                    script_id: Some(m.script_id),
                    reply_to: None,
                    payload: payload.clone(),
                    origin_principal: cx.principal.as_ref().map(|p| p.user_id),
                    trigger_depth: cx.trigger_depth.saturating_add(1),
                    root_execution_id: Some(cx.root_execution_id),
                })
                .await
                .map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?;
        }
        Ok(())
    }
}