feat(t237): import freight orders synchronously

2026-07-29 12:11:56 +08:00
parent 4bce1e162c
commit 8d5b88f4a2
18 changed files with 653 additions and 97 deletions
@@ -20,14 +20,19 @@ const (
 	maxFreightItemsPerOrder = 1000
 	maxFreightWindowDays    = 7
 	freightWatermarkOverlap = 10 * time.Minute
+	orderFreightSyncTimeout = 55 * time.Second
+	freightCleanupTimeout   = 5 * time.Second
 )

 type FreightService struct {
-	repository FreightRepository
-	source     FreightSource
-	clock      Clock
-	ids        IDGenerator
-	timeout    time.Duration
+	repository     FreightRepository
+	source         FreightSource
+	clock          Clock
+	ids            IDGenerator
+	timeout        time.Duration
+	orderTimeout   time.Duration
+	cleanupTimeout time.Duration
+	orderSyncGate  chan struct{}
 }

 type FreightSourcePreflight interface {
@@ -67,11 +72,14 @@ func NewFreightService(
 		return nil, errors.New("freight service dependencies are required")
 	}
 	return &FreightService{
-		repository: repository,
-		source:     source,
-		clock:      clock,
-		ids:        ids,
-		timeout:    timeout,
+		repository:     repository,
+		source:         source,
+		clock:          clock,
+		ids:            ids,
+		timeout:        timeout,
+		orderTimeout:   orderFreightSyncTimeout,
+		cleanupTimeout: freightCleanupTimeout,
+		orderSyncGate:  make(chan struct{}, 1),
 	}, nil
 }

@@ -105,7 +113,16 @@ func (service *FreightService) CreateOrderSync(
 			fields,
 		)
 	}
-	if err := service.ensureSource(ctx); err != nil {
+	if !service.acquireOrderSync() {
+		return CreateFreightSyncResult{}, freightSyncBusyError()
+	}
+	defer service.releaseOrderSync()
+	syncCtx, cancel := context.WithTimeout(ctx, service.orderTimeout)
+	defer cancel()
+	if err := service.ensureSource(syncCtx); err != nil {
+		if syncErr := freightContextError(syncCtx); syncErr != nil {
+			return CreateFreightSyncResult{}, syncErr
+		}
 		return CreateFreightSyncResult{}, err
 	}
 	runID, err := service.ids.NewID()
@@ -121,7 +138,7 @@ func (service *FreightService) CreateOrderSync(
 		OrderNumber string `json:"order_number"`
 	}{command.OrderNumber})
 	run, created, err := service.repository.CreateFreightSync(
-		ctx,
+		syncCtx,
 		domain.FreightSyncRun{
 			ID:              runID,
 			CreatorSubject:  command.CreatorSubject,
@@ -138,10 +155,31 @@ func (service *FreightService) CreateOrderSync(
 	if err != nil {
 		return CreateFreightSyncResult{}, wrapRepositoryError(err)
 	}
-	if created {
-		go service.execute(run)
+	if !created {
+		if run.Status != domain.FreightSyncSucceeded &&
+			run.Status != domain.FreightSyncFailed {
+			return CreateFreightSyncResult{}, freightSyncBusyError()
+		}
+		return CreateFreightSyncResult{Run: run, Replayed: true}, nil
 	}
-	return CreateFreightSyncResult{Run: run, Replayed: !created}, nil
+	completed, err := service.executeRun(syncCtx, run)
+	if err != nil {
+		return CreateFreightSyncResult{}, err
+	}
+	return CreateFreightSyncResult{Run: completed}, nil
+}
+
+func (service *FreightService) acquireOrderSync() bool {
+	select {
+	case service.orderSyncGate <- struct{}{}:
+		return true
+	default:
+		return false
+	}
+}
+
+func (service *FreightService) releaseOrderSync() {
+	<-service.orderSyncGate
 }

 func (service *FreightService) CreateDateSync(
@@ -310,10 +348,24 @@ func (service *FreightService) ensureSource(ctx context.Context) error {
 func (service *FreightService) execute(run domain.FreightSyncRun) {
 	ctx, cancel := context.WithTimeout(context.Background(), service.timeout)
 	defer cancel()
+	_, _ = service.executeRun(ctx, run)
+}
+
+func (service *FreightService) executeRun(
+	ctx context.Context,
+	run domain.FreightSyncRun,
+) (domain.FreightSyncRun, error) {
 	now := service.clock.Now().UTC()
 	if err := service.repository.StartFreightSync(ctx, run.ID, now); err != nil {
-		return
+		code := freightFailureCode(ctx, "STORAGE_UNAVAILABLE")
+		return domain.FreightSyncRun{}, service.finishFreightFailure(
+			run.ID,
+			code,
+			wrapRepositoryError(err),
+		)
 	}
+	run.Status = domain.FreightSyncRunning
+	run.StartedAt = &now
 	var source domain.FreightSourceBatch
 	var err error
 	if run.Mode == domain.FreightSyncCreatedRange {
@@ -322,23 +374,21 @@ func (service *FreightService) execute(run domain.FreightSyncRun) {
 		source, err = service.source.QueryOrder(ctx, run.OrderNumber)
 	}
 	if err != nil {
-		_ = service.repository.FailFreightSync(
-			ctx,
+		code := freightFailureCode(ctx, freightSourceErrorCode(err))
+		return domain.FreightSyncRun{}, service.finishFreightFailure(
 			run.ID,
-			freightSourceErrorCode(err),
-			service.clock.Now().UTC(),
+			code,
+			freightExecutionError(code, err),
 		)
-		return
 	}
 	batch, err := service.normalize(source)
 	if err != nil {
-		_ = service.repository.FailFreightSync(
-			ctx,
+		code := freightFailureCode(ctx, "ERP_RESPONSE_INVALID")
+		return domain.FreightSyncRun{}, service.finishFreightFailure(
 			run.ID,
-			"ERP_RESPONSE_INVALID",
-			service.clock.Now().UTC(),
+			code,
+			freightExecutionError(code, domain.ErrFreightSourceProtocol),
 		)
-		return
 	}
 	finishedAt := service.clock.Now().UTC()
 	if run.Mode == domain.FreightSyncCreatedRange &&
@@ -359,13 +409,107 @@ func (service *FreightService) execute(run domain.FreightSyncRun) {
 		)
 	}
 	if err != nil {
-		_ = service.repository.FailFreightSync(
-			ctx,
+		code := freightFailureCode(ctx, "STORAGE_UNAVAILABLE")
+		return domain.FreightSyncRun{}, service.finishFreightFailure(
 			run.ID,
-			"STORAGE_UNAVAILABLE",
-			service.clock.Now().UTC(),
+			code,
+			wrapRepositoryError(err),
 		)
 	}
+	itemCount := 0
+	for _, order := range batch.Orders {
+		itemCount += len(order.Items)
+	}
+	run.Status = domain.FreightSyncSucceeded
+	run.OrderCount = len(batch.Orders)
+	run.ItemCount = itemCount
+	run.FinishedAt = &finishedAt
+	return run, nil
+}
+
+func (service *FreightService) finishFreightFailure(
+	runID, code string,
+	cause error,
+) error {
+	timeout := service.cleanupTimeout
+	if timeout <= 0 {
+		timeout = freightCleanupTimeout
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	if err := service.repository.FailFreightSync(
+		ctx,
+		runID,
+		code,
+		service.clock.Now().UTC(),
+	); err != nil {
+		return wrapRepositoryError(err)
+	}
+	return cause
+}
+
+func freightFailureCode(ctx context.Context, fallback string) string {
+	switch ctx.Err() {
+	case context.DeadlineExceeded:
+		return "FREIGHT_SYNC_TIMEOUT"
+	case context.Canceled:
+		return "FREIGHT_SYNC_CANCELED"
+	default:
+		return fallback
+	}
+}
+
+func freightContextError(ctx context.Context) error {
+	switch ctx.Err() {
+	case context.DeadlineExceeded:
+		return freightExecutionError(
+			"FREIGHT_SYNC_TIMEOUT",
+			context.DeadlineExceeded,
+		)
+	case context.Canceled:
+		return freightExecutionError(
+			"FREIGHT_SYNC_CANCELED",
+			context.Canceled,
+		)
+	default:
+		return nil
+	}
+}
+
+func freightSyncBusyError() error {
+	result := newError(
+		ErrorKindConflict,
+		"FREIGHT_SYNC_BUSY",
+		"another freight order sync is already running",
+		nil,
+	)
+	result.Retryable = true
+	return result
+}
+
+func freightExecutionError(code string, cause error) error {
+	var kind ErrorKind
+	var message string
+	switch code {
+	case "FREIGHT_SYNC_TIMEOUT":
+		kind = ErrorKindUnavailable
+		message = "freight order sync exceeded its time limit"
+	case "FREIGHT_SYNC_CANCELED":
+		kind = ErrorKindUnavailable
+		message = "freight order sync was canceled"
+	case "ERP_FREIGHT_NOT_FOUND":
+		kind = ErrorKindNotFound
+		message = "ERP freight order was not found"
+	default:
+		kind = ErrorKindUnavailable
+		message = freightPreflightMessage(code)
+	}
+	result := newError(kind, code, message, cause)
+	result.Retryable = code == "FREIGHT_SYNC_TIMEOUT" ||
+		code == "FREIGHT_SYNC_CANCELED" ||
+		code == "ERP_UNAVAILABLE" ||
+		code == "OCR_SERVICE_INVALID"
+	return result
 }

 func (service *FreightService) queryCreatedRange(