noetl-server 2.4.0

//! Event handling API handlers.
//!
//! Handles worker events and command retrieval endpoints.
//!
//! SECURITY: All event payloads are sanitized before storage to prevent
//! sensitive data (bearer tokens, passwords, API keys) from being persisted.

use axum::{
    extract::{Path, State},
    Json,
};
use serde::{Deserialize, Serialize};
use sqlx::Row;
use tracing::{debug, info, warn};

use crate::error::{AppError, AppResult};
use crate::sanitize::sanitize_sensitive_data;
use crate::state::AppState;

/// Worker event request.
///
/// R-1.2 PR-EE-2: aligned with the executor's `ExecutorEvent`
/// shape per the cross-repo event envelope reconciliation tracked
/// on noetl/ai-meta#30.  The legacy `name` field is renamed to
/// `event_type` with a `#[serde(alias = "name")]` so existing
/// worker / CLI clients sending `name` continue to deserialize
/// without breakage during the migration window.  The new
/// optional fields (`event_id`, `status`, `created_at`) align
/// with `noetl_executor::events::ExecutorEvent 0.3.1`; producers
/// that don't populate them get sensible server-side fallbacks
/// (DB-side snowflake_id() for `event_id`, name-derived `status`,
/// `Utc::now()` for `created_at`).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventRequest {
    /// Execution ID.  Wire format is `String` (matches the Python
    /// `EventEmitRequest` and avoids JSON-number precision loss
    /// for large snowflakes in browser clients); parsed to `i64`
    /// before the DB write.
    pub execution_id: String,
    /// Step name.
    pub step: String,
    /// Event type (e.g. `step.enter`, `call.done`, `step.exit`,
    /// `command.completed`).  R-1.2 PR-EE-2: renamed from `name`;
    /// the alias keeps pre-PR-EE clients working.
    #[serde(alias = "name")]
    pub event_type: String,
    /// Event payload/result data.
    ///
    /// R-1.2 PR-EE-2: `context` alias accepted so producers that
    /// send the executor's `ExecutorEvent.context` field
    /// deserialize cleanly into this `payload`.
    #[serde(default, alias = "context")]
    pub payload: serde_json::Value,
    /// Additional metadata.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub meta: Option<serde_json::Value>,
    /// Worker ID.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub worker_id: Option<String>,
    /// Result kind: "data", "ref", or "refs".
    #[serde(default = "default_result_kind")]
    pub result_kind: String,
    /// Result URI for ref kind.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub result_uri: Option<String>,
    /// Event IDs for refs kind.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event_ids: Option<Vec<i64>>,
    /// If true, server should take action.
    #[serde(default = "default_true")]
    pub actionable: bool,
    /// If true, event is for logging/observability.
    #[serde(default = "default_true")]
    pub informative: bool,
    /// Application-side snowflake ID for this event.  Per
    /// `agents/rules/observability.md` Principle 3, the emitting
    /// process generates this BEFORE the row hits the database so
    /// spans / metrics / cross-component correlation can use it
    /// immediately.  Wire format is `String` to avoid JSON-number
    /// precision loss; parsed to `i64` for the DB write.
    /// `None` falls back to the server-side `noetl.snowflake_id()`
    /// function (the existing default).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub event_id: Option<String>,
    /// Lifecycle status (`STARTED` / `RUNNING` / `COMPLETED` /
    /// `FAILED`).  `None` falls back to name-based derivation in
    /// `event_status_from_name`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub status: Option<String>,
    /// Wall-clock when the event was produced.  `None` falls back
    /// to `chrono::Utc::now()`.  Stamping at emit time preserves
    /// per-component ordering across server-clock skew (matters
    /// when multiple workers emit in tight bursts).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub created_at: Option<chrono::DateTime<chrono::Utc>>,
}

fn default_result_kind() -> String {
    "data".to_string()
}

fn default_true() -> bool {
    true
}

/// Response for event handling.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventResponse {
    /// Status of the operation.
    pub status: String,
    /// Event ID that was created.
    pub event_id: i64,
    /// Number of commands generated.
    pub commands_generated: i32,
}

/// Request to claim a command atomically.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClaimRequest {
    /// Worker ID requesting the claim.
    pub worker_id: String,
}

/// Response for successful claim.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClaimResponse {
    /// Status of the claim operation.
    pub status: String,
    /// Command event ID.
    pub event_id: i64,
    /// Execution ID.
    pub execution_id: i64,
    /// Node/step ID.
    pub node_id: String,
    /// Node/step name.
    pub node_name: String,
    /// Action/tool kind.
    pub action: String,
    /// Command context.
    pub context: serde_json::Value,
    /// Command metadata.
    pub meta: serde_json::Value,
}

/// A single batched worker event.
///
/// R-1.2 PR-EE-2: same `name` → `event_type` rename + serde alias
/// as `EventRequest`; same `context` alias for `payload`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BatchEventItem {
    /// Step name.
    pub step: String,
    /// Event type.
    #[serde(alias = "name")]
    pub event_type: String,
    /// Event payload/result data.
    #[serde(default, alias = "context")]
    pub payload: serde_json::Value,
    /// If true, server should take action.
    #[serde(default)]
    pub actionable: bool,
    /// If true, event is for logging/observability.
    #[serde(default = "default_true")]
    pub informative: bool,
}

/// Request for batched event ingestion.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BatchEventRequest {
    /// Execution ID.
    pub execution_id: String,
    /// Worker ID.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub worker_id: Option<String>,
    /// Events to persist.
    pub events: Vec<BatchEventItem>,
}

/// Response for batched event ingestion.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BatchEventResponse {
    /// Status of the operation.
    pub status: String,
    /// Inserted event IDs.
    pub event_ids: Vec<i64>,
    /// Number of generated commands.
    pub commands_generated: i32,
}

/// Command details response.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CommandResponse {
    /// Execution ID.
    pub execution_id: i64,
    /// Node/step ID.
    pub node_id: String,
    /// Node/step name.
    pub node_name: String,
    /// Action/tool kind.
    pub action: String,
    /// Command context (tool config, args, etc.).
    pub context: serde_json::Value,
    /// Command metadata.
    pub meta: serde_json::Value,
}

/// Handle worker event.
///
/// POST /api/events
///
/// Worker reports completion with result (inline or ref).
/// Engine evaluates case/when/then and generates next commands.
///
/// Instrumented per
/// [`agents/rules/observability.md`](https://github.com/noetl/ai-meta/blob/main/agents/rules/observability.md)
/// Principle 1: a counter (`noetl_events_ingested_total{event_type,status}`)
/// and histogram (`noetl_event_ingest_duration_seconds{event_type}`) are
/// recorded on every dispatch.  See [`handle_event_inner`] for the body
/// of the handler.
pub async fn handle_event(
    state: State<AppState>,
    request: Json<EventRequest>,
) -> Result<Json<EventResponse>, AppError> {
    let event_type_for_metrics = request.0.event_type.clone();
    let started_at = std::time::Instant::now();

    let result = handle_event_inner(state, request).await;

    let status_label = if result.is_ok() { "ok" } else { "error" };
    let duration_seconds = started_at.elapsed().as_secs_f64();
    crate::metrics::record_event_ingest(
        &event_type_for_metrics,
        status_label,
        duration_seconds,
    );

    result
}

/// Inner body of [`handle_event`] — same logic, no instrumentation.
///
/// Split out so the wrapper can record metrics on both Ok and Err
/// paths without coupling the body to the recording call.
async fn handle_event_inner(
    State(state): State<AppState>,
    Json(request): Json<EventRequest>,
) -> Result<Json<EventResponse>, AppError> {
    debug!(
        "Event received: execution_id={}, step={}, event_type={}",
        request.execution_id, request.step, request.event_type
    );

    let execution_id: i64 = request
        .execution_id
        .parse()
        .map_err(|_| AppError::Validation("Invalid execution_id".to_string()))?;

    // Events that should NOT trigger engine processing
    let skip_engine_events = [
        "command.claimed",
        "command.started",
        "command.completed",
        "command.failed",
        "step.enter",
    ];

    // For command.claimed, check if already claimed
    if request.event_type == "command.claimed" {
        if let Some(command_id) = get_command_id(&request) {
            if check_already_claimed(&state, execution_id, &command_id, &request.worker_id).await? {
                // Already claimed by same worker - idempotent success
                return Ok(Json(EventResponse {
                    status: "ok".to_string(),
                    event_id: 0,
                    commands_generated: 0,
                }));
            }
        }
    }

    // Build result object based on kind
    let result_obj_raw = build_result_object(&request);
    // SECURITY: Sanitize result data to remove sensitive information (tokens, passwords, etc.)
    let result_obj = sanitize_sensitive_data(&result_obj_raw);

    // Resolve event_id.  R-1.2 PR-EE-2: prefer the
    // application-side snowflake from the request when present
    // (per `agents/rules/observability.md` Principle 3); fall
    // back to the server-side `noetl.snowflake_id()` function
    // when omitted.
    let event_id: i64 = match request.event_id.as_deref() {
        Some(raw) => raw.parse().map_err(|_| {
            AppError::Validation(format!("Invalid event_id: {raw}"))
        })?,
        None => generate_snowflake_id(&state).await?,
    };

    // Get catalog_id from existing events
    let catalog_id = get_catalog_id(&state, execution_id).await?;

    // Build meta object with control flags
    let mut meta_obj = request.meta.clone().unwrap_or(serde_json::json!({}));
    if let serde_json::Value::Object(ref mut map) = meta_obj {
        map.insert(
            "actionable".to_string(),
            serde_json::json!(request.actionable),
        );
        map.insert(
            "informative".to_string(),
            serde_json::json!(request.informative),
        );
        if let Some(ref worker_id) = request.worker_id {
            map.insert("worker_id".to_string(), serde_json::json!(worker_id));
        }
    }
    // SECURITY: Sanitize meta data to remove sensitive information
    let meta_obj = sanitize_sensitive_data(&meta_obj);

    // Resolve status.  R-1.2 PR-EE-2: prefer the
    // application-supplied status when present (so the worker can
    // distinguish STARTED vs RUNNING explicitly); fall back to
    // name-based derivation when omitted for pre-PR-EE clients.
    let derived_status: String = request
        .status
        .clone()
        .unwrap_or_else(|| event_status_from_name(&request.event_type).to_string());

    // Resolve created_at — prefer the application-supplied stamp
    // (avoids server-clock skew when ordering bursts).
    let created_at = request.created_at.unwrap_or_else(chrono::Utc::now);

    // Persist the event
    sqlx::query(
        r#"
        INSERT INTO noetl.event (
            event_id, execution_id, catalog_id, event_type,
            node_id, node_name, status, result, meta, created_at
        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
        "#,
    )
    .bind(event_id)
    .bind(execution_id)
    .bind(catalog_id)
    .bind(&request.event_type)
    .bind(&request.step)
    .bind(&request.step)
    .bind(&derived_status)
    .bind(&result_obj)
    .bind(&meta_obj)
    .bind(created_at)
    .execute(&state.db)
    .await?;

    info!(
        "Event persisted: event_id={}, execution_id={}, event_type={}",
        event_id, execution_id, request.event_type
    );

    // Process through engine if applicable
    let commands_generated = if !skip_engine_events.contains(&request.event_type.as_str()) {
        // TODO: Implement engine event handling
        // This would call the orchestrator to evaluate next steps
        debug!("Would process through engine: event_type={}", request.event_type);
        0
    } else {
        debug!("Skipped engine for administrative event: {}", request.event_type);
        0
    };

    // Trigger orchestrator for workflow progression on command.completed
    if request.event_type == "command.completed" && request.step.to_lowercase() != "end" {
        match trigger_orchestrator(&state, execution_id, event_id).await {
            Ok(cmds) => {
                info!(
                    "Orchestrator generated {} commands for execution {}",
                    cmds, execution_id
                );
            }
            Err(e) => {
                warn!("Orchestrator error: {}", e);
            }
        }
    }

    Ok(Json(EventResponse {
        status: "ok".to_string(),
        event_id,
        commands_generated,
    }))
}

/// Get command details from command.issued event.
///
/// GET /api/commands/{event_id}
///
/// Workers call this to fetch command config after NATS notification.
pub async fn get_command(
    State(state): State<AppState>,
    Path(event_id): Path<i64>,
) -> Result<Json<CommandResponse>, AppError> {
    debug!("Getting command for event_id={}", event_id);

    let row: Option<(i64, String, String, serde_json::Value, serde_json::Value)> =
        sqlx::query_as::<_, (i64, String, String, serde_json::Value, serde_json::Value)>(
            r#"
            SELECT execution_id, node_name, node_type, context, meta
            FROM noetl.event
            WHERE event_id = $1 AND event_type = 'command.issued'
            "#,
        )
        .bind(event_id)
        .fetch_optional(&state.db)
        .await?;

    match row {
        Some((execution_id, node_name, node_type, context, meta)) => Ok(Json(CommandResponse {
            execution_id,
            node_id: node_name.clone(),
            node_name,
            action: node_type,
            context,
            meta,
        })),
        None => Err(AppError::NotFound(format!(
            "command.issued event not found: {}",
            event_id
        ))),
    }
}

/// Atomically claim command and return command details.
///
/// POST /api/commands/{event_id}/claim
pub async fn claim_command(
    State(state): State<AppState>,
    Path(event_id): Path<i64>,
    Json(request): Json<ClaimRequest>,
) -> Result<Json<ClaimResponse>, AppError> {
    debug!(
        "Claim request received: event_id={}, worker_id={}",
        event_id, request.worker_id
    );

    let mut tx = state.db.begin().await?;

    let cmd_row = sqlx::query(
        r#"
        SELECT execution_id, catalog_id, node_name, node_type, context, meta
        FROM noetl.event
        WHERE event_id = $1 AND event_type = 'command.issued'
        "#,
    )
    .bind(event_id)
    .fetch_optional(&mut *tx)
    .await?;

    let Some(row) = cmd_row else {
        return Err(AppError::NotFound(format!(
            "command.issued event not found: {}",
            event_id
        )));
    };

    let execution_id: i64 = row.try_get("execution_id")?;
    let catalog_id: Option<i64> = row.try_get("catalog_id")?;
    let step: String = row.try_get("node_name")?;
    let tool_kind: String = row.try_get("node_type")?;
    let context: serde_json::Value = row
        .try_get("context")
        .unwrap_or_else(|_| serde_json::json!({}));
    let meta: serde_json::Value = row
        .try_get("meta")
        .unwrap_or_else(|_| serde_json::json!({}));
    let command_id = meta
        .get("command_id")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string())
        .unwrap_or_else(|| format!("{}:{}:{}", execution_id, step, event_id));

    // If command already reached terminal state, skip re-claim.
    let terminal_row = sqlx::query(
        r#"
        SELECT event_type
        FROM noetl.event
        WHERE execution_id = $1
          AND event_type IN ('command.completed', 'command.failed')
          AND (meta->>'command_id' = $2 OR result->'data'->>'command_id' = $2)
        ORDER BY event_id DESC
        LIMIT 1
        "#,
    )
    .bind(execution_id)
    .bind(&command_id)
    .fetch_optional(&mut *tx)
    .await?;

    if terminal_row.is_some() {
        return Err(AppError::Conflict(
            "Command already reached terminal state".to_string(),
        ));
    }

    // If execution already cancelled, reject claim.
    let cancelled_row = sqlx::query(
        r#"
        SELECT 1
        FROM noetl.event
        WHERE execution_id = $1
          AND event_type = 'execution.cancelled'
        LIMIT 1
        "#,
    )
    .bind(execution_id)
    .fetch_optional(&mut *tx)
    .await?;

    if cancelled_row.is_some() {
        return Err(AppError::Conflict(
            "Execution has been cancelled".to_string(),
        ));
    }

    // Acquire advisory transaction lock by command id.
    let lock_row =
        sqlx::query("SELECT pg_try_advisory_xact_lock(hashtext($1)::bigint) AS lock_acquired")
            .bind(&command_id)
            .fetch_one(&mut *tx)
            .await?;
    let lock_acquired: bool = lock_row.try_get("lock_acquired")?;
    if !lock_acquired {
        return Err(AppError::Conflict(
            "Command is being claimed by another worker".to_string(),
        ));
    }

    // Check if already claimed by another worker.
    let existing_claim = sqlx::query(
        r#"
        SELECT worker_id, meta
        FROM noetl.event
        WHERE execution_id = $1
          AND event_type = 'command.claimed'
          AND (meta->>'command_id' = $2 OR result->'data'->>'command_id' = $2)
        ORDER BY event_id DESC
        LIMIT 1
        "#,
    )
    .bind(execution_id)
    .bind(&command_id)
    .fetch_optional(&mut *tx)
    .await?;

    if let Some(existing) = existing_claim {
        let worker_id_db: Option<String> = existing.try_get("worker_id").ok();
        let worker_id_meta = existing
            .try_get::<serde_json::Value, _>("meta")
            .ok()
            .and_then(|value| {
                value
                    .get("worker_id")
                    .and_then(|v| v.as_str())
                    .map(|s| s.to_string())
            });
        let existing_worker = worker_id_db.or(worker_id_meta);

        if let Some(existing_worker_id) = existing_worker {
            if existing_worker_id != request.worker_id {
                return Err(AppError::Conflict(format!(
                    "Command already claimed by {}",
                    existing_worker_id
                )));
            }
            // Idempotent claim by same worker.
            tx.commit().await?;
            return Ok(Json(ClaimResponse {
                status: "ok".to_string(),
                event_id,
                execution_id,
                node_id: step.clone(),
                node_name: step,
                action: tool_kind,
                context,
                meta,
            }));
        }
    }

    let claim_event_id = generate_snowflake_id_with_tx(&mut tx).await?;
    let claim_result = serde_json::json!({
        "kind": "data",
        "data": {
            "command_id": command_id,
            "worker_id": request.worker_id,
        }
    });
    let claim_meta = serde_json::json!({
        "command_id": command_id,
        "worker_id": request.worker_id,
        "actionable": false,
        "informative": true,
    });

    sqlx::query(
        r#"
        INSERT INTO noetl.event (
            event_id, execution_id, catalog_id, event_type,
            node_id, node_name, status, result, meta, worker_id, created_at
        ) VALUES (
            $1, $2, $3, $4,
            $5, $6, $7, $8, $9, $10, $11
        )
        "#,
    )
    .bind(claim_event_id)
    .bind(execution_id)
    .bind(catalog_id)
    .bind("command.claimed")
    .bind(&step)
    .bind(&step)
    .bind("RUNNING")
    .bind(claim_result)
    .bind(claim_meta)
    .bind(&request.worker_id)
    .bind(chrono::Utc::now())
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;

    Ok(Json(ClaimResponse {
        status: "ok".to_string(),
        event_id,
        execution_id,
        node_id: step.clone(),
        node_name: step,
        action: tool_kind,
        context,
        meta,
    }))
}

/// Handle batched worker events.
///
/// POST /api/events/batch
pub async fn handle_batch_events(
    State(state): State<AppState>,
    Json(request): Json<BatchEventRequest>,
) -> Result<Json<BatchEventResponse>, AppError> {
    if request.events.is_empty() {
        return Ok(Json(BatchEventResponse {
            status: "ok".to_string(),
            event_ids: Vec::new(),
            commands_generated: 0,
        }));
    }

    let execution_id: i64 = request
        .execution_id
        .parse()
        .map_err(|_| AppError::Validation("Invalid execution_id".to_string()))?;

    let catalog_id = get_catalog_id(&state, execution_id).await?;
    let mut tx = state.db.begin().await?;
    let mut event_ids = Vec::with_capacity(request.events.len());

    for item in &request.events {
        // Batch path uses server-side snowflake (app-side event_id
        // per item isn't carried in BatchEventItem yet; left as a
        // follow-up if batch becomes the worker's primary path).
        let event_id = generate_snowflake_id_with_tx(&mut tx).await?;
        let status = event_status_from_name(&item.event_type);

        let result_obj_raw = serde_json::json!({
            "kind": "data",
            "data": item.payload,
        });
        let result_obj = sanitize_sensitive_data(&result_obj_raw);

        let mut meta_obj = serde_json::json!({
            "actionable": item.actionable,
            "informative": item.informative,
        });
        if let Some(worker_id) = &request.worker_id {
            if let serde_json::Value::Object(ref mut map) = meta_obj {
                map.insert("worker_id".to_string(), serde_json::json!(worker_id));
            }
        }
        let meta_obj = sanitize_sensitive_data(&meta_obj);

        sqlx::query(
            r#"
            INSERT INTO noetl.event (
                event_id, execution_id, catalog_id, event_type,
                node_id, node_name, status, result, meta, worker_id, created_at
            ) VALUES (
                $1, $2, $3, $4,
                $5, $6, $7, $8, $9, $10, $11
            )
            "#,
        )
        .bind(event_id)
        .bind(execution_id)
        .bind(catalog_id)
        .bind(&item.event_type)
        .bind(&item.step)
        .bind(&item.step)
        .bind(status)
        .bind(result_obj)
        .bind(meta_obj)
        .bind(&request.worker_id)
        .bind(chrono::Utc::now())
        .execute(&mut *tx)
        .await?;

        event_ids.push(event_id);
    }

    tx.commit().await?;

    Ok(Json(BatchEventResponse {
        status: "ok".to_string(),
        event_ids,
        commands_generated: 0,
    }))
}

/// Extract command_id from request.
fn get_command_id(request: &EventRequest) -> Option<String> {
    // Try payload first
    if let Some(id) = request.payload.get("command_id").and_then(|v| v.as_str()) {
        return Some(id.to_string());
    }
    // Try meta
    if let Some(meta) = &request.meta {
        if let Some(id) = meta.get("command_id").and_then(|v| v.as_str()) {
            return Some(id.to_string());
        }
    }
    None
}

/// Check if command is already claimed.
async fn check_already_claimed(
    state: &AppState,
    execution_id: i64,
    command_id: &str,
    worker_id: &Option<String>,
) -> AppResult<bool> {
    let row: Option<(Option<String>, Option<serde_json::Value>)> =
        sqlx::query_as::<_, (Option<String>, Option<serde_json::Value>)>(
            r#"
            SELECT worker_id, meta FROM noetl.event
            WHERE execution_id = $1
              AND event_type = 'command.claimed'
              AND (meta->>'command_id' = $2 OR result->'data'->>'command_id' = $2)
            LIMIT 1
            "#,
        )
        .bind(execution_id)
        .bind(command_id)
        .fetch_optional(&state.db)
        .await?;

    if let Some((existing_worker, meta)) = row {
        let existing_worker_id = existing_worker.or_else(|| {
            meta.and_then(|m| {
                m.get("worker_id")
                    .and_then(|v| v.as_str())
                    .map(String::from)
            })
        });

        if let (Some(existing), Some(current)) = (&existing_worker_id, worker_id) {
            if existing != current {
                // Different worker - reject
                return Err(AppError::Conflict(format!(
                    "Command already claimed by {}",
                    existing
                )));
            }
            // Same worker - idempotent
            return Ok(true);
        }
    }

    Ok(false)
}

/// Build result object based on result_kind.
fn build_result_object(request: &EventRequest) -> serde_json::Value {
    match request.result_kind.as_str() {
        "ref" if request.result_uri.is_some() => {
            let uri = request.result_uri.as_ref().unwrap();
            let store_tier = if uri.starts_with("gs://") {
                "gcs"
            } else if uri.starts_with("s3://") {
                "s3"
            } else {
                "artifact"
            };
            serde_json::json!({
                "kind": "ref",
                "store_tier": store_tier,
                "logical_uri": uri,
            })
        }
        "refs" if request.event_ids.is_some() => {
            let event_ids = request.event_ids.as_ref().unwrap();
            serde_json::json!({
                "kind": "refs",
                "event_ids": event_ids,
                "total_parts": event_ids.len(),
            })
        }
        _ => {
            serde_json::json!({
                "kind": "data",
                "data": request.payload,
            })
        }
    }
}

/// Get catalog_id from existing events.
async fn get_catalog_id(state: &AppState, execution_id: i64) -> AppResult<Option<i64>> {
    let row: Option<(i64,)> = sqlx::query_as::<_, (i64,)>(
        "SELECT catalog_id FROM noetl.event WHERE execution_id = $1 LIMIT 1",
    )
    .bind(execution_id)
    .fetch_optional(&state.db)
    .await?;

    Ok(row.map(|(id,)| id))
}

/// Generate a snowflake ID.
async fn generate_snowflake_id(state: &AppState) -> AppResult<i64> {
    let row: (i64,) = sqlx::query_as::<_, (i64,)>("SELECT noetl.snowflake_id()")
        .fetch_one(&state.db)
        .await?;

    Ok(row.0)
}

/// Generate a snowflake ID using an existing transaction.
async fn generate_snowflake_id_with_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> AppResult<i64> {
    let row: (i64,) = sqlx::query_as::<_, (i64,)>("SELECT noetl.snowflake_id()")
        .fetch_one(&mut **tx)
        .await?;
    Ok(row.0)
}

/// Map event name to status.
fn event_status_from_name(event_name: &str) -> &'static str {
    if event_name.contains("done")
        || event_name.contains("exit")
        || event_name.contains("completed")
    {
        "COMPLETED"
    } else if event_name.contains("error") || event_name.contains("failed") {
        "FAILED"
    } else {
        "RUNNING"
    }
}

/// Trigger orchestrator for workflow progression.
async fn trigger_orchestrator(
    _state: &AppState,
    execution_id: i64,
    trigger_event_id: i64,
) -> AppResult<i32> {
    // This would be a full orchestrator implementation
    // For now, just log and return 0
    debug!(
        "Would trigger orchestrator for execution={}, trigger_event={}",
        execution_id, trigger_event_id
    );

    // TODO: Load playbook, reconstruct state from events, evaluate next steps
    Ok(0)
}

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

    fn test_request_skeleton() -> EventRequest {
        EventRequest {
            execution_id: "123".to_string(),
            step: "step1".to_string(),
            event_type: "step.exit".to_string(),
            payload: serde_json::json!({}),
            meta: None,
            worker_id: None,
            result_kind: "data".to_string(),
            result_uri: None,
            event_ids: None,
            actionable: true,
            informative: true,
            event_id: None,
            status: None,
            created_at: None,
        }
    }

    #[test]
    fn test_event_request_defaults() {
        // New canonical field name `event_type`.
        let json = r#"{"execution_id": "123", "step": "step1", "event_type": "step.enter"}"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();

        assert_eq!(request.event_type, "step.enter");
        assert_eq!(request.result_kind, "data");
        assert!(request.actionable);
        assert!(request.informative);
        assert!(request.event_id.is_none());
        assert!(request.status.is_none());
        assert!(request.created_at.is_none());
    }

    #[test]
    fn test_legacy_name_alias_deserializes_into_event_type() {
        // R-1.2 PR-EE-2 back-compat: pre-PR-EE worker / CLI
        // clients send `name` instead of `event_type`.  The
        // alias means they deserialize cleanly without a server
        // restart.
        let json = r#"{"execution_id": "123", "step": "step1", "name": "step.exit"}"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();
        assert_eq!(request.event_type, "step.exit");
    }

    #[test]
    fn test_context_alias_deserializes_into_payload() {
        // Executor producers send the field as `context`; pre-PR
        // clients send `payload`.  Both deserialize into the same
        // field on the server side.
        let json = r#"{
            "execution_id": "123",
            "step": "step1",
            "event_type": "step.exit",
            "context": {"result": 42}
        }"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();
        assert_eq!(request.payload["result"], 42);
    }

    #[test]
    fn test_new_optional_fields_accept_executor_event_shape() {
        // Wire format matching noetl-executor 0.3.1 ExecutorEvent:
        // event_id (snowflake as String), status, created_at, plus
        // the `context` alias.
        let json = r#"{
            "execution_id": "478775660589088776",
            "event_type": "command.completed",
            "step": "fetch_calendar",
            "status": "COMPLETED",
            "created_at": "2026-05-31T03:14:15Z",
            "context": {"items": 42},
            "event_id": "478775660589088777",
            "worker_id": "worker-prod-7",
            "meta": {"attempts": 2}
        }"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();
        assert_eq!(request.event_type, "command.completed");
        assert_eq!(request.event_id.as_deref(), Some("478775660589088777"));
        assert_eq!(request.status.as_deref(), Some("COMPLETED"));
        assert_eq!(request.worker_id.as_deref(), Some("worker-prod-7"));
        assert!(request.created_at.is_some());
    }

    #[test]
    fn test_build_result_object_data() {
        let request = EventRequest {
            payload: serde_json::json!({"output": "success"}),
            ..test_request_skeleton()
        };

        let result = build_result_object(&request);
        assert_eq!(result["kind"], "data");
        assert_eq!(result["data"]["output"], "success");
    }

    #[test]
    fn test_build_result_object_ref() {
        let request = EventRequest {
            result_kind: "ref".to_string(),
            result_uri: Some("gs://bucket/path/to/result".to_string()),
            ..test_request_skeleton()
        };

        let result = build_result_object(&request);
        assert_eq!(result["kind"], "ref");
        assert_eq!(result["store_tier"], "gcs");
        assert_eq!(result["logical_uri"], "gs://bucket/path/to/result");
    }

    #[test]
    fn test_batch_event_item_legacy_name_alias() {
        let json = r#"{"step": "s", "name": "call.done", "payload": {}}"#;
        let item: BatchEventItem = serde_json::from_str(json).unwrap();
        assert_eq!(item.event_type, "call.done");
    }

    #[test]
    fn test_event_response_serialization() {
        let response = EventResponse {
            status: "ok".to_string(),
            event_id: 12345,
            commands_generated: 2,
        };

        let json = serde_json::to_string(&response).unwrap();
        assert!(json.contains("ok"));
        assert!(json.contains("12345"));
    }

    #[test]
    fn test_command_response_serialization() {
        let response = CommandResponse {
            execution_id: 12345,
            node_id: "step1".to_string(),
            node_name: "step1".to_string(),
            action: "python".to_string(),
            context: serde_json::json!({"tool_config": {}}),
            meta: serde_json::json!({"attempt": 1}),
        };

        let json = serde_json::to_string(&response).unwrap();
        assert!(json.contains("step1"));
        assert!(json.contains("python"));
    }
}