feat(t224): add incremental freight sync

2026-07-29 00:26:05 +08:00
parent 433db45100
commit c69879650e
35 changed files with 2085 additions and 156 deletions
@@ -19,6 +19,8 @@ import (
 const (
 	maxFreightOrdersPerSync = 100
 	maxFreightItemsPerOrder = 1000
+	maxFreightWindowDays    = 7
+	freightWatermarkOverlap = 10 * time.Minute
 )

 type FreightService struct {
@@ -36,6 +38,15 @@ type CreateFreightSyncCommand struct {
 	OrderNumber    string
 }

+type CreateFreightDateSyncCommand struct {
+	CreatorSubject string
+	ActorUserID    string
+	IdempotencyKey string
+	CreatedFrom    string
+	CreatedTo      string
+	SyncToNow      bool
+}
+
 type CreateFreightSyncResult struct {
 	Run      domain.FreightSyncRun
 	Replayed bool
@@ -127,6 +138,147 @@ func (service *FreightService) CreateOrderSync(
 	return CreateFreightSyncResult{Run: run, Replayed: !created}, nil
 }

+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,
+		)
+	}
+
+	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) execute(run domain.FreightSyncRun) {
 	ctx, cancel := context.WithTimeout(context.Background(), service.timeout)
 	defer cancel()
@@ -134,7 +286,13 @@ func (service *FreightService) execute(run domain.FreightSyncRun) {
 	if err := service.repository.StartFreightSync(ctx, run.ID, now); err != nil {
 		return
 	}
-	source, err := service.source.QueryOrder(ctx, run.OrderNumber)
+	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 {
 		_ = service.repository.FailFreightSync(
 			ctx,
@@ -154,12 +312,25 @@ func (service *FreightService) execute(run domain.FreightSyncRun) {
 		)
 		return
 	}
-	if err := service.repository.CompleteFreightSync(
-		ctx,
-		run,
-		batch,
-		service.clock.Now().UTC(),
-	); err != nil {
+	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 {
 		_ = service.repository.FailFreightSync(
 			ctx,
 			run.ID,
@@ -169,6 +340,76 @@ func (service *FreightService) execute(run domain.FreightSyncRun) {
 	}
 }

+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{}, erpconnector.ErrProtocol
+	}
+	end, err := parseFreightDate(run.CreatedTo, location)
+	if err != nil || end.Before(start) {
+		return domain.FreightSourceBatch{}, erpconnector.ErrProtocol
+	}
+	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{}, erpconnector.ErrProtocol
+		}
+		for _, order := range batch.Orders {
+			existing, exists := seen[order.ExternalStockID]
+			if exists {
+				if hashJSON(existing) != hashJSON(order) {
+					return domain.FreightSourceBatch{}, erpconnector.ErrProtocol
+				}
+				continue
+			}
+			seen[order.ExternalStockID] = order
+			orders = append(orders, order)
+			if len(orders) > maxFreightOrdersPerSync {
+				return domain.FreightSourceBatch{}, erpconnector.ErrProtocol
+			}
+		}
+		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) {
@@ -197,6 +438,20 @@ func (service *FreightService) GetSync(
 	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,
@@ -242,7 +497,8 @@ func (service *FreightService) normalize(
 	source domain.FreightSourceBatch,
 ) (domain.FreightImportBatch, error) {
 	if source.SchemaVersion != 1 ||
-		source.Query.Mode != domain.FreightSyncOrderNumber ||
+		(source.Query.Mode != domain.FreightSyncOrderNumber &&
+			source.Query.Mode != domain.FreightSyncCreatedRange) ||
 		len(source.Orders) > maxFreightOrdersPerSync {
 		return domain.FreightImportBatch{}, errors.New("invalid source envelope")
 	}
@@ -373,6 +629,47 @@ func (service *FreightService) normalize(
 	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, erpconnector.ErrNotConfigured):