mnote/rust/crates/storage-convex-bridge/src/write_path.rs

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<T>(
    context: &BridgeContext,
    command: &CommandEnvelope<T>,
) -> BridgeResult<ConvexMutationRequest> {
    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<T>(
    context: &BridgeContext,
    command: &CommandEnvelope<T>,
) -> BridgeResult<WritePipelineResult> {
    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,
    })
}