feat(t222): add freight ingestion admin flow

2026-07-28 23:26:34 +08:00
parent c524e734fb
commit 6d052b0462
37 changed files with 3397 additions and 101 deletions
@@ -0,0 +1,630 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"time"
+
+	"cmroubao/backend-api/internal/domain"
+	"cmroubao/backend-api/internal/usecase"
+)
+
+func (store *Store) CreateFreightSync(
+	ctx context.Context,
+	run domain.FreightSyncRun,
+	idempotencyKey string,
+	requestSHA256 string,
+) (domain.FreightSyncRun, bool, error) {
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return domain.FreightSyncRun{}, false, repositoryFailure(err)
+	}
+	defer tx.Rollback()
+	existing, err := scanFreightSync(tx.QueryRowContext(
+		ctx,
+		freightSyncSelect+`
+		WHERE creator_subject = ? AND idempotency_key = ?`,
+		run.CreatorSubject,
+		idempotencyKey,
+	))
+	if err == nil {
+		var storedHash string
+		if err := tx.QueryRowContext(
+			ctx,
+			`SELECT request_sha256
+			FROM erp_sync_runs
+			WHERE creator_subject = ? AND idempotency_key = ?`,
+			run.CreatorSubject,
+			idempotencyKey,
+		).Scan(&storedHash); err != nil {
+			return domain.FreightSyncRun{}, false, repositoryFailure(err)
+		}
+		if storedHash != requestSHA256 {
+			return domain.FreightSyncRun{}, false, usecase.ErrIdempotencyConflict
+		}
+		return existing, false, nil
+	}
+	if !errors.Is(err, sql.ErrNoRows) {
+		return domain.FreightSyncRun{}, false, repositoryFailure(err)
+	}
+	_, err = tx.ExecContext(
+		ctx,
+		`INSERT INTO erp_sync_runs (
+			id, creator_subject, created_by_user_id, mode, order_number,
+			query_sha256, idempotency_key, request_sha256, status,
+			order_count, item_count, created_at
+		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, ?)`,
+		run.ID,
+		run.CreatorSubject,
+		run.CreatedByUserID,
+		run.Mode,
+		run.OrderNumber,
+		run.QuerySHA256,
+		idempotencyKey,
+		requestSHA256,
+		run.Status,
+		formatTimestamp(run.CreatedAt),
+	)
+	if err != nil {
+		return domain.FreightSyncRun{}, false, repositoryFailure(err)
+	}
+	if err := tx.Commit(); err != nil {
+		return domain.FreightSyncRun{}, false, repositoryFailure(err)
+	}
+	return run, true, nil
+}
+
+func (store *Store) StartFreightSync(
+	ctx context.Context,
+	runID string,
+	startedAt time.Time,
+) error {
+	result, err := store.db.ExecContext(
+		ctx,
+		`UPDATE erp_sync_runs
+		SET status = 'RUNNING', started_at = ?
+		WHERE id = ? AND status = 'PENDING'`,
+		formatTimestamp(startedAt),
+		runID,
+	)
+	if err != nil {
+		return repositoryFailure(err)
+	}
+	changed, err := result.RowsAffected()
+	if err != nil {
+		return repositoryFailure(err)
+	}
+	if changed != 1 {
+		return usecase.ErrTaskStateConflict
+	}
+	return nil
+}
+
+func (store *Store) CompleteFreightSync(
+	ctx context.Context,
+	run domain.FreightSyncRun,
+	batch domain.FreightImportBatch,
+	finishedAt time.Time,
+) error {
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return repositoryFailure(err)
+	}
+	defer tx.Rollback()
+	var status string
+	if err := tx.QueryRowContext(
+		ctx,
+		`SELECT status FROM erp_sync_runs WHERE id = ?`,
+		run.ID,
+	).Scan(&status); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return usecase.ErrRepositoryNotFound
+		}
+		return repositoryFailure(err)
+	}
+	if status != string(domain.FreightSyncRunning) {
+		return usecase.ErrTaskStateConflict
+	}
+	itemCount := 0
+	for _, order := range batch.Orders {
+		orderID, err := upsertFreightOrder(
+			ctx,
+			tx,
+			run,
+			order,
+			finishedAt,
+		)
+		if err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(
+			ctx,
+			`UPDATE freight_order_items
+			SET is_present = 0, last_sync_run_id = ?, updated_at = ?
+			WHERE freight_order_id = ?`,
+			run.ID,
+			formatTimestamp(finishedAt),
+			orderID,
+		); err != nil {
+			return repositoryFailure(err)
+		}
+		for _, item := range order.Items {
+			if err := upsertFreightOrderItem(
+				ctx,
+				tx,
+				run.ID,
+				orderID,
+				item,
+				finishedAt,
+			); err != nil {
+				return err
+			}
+			itemCount++
+		}
+	}
+	result, err := tx.ExecContext(
+		ctx,
+		`UPDATE erp_sync_runs
+		SET status = 'SUCCEEDED', error_code = NULL, order_count = ?,
+			item_count = ?, finished_at = ?
+		WHERE id = ? AND status = 'RUNNING'`,
+		len(batch.Orders),
+		itemCount,
+		formatTimestamp(finishedAt),
+		run.ID,
+	)
+	if err != nil {
+		return repositoryFailure(err)
+	}
+	changed, err := result.RowsAffected()
+	if err != nil {
+		return repositoryFailure(err)
+	}
+	if changed != 1 {
+		return usecase.ErrTaskStateConflict
+	}
+	if err := tx.Commit(); err != nil {
+		return repositoryFailure(err)
+	}
+	return nil
+}
+
+func upsertFreightOrder(
+	ctx context.Context,
+	tx *sql.Tx,
+	run domain.FreightSyncRun,
+	order domain.FreightImportOrder,
+	now time.Time,
+) (string, error) {
+	_, err := tx.ExecContext(
+		ctx,
+		`INSERT INTO freight_orders (
+			id, creator_subject, source_system, external_stock_id,
+			source_code, platform_order_no, shop_name, source_created_at,
+			order_status, purchase_status, is_canceled, canonical_sha256,
+			revision, first_sync_run_id, last_sync_run_id, created_at, updated_at
+		) VALUES (?, ?, 'SHUNYUNBAO', ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?, ?)
+		ON CONFLICT (creator_subject, source_system, external_stock_id)
+		DO UPDATE SET
+			source_code = excluded.source_code,
+			platform_order_no = excluded.platform_order_no,
+			shop_name = excluded.shop_name,
+			source_created_at = excluded.source_created_at,
+			order_status = excluded.order_status,
+			purchase_status = excluded.purchase_status,
+			is_canceled = excluded.is_canceled,
+			revision = CASE
+				WHEN freight_orders.canonical_sha256 != excluded.canonical_sha256
+				THEN freight_orders.revision + 1
+				ELSE freight_orders.revision
+			END,
+			canonical_sha256 = excluded.canonical_sha256,
+			last_sync_run_id = excluded.last_sync_run_id,
+			updated_at = excluded.updated_at`,
+		order.ID,
+		run.CreatorSubject,
+		order.ExternalStockID,
+		order.SourceCode,
+		nullableString(order.PlatformOrderNo),
+		nullableString(order.ShopName),
+		nullableTimestamp(order.SourceCreatedAt),
+		nullableString(order.OrderStatus),
+		nullableString(order.PurchaseStatus),
+		nullableBool(order.IsCanceled),
+		order.CanonicalSHA256,
+		run.ID,
+		run.ID,
+		formatTimestamp(now),
+		formatTimestamp(now),
+	)
+	if err != nil {
+		return "", repositoryFailure(err)
+	}
+	var orderID string
+	if err := tx.QueryRowContext(
+		ctx,
+		`SELECT id FROM freight_orders
+		WHERE creator_subject = ? AND source_system = 'SHUNYUNBAO'
+			AND external_stock_id = ?`,
+		run.CreatorSubject,
+		order.ExternalStockID,
+	).Scan(&orderID); err != nil {
+		return "", repositoryFailure(err)
+	}
+	return orderID, nil
+}
+
+func upsertFreightOrderItem(
+	ctx context.Context,
+	tx *sql.Tx,
+	runID, orderID string,
+	item domain.FreightImportItem,
+	now time.Time,
+) error {
+	_, err := tx.ExecContext(
+		ctx,
+		`INSERT INTO freight_order_items (
+			id, freight_order_id, external_item_id, title, product_spec,
+			sku, quantity, product_thumb_ref, purchase_status,
+			canonical_sha256, revision, is_present, first_sync_run_id,
+			last_sync_run_id, created_at, updated_at
+		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, 1, ?, ?, ?, ?)
+		ON CONFLICT (freight_order_id, external_item_id)
+		DO UPDATE SET
+			title = excluded.title,
+			product_spec = excluded.product_spec,
+			sku = excluded.sku,
+			quantity = excluded.quantity,
+			product_thumb_ref = excluded.product_thumb_ref,
+			purchase_status = excluded.purchase_status,
+			revision = CASE
+				WHEN freight_order_items.canonical_sha256 != excluded.canonical_sha256
+				THEN freight_order_items.revision + 1
+				ELSE freight_order_items.revision
+			END,
+			canonical_sha256 = excluded.canonical_sha256,
+			is_present = 1,
+			last_sync_run_id = excluded.last_sync_run_id,
+			updated_at = excluded.updated_at`,
+		item.ID,
+		orderID,
+		item.ExternalItemID,
+		item.Title,
+		item.ProductSpec,
+		item.SKU,
+		nullableFreightQuantity(item.Quantity),
+		nullableString(item.ProductThumbRef),
+		nullableString(item.PurchaseStatus),
+		item.CanonicalSHA256,
+		runID,
+		runID,
+		formatTimestamp(now),
+		formatTimestamp(now),
+	)
+	if err != nil {
+		return repositoryFailure(err)
+	}
+	return nil
+}
+
+func (store *Store) FailFreightSync(
+	ctx context.Context,
+	runID, errorCode string,
+	finishedAt time.Time,
+) error {
+	result, err := store.db.ExecContext(
+		ctx,
+		`UPDATE erp_sync_runs
+		SET status = 'FAILED', error_code = ?, finished_at = ?
+		WHERE id = ? AND status IN ('PENDING', 'RUNNING')`,
+		errorCode,
+		formatTimestamp(finishedAt),
+		runID,
+	)
+	if err != nil {
+		return repositoryFailure(err)
+	}
+	changed, err := result.RowsAffected()
+	if err != nil {
+		return repositoryFailure(err)
+	}
+	if changed != 1 {
+		return usecase.ErrTaskStateConflict
+	}
+	return nil
+}
+
+func (store *Store) RecoverFreightSyncs(
+	ctx context.Context,
+	finishedAt time.Time,
+) (int64, error) {
+	result, err := store.db.ExecContext(
+		ctx,
+		`UPDATE erp_sync_runs
+		SET status = 'FAILED', error_code = 'PROCESS_INTERRUPTED',
+			finished_at = ?
+		WHERE status IN ('PENDING', 'RUNNING')`,
+		formatTimestamp(finishedAt),
+	)
+	if err != nil {
+		return 0, repositoryFailure(err)
+	}
+	changed, err := result.RowsAffected()
+	if err != nil {
+		return 0, repositoryFailure(err)
+	}
+	return changed, nil
+}
+
+func (store *Store) GetFreightSync(
+	ctx context.Context,
+	creatorSubject, runID string,
+) (domain.FreightSyncRun, error) {
+	run, err := scanFreightSync(store.db.QueryRowContext(
+		ctx,
+		freightSyncSelect+`
+		WHERE creator_subject = ? AND id = ?`,
+		creatorSubject,
+		runID,
+	))
+	if errors.Is(err, sql.ErrNoRows) {
+		return domain.FreightSyncRun{}, usecase.ErrRepositoryNotFound
+	}
+	if err != nil {
+		return domain.FreightSyncRun{}, repositoryFailure(err)
+	}
+	return run, nil
+}
+
+func (store *Store) ListFreightOrders(
+	ctx context.Context,
+	creatorSubject string,
+	limit int,
+) ([]domain.FreightOrder, error) {
+	rows, err := store.db.QueryContext(
+		ctx,
+		freightOrderSelect+`
+		WHERE freight_orders.creator_subject = ?
+		ORDER BY freight_orders.updated_at DESC, freight_orders.id DESC
+		LIMIT ?`,
+		creatorSubject,
+		limit,
+	)
+	if err != nil {
+		return nil, repositoryFailure(err)
+	}
+	defer rows.Close()
+	orders := make([]domain.FreightOrder, 0)
+	for rows.Next() {
+		order, err := scanFreightOrder(rows)
+		if err != nil {
+			return nil, repositoryFailure(err)
+		}
+		orders = append(orders, order)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, repositoryFailure(err)
+	}
+	return orders, nil
+}
+
+func (store *Store) GetFreightOrder(
+	ctx context.Context,
+	creatorSubject, orderID string,
+) (domain.FreightOrderDetail, error) {
+	order, err := scanFreightOrder(store.db.QueryRowContext(
+		ctx,
+		freightOrderSelect+`
+		WHERE freight_orders.creator_subject = ? AND freight_orders.id = ?`,
+		creatorSubject,
+		orderID,
+	))
+	if errors.Is(err, sql.ErrNoRows) {
+		return domain.FreightOrderDetail{}, usecase.ErrRepositoryNotFound
+	}
+	if err != nil {
+		return domain.FreightOrderDetail{}, repositoryFailure(err)
+	}
+	rows, err := store.db.QueryContext(
+		ctx,
+		`SELECT
+			id, freight_order_id, external_item_id, title, product_spec,
+			sku, quantity, product_thumb_ref, purchase_status,
+			canonical_sha256, revision, is_present, first_sync_run_id,
+			last_sync_run_id, created_at, updated_at
+		FROM freight_order_items
+		WHERE freight_order_id = ? AND is_present = 1
+		ORDER BY CAST(external_item_id AS INTEGER), external_item_id`,
+		order.ID,
+	)
+	if err != nil {
+		return domain.FreightOrderDetail{}, repositoryFailure(err)
+	}
+	defer rows.Close()
+	items := make([]domain.FreightOrderItem, 0)
+	for rows.Next() {
+		item, err := scanFreightOrderItem(rows)
+		if err != nil {
+			return domain.FreightOrderDetail{}, repositoryFailure(err)
+		}
+		items = append(items, item)
+	}
+	if err := rows.Err(); err != nil {
+		return domain.FreightOrderDetail{}, repositoryFailure(err)
+	}
+	return domain.FreightOrderDetail{Order: order, Items: items}, nil
+}
+
+const freightSyncSelect = `SELECT
+	id, creator_subject, created_by_user_id, mode, order_number,
+	query_sha256, status, error_code, order_count, item_count,
+	created_at, started_at, finished_at
+	FROM erp_sync_runs
+	`
+
+func scanFreightSync(scanner rowScanner) (domain.FreightSyncRun, error) {
+	var run domain.FreightSyncRun
+	var errorCode sql.NullString
+	var createdAt string
+	var startedAt sql.NullString
+	var finishedAt sql.NullString
+	err := scanner.Scan(
+		&run.ID,
+		&run.CreatorSubject,
+		&run.CreatedByUserID,
+		&run.Mode,
+		&run.OrderNumber,
+		&run.QuerySHA256,
+		&run.Status,
+		&errorCode,
+		&run.OrderCount,
+		&run.ItemCount,
+		&createdAt,
+		&startedAt,
+		&finishedAt,
+	)
+	if err != nil {
+		return domain.FreightSyncRun{}, err
+	}
+	if errorCode.Valid {
+		run.ErrorCode = &errorCode.String
+	}
+	run.CreatedAt, err = parseTimestamp(createdAt)
+	if err != nil {
+		return domain.FreightSyncRun{}, err
+	}
+	run.StartedAt, err = parseNullableTimestamp(startedAt)
+	if err != nil {
+		return domain.FreightSyncRun{}, err
+	}
+	run.FinishedAt, err = parseNullableTimestamp(finishedAt)
+	return run, err
+}
+
+const freightOrderSelect = `SELECT
+	freight_orders.id, freight_orders.creator_subject,
+	freight_orders.source_system, freight_orders.external_stock_id,
+	freight_orders.source_code, freight_orders.platform_order_no,
+	freight_orders.shop_name, freight_orders.source_created_at,
+	freight_orders.order_status, freight_orders.purchase_status,
+	freight_orders.is_canceled, freight_orders.canonical_sha256,
+	freight_orders.revision, freight_orders.first_sync_run_id,
+	freight_orders.last_sync_run_id, freight_orders.created_at,
+	freight_orders.updated_at,
+	(SELECT count(*) FROM freight_order_items
+		WHERE freight_order_id = freight_orders.id AND is_present = 1)
+	FROM freight_orders
+	`
+
+func scanFreightOrder(scanner rowScanner) (domain.FreightOrder, error) {
+	var order domain.FreightOrder
+	var platformOrderNo, shopName, sourceCreatedAt sql.NullString
+	var orderStatus, purchaseStatus sql.NullString
+	var isCanceled sql.NullBool
+	var createdAt, updatedAt string
+	err := scanner.Scan(
+		&order.ID,
+		&order.CreatorSubject,
+		&order.SourceSystem,
+		&order.ExternalStockID,
+		&order.SourceCode,
+		&platformOrderNo,
+		&shopName,
+		&sourceCreatedAt,
+		&orderStatus,
+		&purchaseStatus,
+		&isCanceled,
+		&order.CanonicalSHA256,
+		&order.Revision,
+		&order.FirstSyncRunID,
+		&order.LastSyncRunID,
+		&createdAt,
+		&updatedAt,
+		&order.ItemCount,
+	)
+	if err != nil {
+		return domain.FreightOrder{}, err
+	}
+	order.PlatformOrderNo = optionalString(platformOrderNo)
+	order.ShopName = optionalString(shopName)
+	order.OrderStatus = optionalString(orderStatus)
+	order.PurchaseStatus = optionalString(purchaseStatus)
+	if isCanceled.Valid {
+		order.IsCanceled = &isCanceled.Bool
+	}
+	if sourceCreatedAt.Valid {
+		parsed, err := parseTimestamp(sourceCreatedAt.String)
+		if err != nil {
+			return domain.FreightOrder{}, err
+		}
+		order.SourceCreatedAt = &parsed
+	}
+	order.CreatedAt, err = parseTimestamp(createdAt)
+	if err != nil {
+		return domain.FreightOrder{}, err
+	}
+	order.UpdatedAt, err = parseTimestamp(updatedAt)
+	return order, err
+}
+
+func scanFreightOrderItem(scanner rowScanner) (domain.FreightOrderItem, error) {
+	var item domain.FreightOrderItem
+	var quantity sql.NullInt64
+	var thumb, purchaseStatus sql.NullString
+	var createdAt, updatedAt string
+	err := scanner.Scan(
+		&item.ID,
+		&item.FreightOrderID,
+		&item.ExternalItemID,
+		&item.Title,
+		&item.ProductSpec,
+		&item.SKU,
+		&quantity,
+		&thumb,
+		&purchaseStatus,
+		&item.CanonicalSHA256,
+		&item.Revision,
+		&item.IsPresent,
+		&item.FirstSyncRunID,
+		&item.LastSyncRunID,
+		&createdAt,
+		&updatedAt,
+	)
+	if err != nil {
+		return domain.FreightOrderItem{}, err
+	}
+	if quantity.Valid {
+		value := int(quantity.Int64)
+		item.Quantity = &value
+	}
+	item.ProductThumbRef = optionalString(thumb)
+	item.PurchaseStatus = optionalString(purchaseStatus)
+	item.CreatedAt, err = parseTimestamp(createdAt)
+	if err != nil {
+		return domain.FreightOrderItem{}, err
+	}
+	item.UpdatedAt, err = parseTimestamp(updatedAt)
+	return item, err
+}
+
+func nullableTimestamp(value *time.Time) any {
+	if value == nil {
+		return nil
+	}
+	return formatTimestamp(*value)
+}
+
+func nullableFreightQuantity(value *int) any {
+	if value == nil {
+		return nil
+	}
+	return *value
+}
+
+func optionalString(value sql.NullString) *string {
+	if !value.Valid {
+		return nil
+	}
+	return &value.String
+}