feat: 建立统一检测任务、样本链与发病事件

server-go/internal/service/detection_task_outbox.go

@@ -0,0 +1,82 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+
+	"silk-server-go/internal/model"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+// DetectionTaskCreatePayload 创建待确认检测任务的 outbox 载荷。
+type DetectionTaskCreatePayload struct {
+	SourceKey    string `json:"sourceKey"`
+	SourceType   string `json:"sourceType"`
+	SourceID     string `json:"sourceId"`
+	RoomID       string `json:"roomId"`
+	BatchID      string `json:"batchId"`
+	InspectionID string `json:"inspectionId"`
+	Disease      string `json:"disease"`
+	Priority     string `json:"priority"`
+}
+
+// NewDetectionTaskOutboxHandler 创建检测任务事件处理器。
+func NewDetectionTaskOutboxHandler(db *gorm.DB) EventHandler {
+	return func(ctx context.Context, event model.OutboxEvent) error {
+		if event.EventType != OutboxEventDetectionTaskCreate {
+			return nil
+		}
+		var payload DetectionTaskCreatePayload
+		if err := json.Unmarshal(event.Payload, &payload); err != nil {
+			return err
+		}
+		if payload.SourceKey == "" {
+			return errors.New("detection task sourceKey is required")
+		}
+		disease := payload.Disease
+		if disease == "" {
+			disease = "待确认"
+		}
+		priority := payload.Priority
+		if priority == "" {
+			priority = "routine"
+		}
+		task := model.DetectionTask{
+			SourceKey:    payload.SourceKey,
+			SourceType:   payload.SourceType,
+			SourceID:     payload.SourceID,
+			RoomID:       strPtrOrNil(payload.RoomID),
+			BatchID:      strPtrOrNil(payload.BatchID),
+			InspectionID: strPtrOrNil(payload.InspectionID),
+			Disease:      disease,
+			Priority:     priority,
+			Status:       "pending",
+		}
+		return db.WithContext(ctx).
+			Session(&gorm.Session{SkipDefaultTransaction: true}).
+			Clauses(clause.OnConflict{
+				Columns:   []clause.Column{{Name: "source_key"}},
+				DoNothing: true,
+			}).
+			Create(&task).Error
+	}
+}
+
+// NewOutboxHandler 分发当前已支持的业务事件。
+func NewOutboxHandler(db *gorm.DB, wechat *WechatService) EventHandler {
+	wechatHandler := NewWechatOutboxHandler(db, wechat)
+	detectionHandler := NewDetectionTaskOutboxHandler(db)
+	return func(ctx context.Context, event model.OutboxEvent) error {
+		switch event.EventType {
+		case OutboxEventWechatSubscribe:
+			return wechatHandler(ctx, event)
+		case OutboxEventDetectionTaskCreate:
+			return detectionHandler(ctx, event)
+		default:
+			return nil
+		}
+	}
+}

server-go/internal/service/detection_task_outbox_test.go

@@ -0,0 +1,44 @@
+package service
+
+import (
+	"context"
+	"database/sql/driver"
+	"encoding/json"
+	"testing"
+
+	"github.com/DATA-DOG/go-sqlmock"
+
+	"silk-server-go/internal/model"
+)
+
+func TestDetectionTaskCreateOutboxIsIdempotent(t *testing.T) {
+	o, mock := newOutboxMock(t)
+	handler := NewDetectionTaskOutboxHandler(o.db)
+	payload, _ := json.Marshal(DetectionTaskCreatePayload{
+		SourceKey:    "inspection-inspection-1",
+		SourceType:   "inspection",
+		SourceID:     "inspection-1",
+		InspectionID: "inspection-1",
+		Disease:      "待确认",
+	})
+	event := model.OutboxEvent{
+		EventType: OutboxEventDetectionTaskCreate,
+		Payload:   payload,
+	}
+	args := make([]driver.Value, 0, 20)
+	for i := 0; i < 20; i++ {
+		args = append(args, sqlmock.AnyArg())
+	}
+	mock.ExpectQuery(`INSERT INTO "detection_tasks".*ON CONFLICT \("source_key"\) DO NOTHING`).
+		WithArgs(args...).
+		WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow("task-1"))
+	if err := handler(context.Background(), event); err != nil {
+		t.Fatalf("first create failed: %v", err)
+	}
+	mock.ExpectQuery(`INSERT INTO "detection_tasks".*ON CONFLICT \("source_key"\) DO NOTHING`).
+		WithArgs(args...).
+		WillReturnRows(sqlmock.NewRows([]string{"id"}))
+	if err := handler(context.Background(), event); err != nil {
+		t.Fatalf("duplicate create failed: %v", err)
+	}
+}

server-go/internal/service/outbox.go

@@ -23,7 +23,8 @@ const (
 	OutboxStatusFailed    = "failed"
 	OutboxStatusCancelled = "cancelled"
 
-	OutboxEventWechatSubscribe = "wechat.subscribe"
+	OutboxEventWechatSubscribe     = "wechat.subscribe"
+	OutboxEventDetectionTaskCreate = "detection.task.create"
 )
 
 // Event 写入 outbox 的领域事件。