cmroubao/backend-api/internal/repository/sqlite/freight_repository.go

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,
			created_from, created_to, watermark_through, query_sha256,
			idempotency_key, request_sha256, status, order_count, item_count,
			created_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, ?)`,
		run.ID,
		run.CreatorSubject,
		run.CreatedByUserID,
		run.Mode,
		nullableFreightSyncValue(run.OrderNumber),
		nullableFreightSyncValue(run.CreatedFrom),
		nullableFreightSyncValue(run.CreatedTo),
		nullableTimestamp(run.WatermarkThrough),
		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 {
	return store.completeFreightSync(ctx, run, batch, nil, finishedAt)
}

func (store *Store) CompleteFreightDateSync(
	ctx context.Context,
	run domain.FreightSyncRun,
	batch domain.FreightImportBatch,
	watermarkThrough time.Time,
	finishedAt time.Time,
) error {
	return store.completeFreightSync(
		ctx,
		run,
		batch,
		&watermarkThrough,
		finishedAt,
	)
}

func (store *Store) completeFreightSync(
	ctx context.Context,
	run domain.FreightSyncRun,
	batch domain.FreightImportBatch,
	watermarkThrough *time.Time,
	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 watermarkThrough != nil {
		if _, err := tx.ExecContext(
			ctx,
			`INSERT INTO erp_sync_watermarks (
				creator_subject, source_system, last_successful_to,
				last_successful_run_id, updated_at
			) VALUES (?, 'SHUNYUNBAO', ?, ?, ?)
			ON CONFLICT (creator_subject, source_system)
			DO UPDATE SET
				last_successful_to = excluded.last_successful_to,
				last_successful_run_id = excluded.last_successful_run_id,
				updated_at = excluded.updated_at
			WHERE julianday(erp_sync_watermarks.last_successful_to)
				< julianday(excluded.last_successful_to)`,
			run.CreatorSubject,
			formatTimestamp(*watermarkThrough),
			run.ID,
			formatTimestamp(finishedAt),
		); err != nil {
			return repositoryFailure(err)
		}
	}
	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) GetFreightSyncWatermark(
	ctx context.Context,
	creatorSubject string,
) (*domain.FreightSyncWatermark, error) {
	var watermark domain.FreightSyncWatermark
	var lastSuccessfulTo, updatedAt string
	err := store.db.QueryRowContext(
		ctx,
		`SELECT creator_subject, source_system, last_successful_to,
			last_successful_run_id, updated_at
		FROM erp_sync_watermarks
		WHERE creator_subject = ? AND source_system = 'SHUNYUNBAO'`,
		creatorSubject,
	).Scan(
		&watermark.CreatorSubject,
		&watermark.SourceSystem,
		&lastSuccessfulTo,
		&watermark.LastSuccessfulRunID,
		&updatedAt,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, repositoryFailure(err)
	}
	watermark.LastSuccessfulTo, err = parseTimestamp(lastSuccessfulTo)
	if err != nil {
		return nil, repositoryFailure(err)
	}
	watermark.UpdatedAt, err = parseTimestamp(updatedAt)
	if err != nil {
		return nil, repositoryFailure(err)
	}
	return &watermark, 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,
	created_from, created_to, watermark_through, 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 orderNumber, createdFrom, createdTo sql.NullString
	var watermarkThrough sql.NullString
	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,
		&orderNumber,
		&createdFrom,
		&createdTo,
		&watermarkThrough,
		&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
	}
	if orderNumber.Valid {
		run.OrderNumber = orderNumber.String
	}
	if createdFrom.Valid {
		run.CreatedFrom = createdFrom.String
	}
	if createdTo.Valid {
		run.CreatedTo = createdTo.String
	}
	run.WatermarkThrough, err = parseNullableTimestamp(watermarkThrough)
	if err != nil {
		return domain.FreightSyncRun{}, err
	}
	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 nullableFreightSyncValue(value string) any {
	if value == "" {
		return nil
	}
	return value
}

func optionalString(value sql.NullString) *string {
	if !value.Valid {
		return nil
	}
	return &value.String
}