cmroubao/backend-api/internal/usecase/freight_service.go

package usecase

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"sort"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"

	"cmroubao/backend-api/internal/domain"
)

const (
	maxFreightOrdersPerSync = 100
	maxFreightItemsPerOrder = 1000
	maxFreightWindowDays    = 7
	freightWatermarkOverlap = 10 * time.Minute
	orderFreightSyncTimeout = 55 * time.Second
	freightCleanupTimeout   = 5 * time.Second
	freightImageCacheBudget = 15 * time.Second
)

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

type FreightServiceOption func(*FreightService) error

func WithFreightImageCache(cache FreightImageCache) FreightServiceOption {
	return func(service *FreightService) error {
		if cache == nil {
			return errors.New("freight image cache is required")
		}
		service.imageCache = cache
		return nil
	}
}

type FreightSourcePreflight interface {
	EnsureAuthenticated(context.Context) error
}

type CreateFreightSyncCommand struct {
	CreatorSubject string
	ActorUserID    string
	IdempotencyKey string
	OrderNumber    string
}

type CreateFreightDateSyncCommand struct {
	CreatorSubject string
	ActorUserID    string
	IdempotencyKey string
	CreatedFrom    string
	CreatedTo      string
	SyncToNow      bool
}

type DeleteFreightOrderCommand struct {
	CreatorSubject string
	ActorUserID    string
	OrderID        string
}

type CreateFreightSyncResult struct {
	Run      domain.FreightSyncRun
	Replayed bool
}

func NewFreightService(
	repository FreightRepository,
	source FreightSource,
	clock Clock,
	ids IDGenerator,
	timeout time.Duration,
	options ...FreightServiceOption,
) (*FreightService, error) {
	if repository == nil || source == nil || clock == nil || ids == nil ||
		timeout <= 0 {
		return nil, errors.New("freight service dependencies are required")
	}
	service := &FreightService{
		repository:     repository,
		source:         source,
		clock:          clock,
		ids:            ids,
		timeout:        timeout,
		orderTimeout:   orderFreightSyncTimeout,
		cleanupTimeout: freightCleanupTimeout,
		orderSyncGate:  make(chan struct{}, 1),
	}
	for _, option := range options {
		if option == nil {
			return nil, errors.New("freight service option is required")
		}
		if err := option(service); err != nil {
			return nil, err
		}
	}
	return service, nil
}

func (service *FreightService) CreateOrderSync(
	ctx context.Context,
	command CreateFreightSyncCommand,
) (CreateFreightSyncResult, error) {
	command.CreatorSubject = strings.TrimSpace(command.CreatorSubject)
	command.ActorUserID = strings.TrimSpace(command.ActorUserID)
	command.IdempotencyKey = strings.TrimSpace(command.IdempotencyKey)
	command.OrderNumber = strings.TrimSpace(command.OrderNumber)
	fields := map[string]string{}
	if command.CreatorSubject == "" {
		fields["creator_subject"] = "is required"
	}
	if command.ActorUserID == "" {
		fields["actor_user_id"] = "is required"
	}
	if command.IdempotencyKey == "" || len([]byte(command.IdempotencyKey)) > 128 {
		fields["idempotency_key"] = "must contain 1 to 128 bytes"
	}
	if command.OrderNumber == "" ||
		len([]byte(command.OrderNumber)) > 128 ||
		hasControl(command.OrderNumber) {
		fields["order_number"] = "must contain 1 to 128 bytes without control characters"
	}
	if len(fields) > 0 {
		return CreateFreightSyncResult{}, invalidError(
			"FREIGHT_SYNC_INVALID",
			"freight sync request is invalid",
			fields,
		)
	}
	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()
	if err != nil {
		return CreateFreightSyncResult{}, wrapRepositoryError(err)
	}
	now := service.clock.Now().UTC()
	queryHash := hashJSON(struct {
		Mode        string `json:"mode"`
		OrderNumber string `json:"order_number"`
	}{domain.FreightSyncOrderNumber, command.OrderNumber})
	requestHash := hashJSON(struct {
		OrderNumber string `json:"order_number"`
	}{command.OrderNumber})
	run, created, err := service.repository.CreateFreightSync(
		syncCtx,
		domain.FreightSyncRun{
			ID:              runID,
			CreatorSubject:  command.CreatorSubject,
			CreatedByUserID: command.ActorUserID,
			Mode:            domain.FreightSyncOrderNumber,
			OrderNumber:     command.OrderNumber,
			QuerySHA256:     queryHash,
			Status:          domain.FreightSyncPending,
			CreatedAt:       now,
		},
		command.IdempotencyKey,
		requestHash,
	)
	if err != nil {
		return CreateFreightSyncResult{}, wrapRepositoryError(err)
	}
	if !created {
		if run.Status != domain.FreightSyncSucceeded &&
			run.Status != domain.FreightSyncFailed {
			return CreateFreightSyncResult{}, freightSyncBusyError()
		}
		return CreateFreightSyncResult{Run: run, Replayed: true}, 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(
	ctx context.Context,
	command CreateFreightDateSyncCommand,
) (CreateFreightSyncResult, error) {
	command.CreatorSubject = strings.TrimSpace(command.CreatorSubject)
	command.ActorUserID = strings.TrimSpace(command.ActorUserID)
	command.IdempotencyKey = strings.TrimSpace(command.IdempotencyKey)
	command.CreatedFrom = strings.TrimSpace(command.CreatedFrom)
	command.CreatedTo = strings.TrimSpace(command.CreatedTo)
	fields := validateFreightActor(
		command.CreatorSubject,
		command.ActorUserID,
		command.IdempotencyKey,
	)
	if command.SyncToNow &&
		(command.CreatedFrom != "" || command.CreatedTo != "") {
		fields["sync_to_now"] = "cannot be combined with a manual range"
	}
	if !command.SyncToNow &&
		(command.CreatedFrom == "" || command.CreatedTo == "") {
		fields["created_range"] = "created_from and created_to are required"
	}
	if len(fields) > 0 {
		return CreateFreightSyncResult{}, invalidError(
			"FREIGHT_SYNC_INVALID",
			"freight sync request is invalid",
			fields,
		)
	}
	if err := service.ensureSource(ctx); err != nil {
		return CreateFreightSyncResult{}, err
	}

	now := service.clock.Now().UTC()
	location, err := time.LoadLocation("Asia/Shanghai")
	if err != nil {
		return CreateFreightSyncResult{}, wrapRepositoryError(err)
	}
	var startDate, endDate time.Time
	var watermarkThrough time.Time
	if command.SyncToNow {
		watermark, err := service.repository.GetFreightSyncWatermark(
			ctx,
			command.CreatorSubject,
		)
		if err != nil {
			return CreateFreightSyncResult{}, wrapRepositoryError(err)
		}
		localNow := now.In(location)
		endDate = startOfLocalDay(localNow, location)
		startDate = endDate
		if watermark != nil {
			overlap := watermark.LastSuccessfulTo.Add(
				-freightWatermarkOverlap,
			).In(location)
			startDate = startOfLocalDay(overlap, location)
			if startDate.After(endDate) {
				startDate = endDate
			}
		}
		watermarkThrough = now
	} else {
		startDate, err = parseFreightDate(command.CreatedFrom, location)
		if err != nil {
			fields["created_from"] = "must be YYYY-MM-DD"
		}
		endDate, err = parseFreightDate(command.CreatedTo, location)
		if err != nil {
			fields["created_to"] = "must be YYYY-MM-DD"
		}
		if len(fields) == 0 {
			if endDate.Before(startDate) {
				fields["created_to"] = "must not be before created_from"
			} else if daysInclusive(startDate, endDate) >
				maxFreightWindowDays {
				fields["created_range"] = "must not exceed 7 inclusive days"
			}
			if endDate.After(startOfLocalDay(now.In(location), location)) {
				fields["created_to"] = "must not be in the future"
			}
		}
		if len(fields) > 0 {
			return CreateFreightSyncResult{}, invalidError(
				"FREIGHT_SYNC_INVALID",
				"freight sync request is invalid",
				fields,
			)
		}
		watermarkThrough = endDate.AddDate(0, 0, 1).Add(-time.Nanosecond).UTC()
		if watermarkThrough.After(now) {
			watermarkThrough = now
		}
	}

	createdFrom := startDate.Format(time.DateOnly)
	createdTo := endDate.Format(time.DateOnly)
	runID, err := service.ids.NewID()
	if err != nil {
		return CreateFreightSyncResult{}, wrapRepositoryError(err)
	}
	queryHash := hashJSON(struct {
		Mode        string `json:"mode"`
		CreatedFrom string `json:"created_from"`
		CreatedTo   string `json:"created_to"`
	}{
		domain.FreightSyncCreatedRange,
		createdFrom,
		createdTo,
	})
	requestHash := hashJSON(struct {
		SyncToNow   bool   `json:"sync_to_now"`
		CreatedFrom string `json:"created_from"`
		CreatedTo   string `json:"created_to"`
	}{
		command.SyncToNow,
		command.CreatedFrom,
		command.CreatedTo,
	})
	run, created, err := service.repository.CreateFreightSync(
		ctx,
		domain.FreightSyncRun{
			ID:               runID,
			CreatorSubject:   command.CreatorSubject,
			CreatedByUserID:  command.ActorUserID,
			Mode:             domain.FreightSyncCreatedRange,
			CreatedFrom:      createdFrom,
			CreatedTo:        createdTo,
			WatermarkThrough: &watermarkThrough,
			QuerySHA256:      queryHash,
			Status:           domain.FreightSyncPending,
			CreatedAt:        now,
		},
		command.IdempotencyKey,
		requestHash,
	)
	if err != nil {
		return CreateFreightSyncResult{}, wrapRepositoryError(err)
	}
	if created {
		go service.execute(run)
	}
	return CreateFreightSyncResult{Run: run, Replayed: !created}, nil
}

func (service *FreightService) ensureSource(ctx context.Context) error {
	preflight, ok := service.source.(FreightSourcePreflight)
	if !ok {
		return nil
	}
	if err := preflight.EnsureAuthenticated(ctx); err != nil {
		code := freightSourceErrorCode(err)
		result := newError(
			ErrorKindUnavailable,
			code,
			freightPreflightMessage(code),
			err,
		)
		result.Retryable = code == "OCR_SERVICE_INVALID" || code == "ERP_UNAVAILABLE"
		return result
	}
	return nil
}

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 {
		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 {
		source, err = service.queryCreatedRange(ctx, run)
	} else {
		source, err = service.source.QueryOrder(ctx, run.OrderNumber)
	}
	if err != nil {
		code := freightFailureCode(ctx, freightSourceErrorCode(err))
		return domain.FreightSyncRun{}, service.finishFreightFailure(
			run.ID,
			code,
			freightExecutionError(code, err),
		)
	}
	batch, err := service.normalize(source)
	if err != nil {
		code := freightFailureCode(ctx, "ERP_RESPONSE_INVALID")
		return domain.FreightSyncRun{}, service.finishFreightFailure(
			run.ID,
			code,
			freightExecutionError(code, domain.ErrFreightSourceProtocol),
		)
	}
	finishedAt := service.clock.Now().UTC()
	if run.Mode == domain.FreightSyncCreatedRange &&
		run.WatermarkThrough != nil {
		err = service.repository.CompleteFreightDateSync(
			ctx,
			run,
			batch,
			*run.WatermarkThrough,
			finishedAt,
		)
	} else {
		err = service.repository.CompleteFreightSync(
			ctx,
			run,
			batch,
			finishedAt,
		)
	}
	if err != nil {
		code := freightFailureCode(ctx, "STORAGE_UNAVAILABLE")
		return domain.FreightSyncRun{}, service.finishFreightFailure(
			run.ID,
			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
	if service.imageCache != nil {
		cacheCtx, cancel := context.WithTimeout(
			ctx,
			freightImageCacheBudget,
		)
		_ = service.imageCache.CacheRun(
			cacheCtx,
			run.CreatorSubject,
			run.ID,
		)
		cancel()
	}
	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(
	ctx context.Context,
	run domain.FreightSyncRun,
) (domain.FreightSourceBatch, error) {
	location, err := time.LoadLocation("Asia/Shanghai")
	if err != nil {
		return domain.FreightSourceBatch{}, err
	}
	start, err := parseFreightDate(run.CreatedFrom, location)
	if err != nil {
		return domain.FreightSourceBatch{}, domain.ErrFreightSourceProtocol
	}
	end, err := parseFreightDate(run.CreatedTo, location)
	if err != nil || end.Before(start) {
		return domain.FreightSourceBatch{}, domain.ErrFreightSourceProtocol
	}
	orders := make([]domain.FreightSourceOrder, 0)
	seen := make(map[string]domain.FreightSourceOrder)
	for windowStart := start; !windowStart.After(end); {
		windowEnd := windowStart.AddDate(0, 0, maxFreightWindowDays-1)
		if windowEnd.After(end) {
			windowEnd = end
		}
		fromValue := windowStart.Format(time.DateOnly)
		toValue := windowEnd.Format(time.DateOnly)
		batch, err := service.source.QueryCreatedRange(
			ctx,
			fromValue,
			toValue,
		)
		if err != nil {
			return domain.FreightSourceBatch{}, err
		}
		if batch.SchemaVersion != 1 ||
			batch.Query.Mode != domain.FreightSyncCreatedRange ||
			batch.Query.CreatedFrom == nil ||
			batch.Query.CreatedTo == nil ||
			*batch.Query.CreatedFrom != fromValue ||
			*batch.Query.CreatedTo != toValue {
			return domain.FreightSourceBatch{}, domain.ErrFreightSourceProtocol
		}
		for _, order := range batch.Orders {
			existing, exists := seen[order.ExternalStockID]
			if exists {
				if hashJSON(existing) != hashJSON(order) {
					return domain.FreightSourceBatch{}, domain.ErrFreightSourceProtocol
				}
				continue
			}
			seen[order.ExternalStockID] = order
			orders = append(orders, order)
			if len(orders) > maxFreightOrdersPerSync {
				return domain.FreightSourceBatch{}, domain.ErrFreightSourceProtocol
			}
		}
		windowStart = windowEnd.AddDate(0, 0, 1)
	}
	createdFrom := run.CreatedFrom
	createdTo := run.CreatedTo
	return domain.FreightSourceBatch{
		SchemaVersion: 1,
		Query: domain.FreightSourceQuery{
			Mode:        domain.FreightSyncCreatedRange,
			CreatedFrom: &createdFrom,
			CreatedTo:   &createdTo,
		},
		Orders: orders,
	}, nil
}

func (service *FreightService) RecoverInterrupted(
	ctx context.Context,
) (int64, error) {
	count, err := service.repository.RecoverFreightSyncs(
		ctx,
		service.clock.Now().UTC(),
	)
	if err != nil {
		return 0, wrapRepositoryError(err)
	}
	return count, nil
}

func (service *FreightService) GetSync(
	ctx context.Context,
	creatorSubject, syncID string,
) (domain.FreightSyncRun, error) {
	run, err := service.repository.GetFreightSync(
		ctx,
		strings.TrimSpace(creatorSubject),
		strings.TrimSpace(syncID),
	)
	if err != nil {
		return domain.FreightSyncRun{}, wrapRepositoryError(err)
	}
	return run, nil
}

func (service *FreightService) GetWatermark(
	ctx context.Context,
	creatorSubject string,
) (*domain.FreightSyncWatermark, error) {
	watermark, err := service.repository.GetFreightSyncWatermark(
		ctx,
		strings.TrimSpace(creatorSubject),
	)
	if err != nil {
		return nil, wrapRepositoryError(err)
	}
	return watermark, nil
}

func (service *FreightService) ListOrders(
	ctx context.Context,
	creatorSubject string,
	limit int,
) ([]domain.FreightOrder, error) {
	if limit == 0 {
		limit = 50
	}
	if limit < 1 || limit > 100 {
		return nil, invalidError(
			"FREIGHT_LIST_INVALID",
			"freight list filter is invalid",
			map[string]string{"limit": "must be between 1 and 100"},
		)
	}
	orders, err := service.repository.ListFreightOrders(
		ctx,
		strings.TrimSpace(creatorSubject),
		limit,
	)
	if err != nil {
		return nil, wrapRepositoryError(err)
	}
	return orders, nil
}

func (service *FreightService) GetOrder(
	ctx context.Context,
	creatorSubject, orderID string,
) (domain.FreightOrderDetail, error) {
	detail, err := service.repository.GetFreightOrder(
		ctx,
		strings.TrimSpace(creatorSubject),
		strings.TrimSpace(orderID),
	)
	if err != nil {
		return domain.FreightOrderDetail{}, wrapRepositoryError(err)
	}
	return detail, nil
}

func (service *FreightService) DeleteOrder(
	ctx context.Context,
	command DeleteFreightOrderCommand,
) error {
	command.CreatorSubject = strings.TrimSpace(command.CreatorSubject)
	command.ActorUserID = strings.TrimSpace(command.ActorUserID)
	command.OrderID = strings.TrimSpace(command.OrderID)
	fields := map[string]string{}
	if command.CreatorSubject == "" {
		fields["creator_subject"] = "is required"
	}
	if !isUUID(command.ActorUserID) {
		fields["actor_user_id"] = "must be a UUID"
	}
	if !isUUID(command.OrderID) {
		fields["order_id"] = "must be a UUID"
	}
	if len(fields) > 0 {
		return invalidError(
			"FREIGHT_DELETE_INVALID",
			"freight order deletion is invalid",
			fields,
		)
	}
	storageKeys, err := service.repository.DeleteFreightOrder(
		ctx,
		command.CreatorSubject,
		command.OrderID,
	)
	if err != nil {
		return wrapRepositoryError(err)
	}
	if service.imageCache != nil {
		service.imageCache.DeleteStoredImages(storageKeys)
	}
	return nil
}

func (service *FreightService) normalize(
	source domain.FreightSourceBatch,
) (domain.FreightImportBatch, error) {
	if source.SchemaVersion != 1 ||
		(source.Query.Mode != domain.FreightSyncOrderNumber &&
			source.Query.Mode != domain.FreightSyncCreatedRange) ||
		len(source.Orders) > maxFreightOrdersPerSync {
		return domain.FreightImportBatch{}, errors.New("invalid source envelope")
	}
	seenOrders := map[string]struct{}{}
	result := domain.FreightImportBatch{
		Orders: make([]domain.FreightImportOrder, 0, len(source.Orders)),
	}
	for _, sourceOrder := range source.Orders {
		externalID, ok := validExternalID(sourceOrder.ExternalStockID)
		if !ok || len(sourceOrder.Items) > maxFreightItemsPerOrder {
			return domain.FreightImportBatch{}, errors.New("invalid freight order")
		}
		if _, exists := seenOrders[externalID]; exists {
			return domain.FreightImportBatch{}, errors.New("duplicate freight order")
		}
		seenOrders[externalID] = struct{}{}
		if !validBytes(sourceOrder.SourceCode, 256) ||
			!validOptional(sourceOrder.PlatformOrderNo, 256) ||
			!validOptional(sourceOrder.ShopName, 512) ||
			!validOptional(sourceOrder.OrderStatus, 128) ||
			!validOptional(sourceOrder.PurchaseStatus, 128) {
			return domain.FreightImportBatch{}, errors.New("invalid freight fields")
		}
		sourceCreatedAt, err := parseERPTime(sourceOrder.SourceCreatedAt)
		if err != nil {
			return domain.FreightImportBatch{}, err
		}
		orderID, err := service.ids.NewID()
		if err != nil {
			return domain.FreightImportBatch{}, err
		}
		order := domain.FreightImportOrder{
			ID:              orderID,
			ExternalStockID: externalID,
			SourceCode:      sourceOrder.SourceCode,
			PlatformOrderNo: cleanOptional(sourceOrder.PlatformOrderNo),
			ShopName:        cleanOptional(sourceOrder.ShopName),
			SourceCreatedAt: sourceCreatedAt,
			OrderStatus:     cleanOptional(sourceOrder.OrderStatus),
			PurchaseStatus:  cleanOptional(sourceOrder.PurchaseStatus),
			IsCanceled:      sourceOrder.IsCanceled,
			Items:           make([]domain.FreightImportItem, 0, len(sourceOrder.Items)),
		}
		seenItems := map[string]struct{}{}
		for _, sourceItem := range sourceOrder.Items {
			itemExternalID, ok := validExternalID(sourceItem.ExternalItemID)
			if !ok {
				return domain.FreightImportBatch{}, errors.New("invalid freight item")
			}
			if _, exists := seenItems[itemExternalID]; exists {
				return domain.FreightImportBatch{}, errors.New("duplicate freight item")
			}
			seenItems[itemExternalID] = struct{}{}
			if !validBytes(sourceItem.Title, 2048) ||
				!validBytes(sourceItem.ProductSpec, 1024) ||
				!validBytes(sourceItem.SKU, 512) ||
				!validOptional(sourceItem.ProductThumbRef, 512) ||
				sourceItem.OriginalCurrency != domain.FreightCurrencyTWD ||
				(sourceItem.OriginalUnitPriceMinor != nil &&
					*sourceItem.OriginalUnitPriceMinor < 0) ||
				!validOptional(sourceItem.PurchaseStatus, 128) ||
				(sourceItem.Quantity != nil && *sourceItem.Quantity <= 0) {
				return domain.FreightImportBatch{}, errors.New("invalid freight item fields")
			}
			itemID, err := service.ids.NewID()
			if err != nil {
				return domain.FreightImportBatch{}, err
			}
			item := domain.FreightImportItem{
				ID:                     itemID,
				ExternalItemID:         itemExternalID,
				Title:                  sourceItem.Title,
				ProductSpec:            sourceItem.ProductSpec,
				SKU:                    sourceItem.SKU,
				Quantity:               sourceItem.Quantity,
				ProductThumbRef:        cleanOptional(sourceItem.ProductThumbRef),
				OriginalUnitPriceMinor: sourceItem.OriginalUnitPriceMinor,
				OriginalCurrency:       sourceItem.OriginalCurrency,
				PurchaseStatus:         cleanOptional(sourceItem.PurchaseStatus),
			}
			item.CanonicalSHA256 = hashJSON(struct {
				ExternalItemID         string  `json:"external_item_id"`
				Title                  string  `json:"title"`
				ProductSpec            string  `json:"product_spec"`
				SKU                    string  `json:"sku"`
				Quantity               *int    `json:"quantity"`
				ProductThumbRef        *string `json:"product_thumb_ref"`
				OriginalUnitPriceMinor *int64  `json:"original_unit_price_minor"`
				OriginalCurrency       string  `json:"original_currency"`
				PurchaseStatus         *string `json:"purchase_status"`
			}{
				item.ExternalItemID, item.Title, item.ProductSpec, item.SKU,
				item.Quantity, item.ProductThumbRef,
				item.OriginalUnitPriceMinor, item.OriginalCurrency,
				item.PurchaseStatus,
			})
			order.Items = append(order.Items, item)
		}
		sort.Slice(order.Items, func(i, j int) bool {
			left, _ := strconv.ParseUint(order.Items[i].ExternalItemID, 10, 64)
			right, _ := strconv.ParseUint(order.Items[j].ExternalItemID, 10, 64)
			return left < right
		})
		type canonicalItem struct {
			ExternalItemID  string `json:"external_item_id"`
			CanonicalSHA256 string `json:"canonical_sha256"`
		}
		canonicalItems := make([]canonicalItem, 0, len(order.Items))
		for _, item := range order.Items {
			canonicalItems = append(canonicalItems, canonicalItem{
				ExternalItemID:  item.ExternalItemID,
				CanonicalSHA256: item.CanonicalSHA256,
			})
		}
		order.CanonicalSHA256 = hashJSON(struct {
			ExternalStockID string          `json:"external_stock_id"`
			SourceCode      string          `json:"source_code"`
			PlatformOrderNo *string         `json:"platform_order_no"`
			ShopName        *string         `json:"shop_name"`
			SourceCreatedAt *time.Time      `json:"source_created_at"`
			OrderStatus     *string         `json:"order_status"`
			PurchaseStatus  *string         `json:"purchase_status"`
			IsCanceled      *bool           `json:"is_canceled"`
			Items           []canonicalItem `json:"items"`
		}{
			order.ExternalStockID, order.SourceCode, order.PlatformOrderNo,
			order.ShopName, order.SourceCreatedAt, order.OrderStatus,
			order.PurchaseStatus, order.IsCanceled, canonicalItems,
		})
		result.Orders = append(result.Orders, order)
	}
	sort.Slice(result.Orders, func(i, j int) bool {
		left, _ := strconv.ParseUint(result.Orders[i].ExternalStockID, 10, 64)
		right, _ := strconv.ParseUint(result.Orders[j].ExternalStockID, 10, 64)
		return left < right
	})
	return result, nil
}

func validateFreightActor(
	creatorSubject, actorUserID, idempotencyKey string,
) map[string]string {
	fields := map[string]string{}
	if creatorSubject == "" {
		fields["creator_subject"] = "is required"
	}
	if actorUserID == "" {
		fields["actor_user_id"] = "is required"
	}
	if idempotencyKey == "" || len([]byte(idempotencyKey)) > 128 {
		fields["idempotency_key"] = "must contain 1 to 128 bytes"
	}
	return fields
}

func parseFreightDate(value string, location *time.Location) (time.Time, error) {
	parsed, err := time.ParseInLocation(time.DateOnly, value, location)
	if err != nil || parsed.Format(time.DateOnly) != value {
		return time.Time{}, errors.New("invalid freight date")
	}
	return parsed, nil
}

func startOfLocalDay(value time.Time, location *time.Location) time.Time {
	return time.Date(
		value.Year(),
		value.Month(),
		value.Day(),
		0,
		0,
		0,
		0,
		location,
	)
}

func daysInclusive(start, end time.Time) int {
	return int(end.Sub(start)/(24*time.Hour)) + 1
}

func freightSourceErrorCode(err error) string {
	switch {
	case errors.Is(err, domain.ErrFreightSourceNotConfigured):
		return "ERP_NOT_CONFIGURED"
	case errors.Is(err, domain.ErrFreightSourceSessionNeeded):
		return "ERP_SESSION_REQUIRED"
	case errors.Is(err, domain.ErrFreightSourceNotFound):
		return "ERP_FREIGHT_NOT_FOUND"
	case errors.Is(err, domain.ErrFreightSourceProtocol):
		return "ERP_RESPONSE_INVALID"
	case errors.Is(err, domain.ErrFreightSourceOCRInvalid):
		return "OCR_SERVICE_INVALID"
	case errors.Is(err, domain.ErrFreightSourceLoginRejected):
		return "ERP_LOGIN_REJECTED"
	default:
		return "ERP_UNAVAILABLE"
	}
}

func freightPreflightMessage(code string) string {
	switch code {
	case "ERP_NOT_CONFIGURED":
		return "ERP credentials are not configured"
	case "ERP_LOGIN_REJECTED":
		return "ERP login was rejected"
	case "ERP_RESPONSE_INVALID":
		return "ERP response is invalid"
	case "OCR_SERVICE_INVALID":
		return "OCR service is invalid"
	default:
		return "ERP is temporarily unavailable"
	}
}

func validExternalID(value string) (string, bool) {
	value = strings.TrimSpace(value)
	number, err := strconv.ParseUint(value, 10, 64)
	return value, err == nil && number > 0 && strconv.FormatUint(number, 10) == value
}

func validBytes(value string, maximum int) bool {
	return utf8.ValidString(value) && len([]byte(value)) <= maximum &&
		!hasControl(value)
}

func validOptional(value *string, maximum int) bool {
	return value == nil || validBytes(strings.TrimSpace(*value), maximum)
}

func cleanOptional(value *string) *string {
	if value == nil {
		return nil
	}
	trimmed := strings.TrimSpace(*value)
	if trimmed == "" {
		return nil
	}
	return &trimmed
}

func hasControl(value string) bool {
	for _, character := range value {
		if character < 0x20 || character == 0x7f {
			return true
		}
	}
	return false
}

func parseERPTime(value *string) (*time.Time, error) {
	value = cleanOptional(value)
	if value == nil {
		return nil, nil
	}
	location, err := time.LoadLocation("Asia/Shanghai")
	if err != nil {
		return nil, err
	}
	layouts := []string{
		time.RFC3339Nano,
		"2006-01-02 15:04:05",
		"2006-01-02T15:04:05",
		"2006-01-02 15:04",
		"2006-01-02T15:04",
	}
	for _, layout := range layouts {
		var parsed time.Time
		if layout == time.RFC3339Nano {
			parsed, err = time.Parse(layout, *value)
		} else {
			parsed, err = time.ParseInLocation(layout, *value, location)
		}
		if err == nil {
			result := parsed.UTC()
			return &result, nil
		}
	}
	return nil, errors.New("invalid ERP source time")
}

func hashJSON(value any) string {
	encoded, _ := json.Marshal(value)
	sum := sha256.Sum256(encoded)
	return hex.EncodeToString(sum[:])
}