132 lines
4.7 KiB
Rust
132 lines
4.7 KiB
Rust
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,
|
|
})
|
|
}
|