use crate::context::BridgeContext; use crate::mapping::{map_command_name_to_convex, payload_json_for_command, ConvexMutationRequest}; use crate::types::BridgeResult; use crate::validation::validate_command_envelope; use core_domain::{RefLink, Timestamp, WorkspaceId}; use core_protocol::{command::CommandResult, AffectedObject, CommandEnvelope}; use event_log::{CommandLogRecord, CommandLogStatus, DomainEventRecord, EventStatus}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct WritePipelineResult { pub mutation: ConvexMutationRequest, pub command_log: CommandLogRecord, pub domain_event: DomainEventRecord, pub result: CommandResult, } pub fn build_write_request( context: &BridgeContext, command: &CommandEnvelope, ) -> BridgeResult { validate_command_envelope(context, command)?; let workspace_id = command .target .as_ref() .and_then(|target| target.workspace_id.clone()) .or_else(|| context.workspace_id.clone()); Ok(ConvexMutationRequest { function_name: map_command_name_to_convex(&command.name).to_string(), deployment_id: context.deployment_id.clone(), project_id: context.project_id.clone(), workspace_id, request_id: context.request_id.clone(), trace_id: context.trace_id.clone(), idempotency_key: command .idempotency_key .clone() .or_else(|| context.idempotency_key.clone()), actor_id: context.actor_id.clone(), payload_json: payload_json_for_command(context, &command.name), }) } pub fn build_command_log_record( command_id: &str, command_name: &str, context: &BridgeContext, workspace_id: &str, ) -> CommandLogRecord { CommandLogRecord { command_log_id: format!("clog_{command_id}"), command_name: command_name.into(), actor_type: context.actor_type.clone(), actor_id: context.actor_id.clone(), source: context.source_channel.clone(), workspace_id: WorkspaceId::new(workspace_id), target_objects: vec![RefLink::External(format!("workspace:{workspace_id}"))], payload_summary: format!( "command={command_name};request_id={};trace_id={}", context.request_id, context.trace_id ), refs_json: format!( "[\"request:{}\",\"trace:{}\"]", context.request_id, context.trace_id ), idempotency_key: context.idempotency_key.clone(), status: CommandLogStatus::Pending, created_at: Timestamp::new("2026-04-11T00:00:00Z"), finished_at: None, trace_id: context.trace_id.clone(), request_id: context.request_id.clone(), } } pub fn build_domain_event_record( command_id: &str, command_name: &str, context: &BridgeContext, workspace_id: &str, ) -> DomainEventRecord { DomainEventRecord { event_id: format!("evt_{command_id}"), workspace_id: workspace_id.into(), aggregate_type: "workspace".into(), aggregate_id: workspace_id.into(), event_type: format!("{command_name}_requested"), event_version: 1, payload_json: format!( "{{\"request_id\":\"{}\",\"trace_id\":\"{}\",\"command_id\":\"{}\",\"command_name\":\"{}\"}}", context.request_id, context.trace_id, command_id, command_name ), command_log_id: format!("clog_{command_id}"), actor_type: context.actor_type.clone(), created_at: Timestamp::new("2026-04-11T00:00:00Z"), status: EventStatus::Pending, trace_id: context.trace_id.clone(), request_id: context.request_id.clone(), command_id: command_id.into(), } } pub fn build_write_pipeline( context: &BridgeContext, command: &CommandEnvelope, ) -> BridgeResult { let mutation = build_write_request(context, command)?; let workspace_id = mutation .workspace_id .clone() .unwrap_or_else(|| "workspace_unknown".into()); let command_log = build_command_log_record(&command.command_id, &command.name, context, &workspace_id); let domain_event = build_domain_event_record(&command.command_id, &command.name, context, &workspace_id); let result = CommandResult { ok: true, command_id: command.command_id.clone(), event_ids: vec![domain_event.event_id.clone()], affected_objects: vec![AffectedObject { object_type: "workspace".into(), object_id: workspace_id.clone(), }], revision: Some(1), warnings: vec![], }; Ok(WritePipelineResult { mutation, command_log, domain_event, result, }) }