推进 server-rs DDD 分层与新接口接线

server-rs/crates/spacetime-module/src/ai/events.rs

@@ -0,0 +1,100 @@
+use crate::*;
+
+/// AI 任务事件类型。
+///
+/// 事件表用于给订阅端和 BFF 增量消费状态变化；正式任务真相仍以
+/// `ai_task`、`ai_task_stage`、`ai_text_chunk` 和 `ai_result_reference` 为准。
+#[derive(Clone, Copy, Debug, PartialEq, Eq, SpacetimeType)]
+pub enum AiTaskEventKind {
+    TaskCreated,
+    TaskStatusChanged,
+    StageStarted,
+    StageCompleted,
+    TextChunkAppended,
+    ResultReferenceAttached,
+}
+
+#[spacetimedb::table(
+    accessor = ai_task_event,
+    public,
+    event,
+    index(accessor = by_ai_task_event_task_id, btree(columns = [task_id])),
+    index(accessor = by_ai_task_event_owner_user_id, btree(columns = [owner_user_id]))
+)]
+pub struct AiTaskEvent {
+    #[primary_key]
+    pub(crate) event_id: String,
+    pub(crate) task_id: String,
+    pub(crate) owner_user_id: String,
+    pub(crate) event_kind: AiTaskEventKind,
+    pub(crate) task_status: Option<AiTaskStatus>,
+    pub(crate) stage_kind: Option<AiTaskStageKind>,
+    pub(crate) text_chunk_row_id: Option<String>,
+    pub(crate) result_reference_row_id: Option<String>,
+    pub(crate) occurred_at: Timestamp,
+}
+
+pub(crate) fn emit_ai_task_event(
+    ctx: &ReducerContext,
+    task: &AiTaskSnapshot,
+    event_kind: AiTaskEventKind,
+    stage_kind: Option<AiTaskStageKind>,
+    text_chunk_row_id: Option<String>,
+    result_reference_row_id: Option<String>,
+    occurred_at_micros: i64,
+) {
+    let suffix = match event_kind {
+        AiTaskEventKind::TaskCreated => "created".to_string(),
+        AiTaskEventKind::TaskStatusChanged => format!("status_{}", task.status.as_event_slug()),
+        AiTaskEventKind::StageStarted => {
+            format!("stage_started_{}", stage_kind_slug(stage_kind))
+        }
+        AiTaskEventKind::StageCompleted => {
+            format!("stage_completed_{}", stage_kind_slug(stage_kind))
+        }
+        AiTaskEventKind::TextChunkAppended => {
+            format!(
+                "chunk_{}",
+                text_chunk_row_id.as_deref().unwrap_or("unknown")
+            )
+        }
+        AiTaskEventKind::ResultReferenceAttached => {
+            format!(
+                "result_{}",
+                result_reference_row_id.as_deref().unwrap_or("unknown")
+            )
+        }
+    };
+
+    ctx.db.ai_task_event().insert(AiTaskEvent {
+        event_id: format!("aievt_{}_{}_{}", task.task_id, occurred_at_micros, suffix),
+        task_id: task.task_id.clone(),
+        owner_user_id: task.owner_user_id.clone(),
+        event_kind,
+        task_status: Some(task.status),
+        stage_kind,
+        text_chunk_row_id,
+        result_reference_row_id,
+        occurred_at: Timestamp::from_micros_since_unix_epoch(occurred_at_micros),
+    });
+}
+
+fn stage_kind_slug(stage_kind: Option<AiTaskStageKind>) -> &'static str {
+    stage_kind.map(AiTaskStageKind::as_str).unwrap_or("unknown")
+}
+
+trait AiTaskStatusEventSlug {
+    fn as_event_slug(self) -> &'static str;
+}
+
+impl AiTaskStatusEventSlug for AiTaskStatus {
+    fn as_event_slug(self) -> &'static str {
+        match self {
+            Self::Pending => "pending",
+            Self::Running => "running",
+            Self::Completed => "completed",
+            Self::Failed => "failed",
+            Self::Cancelled => "cancelled",
+        }
+    }
+}

server-rs/crates/spacetime-module/src/ai/mod.rs

@@ -1,7 +1,9 @@
+mod events;
 mod snapshots;
 mod stages;
 mod tasks;
 
+pub(crate) use events::*;
 pub(crate) use snapshots::*;
 pub use stages::*;
 pub use tasks::*;

server-rs/crates/spacetime-module/src/ai/snapshots.rs

@@ -119,13 +119,7 @@ pub(crate) fn build_ai_task_stage_snapshot_from_row(row: &AiTaskStage) -> AiTask
 
 pub(crate) fn build_ai_text_chunk_row(snapshot: &AiTextChunkSnapshot) -> AiTextChunk {
     AiTextChunk {
-        text_chunk_row_id: format!(
-            "{}{}_{}_{}",
-            AI_TEXT_CHUNK_ID_PREFIX,
-            snapshot.task_id,
-            snapshot.stage_kind.as_str(),
-            snapshot.sequence
-        ),
+        text_chunk_row_id: build_ai_text_chunk_row_id(snapshot),
         chunk_id: snapshot.chunk_id.clone(),
         task_id: snapshot.task_id.clone(),
         stage_kind: snapshot.stage_kind,
@@ -135,6 +129,16 @@ pub(crate) fn build_ai_text_chunk_row(snapshot: &AiTextChunkSnapshot) -> AiTextC
     }
 }
 
+pub(crate) fn build_ai_text_chunk_row_id(snapshot: &AiTextChunkSnapshot) -> String {
+    format!(
+        "{}{}_{}_{}",
+        AI_TEXT_CHUNK_ID_PREFIX,
+        snapshot.task_id,
+        snapshot.stage_kind.as_str(),
+        snapshot.sequence
+    )
+}
+
 pub(crate) fn build_ai_text_chunk_snapshot_from_row(row: &AiTextChunk) -> AiTextChunkSnapshot {
     AiTextChunkSnapshot {
         chunk_id: row.chunk_id.clone(),
@@ -150,10 +154,7 @@ pub(crate) fn build_ai_result_reference_row(
     snapshot: &AiResultReferenceSnapshot,
 ) -> AiResultReference {
     AiResultReference {
-        result_reference_row_id: format!(
-            "{}{}_{}",
-            AI_RESULT_REF_ID_PREFIX, snapshot.task_id, snapshot.result_ref_id
-        ),
+        result_reference_row_id: build_ai_result_reference_row_id(snapshot),
         result_ref_id: snapshot.result_ref_id.clone(),
         task_id: snapshot.task_id.clone(),
         reference_kind: snapshot.reference_kind,
@@ -163,6 +164,13 @@ pub(crate) fn build_ai_result_reference_row(
     }
 }
 
+pub(crate) fn build_ai_result_reference_row_id(snapshot: &AiResultReferenceSnapshot) -> String {
+    format!(
+        "{}{}_{}",
+        AI_RESULT_REF_ID_PREFIX, snapshot.task_id, snapshot.result_ref_id
+    )
+}
+
 pub(crate) fn build_ai_result_reference_snapshot_from_row(
     row: &AiResultReference,
 ) -> AiResultReferenceSnapshot {

server-rs/crates/spacetime-module/src/ai/stages.rs

@@ -156,6 +156,15 @@ pub(crate) fn start_ai_task_stage_tx(
     snapshot.version += 1;
 
     persist_ai_task_snapshot(ctx, &snapshot)?;
+    emit_ai_task_event(
+        ctx,
+        &snapshot,
+        AiTaskEventKind::StageStarted,
+        Some(input.stage_kind),
+        None,
+        None,
+        input.started_at_micros,
+    );
     Ok(snapshot)
 }
 
@@ -207,6 +216,15 @@ pub(crate) fn append_ai_text_chunk_tx(
     snapshot.version += 1;
 
     persist_ai_task_snapshot(ctx, &snapshot)?;
+    emit_ai_task_event(
+        ctx,
+        &snapshot,
+        AiTaskEventKind::TextChunkAppended,
+        Some(chunk.stage_kind),
+        Some(build_ai_text_chunk_row_id(&chunk)),
+        None,
+        chunk.created_at_micros,
+    );
     Ok((snapshot, chunk))
 }
 
@@ -235,6 +253,15 @@ pub(crate) fn complete_ai_stage_tx(
     snapshot.version += 1;
 
     persist_ai_task_snapshot(ctx, &snapshot)?;
+    emit_ai_task_event(
+        ctx,
+        &snapshot,
+        AiTaskEventKind::StageCompleted,
+        Some(input.stage_kind),
+        None,
+        None,
+        input.completed_at_micros,
+    );
     Ok(snapshot)
 }
 
@@ -267,6 +294,19 @@ pub(crate) fn attach_ai_result_reference_tx(
     snapshot.version += 1;
 
     persist_ai_task_snapshot(ctx, &snapshot)?;
+    let reference = snapshot
+        .result_references
+        .last()
+        .ok_or_else(|| "ai_result_reference 写入后缺少快照".to_string())?;
+    emit_ai_task_event(
+        ctx,
+        &snapshot,
+        AiTaskEventKind::ResultReferenceAttached,
+        None,
+        None,
+        Some(build_ai_result_reference_row_id(reference)),
+        input.created_at_micros,
+    );
     Ok(snapshot)
 }
 

server-rs/crates/spacetime-module/src/ai/tasks.rs

@@ -135,6 +135,15 @@ fn create_ai_task_tx(
     let task_snapshot = build_ai_task_snapshot_from_create_input(&input);
     ctx.db.ai_task().insert(build_ai_task_row(&task_snapshot));
     replace_ai_task_stages(ctx, &task_snapshot.task_id, &task_snapshot.stages);
+    emit_ai_task_event(
+        ctx,
+        &task_snapshot,
+        AiTaskEventKind::TaskCreated,
+        None,
+        None,
+        None,
+        task_snapshot.created_at_micros,
+    );
 
     get_ai_task_snapshot_tx(ctx, &task_snapshot.task_id)
 }
@@ -154,6 +163,15 @@ fn start_ai_task_tx(
     snapshot.version += 1;
 
     persist_ai_task_snapshot(ctx, &snapshot)?;
+    emit_ai_task_event(
+        ctx,
+        &snapshot,
+        AiTaskEventKind::TaskStatusChanged,
+        None,
+        None,
+        None,
+        input.started_at_micros,
+    );
     Ok(snapshot)
 }
 
@@ -170,6 +188,15 @@ fn complete_ai_task_tx(
     snapshot.version += 1;
 
     persist_ai_task_snapshot(ctx, &snapshot)?;
+    emit_ai_task_event(
+        ctx,
+        &snapshot,
+        AiTaskEventKind::TaskStatusChanged,
+        None,
+        None,
+        None,
+        input.completed_at_micros,
+    );
     Ok(snapshot)
 }
 
@@ -192,6 +219,15 @@ fn fail_ai_task_tx(
     snapshot.version += 1;
 
     persist_ai_task_snapshot(ctx, &snapshot)?;
+    emit_ai_task_event(
+        ctx,
+        &snapshot,
+        AiTaskEventKind::TaskStatusChanged,
+        None,
+        None,
+        None,
+        input.completed_at_micros,
+    );
     Ok(snapshot)
 }
 
@@ -208,6 +244,15 @@ fn cancel_ai_task_tx(
     snapshot.version += 1;
 
     persist_ai_task_snapshot(ctx, &snapshot)?;
+    emit_ai_task_event(
+        ctx,
+        &snapshot,
+        AiTaskEventKind::TaskStatusChanged,
+        None,
+        None,
+        None,
+        input.completed_at_micros,
+    );
     Ok(snapshot)
 }