feat(t223): generate tasks from freight items

2026-07-28 23:51:59 +08:00
parent 6d052b0462
commit 433db45100
25 changed files with 2697 additions and 48 deletions
@@ -0,0 +1,632 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"time"
+
+	"cmroubao/backend-api/internal/domain"
+	"cmroubao/backend-api/internal/usecase"
+)
+
+const createProcurementTaskOperation = "CREATE_PROCUREMENT_TASK"
+
+func (store *Store) GetProcurementSourceItem(
+	ctx context.Context,
+	creatorSubject, itemID string,
+) (domain.ProcurementSourceItem, error) {
+	item, err := scanFreightOrderItem(store.db.QueryRowContext(
+		ctx,
+		`SELECT
+			item.id, item.freight_order_id, item.external_item_id,
+			item.title, item.product_spec, item.sku, item.quantity,
+			item.product_thumb_ref, item.purchase_status,
+			item.canonical_sha256, item.revision, item.is_present,
+			item.first_sync_run_id, item.last_sync_run_id,
+			item.created_at, item.updated_at
+		FROM freight_order_items AS item
+		JOIN freight_orders AS freight
+			ON freight.id = item.freight_order_id
+		WHERE freight.creator_subject = ? AND item.id = ?
+			AND item.is_present = 1`,
+		creatorSubject,
+		itemID,
+	))
+	if errors.Is(err, sql.ErrNoRows) {
+		return domain.ProcurementSourceItem{}, usecase.ErrRepositoryNotFound
+	}
+	if err != nil {
+		return domain.ProcurementSourceItem{}, repositoryFailure(err)
+	}
+	var canceled sql.NullBool
+	if err := store.db.QueryRowContext(
+		ctx,
+		`SELECT freight.is_canceled
+		FROM freight_orders AS freight
+		JOIN freight_order_items AS item
+			ON item.freight_order_id = freight.id
+		WHERE freight.creator_subject = ? AND item.id = ?`,
+		creatorSubject,
+		itemID,
+	).Scan(&canceled); err != nil {
+		return domain.ProcurementSourceItem{}, repositoryFailure(err)
+	}
+	var isCanceled *bool
+	if canceled.Valid {
+		isCanceled = &canceled.Bool
+	}
+	return domain.ProcurementSourceItem{
+		Item:       item,
+		IsCanceled: isCanceled,
+	}, nil
+}
+
+func (store *Store) CreateProcurementRequest(
+	ctx context.Context,
+	candidate domain.ProcurementRequest,
+) (domain.ProcurementRequest, bool, error) {
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return domain.ProcurementRequest{}, false, repositoryFailure(err)
+	}
+	defer tx.Rollback()
+	existing, err := getProcurementRequestBySource(
+		ctx,
+		tx,
+		candidate.CreatorSubject,
+		candidate.FreightOrderItemID,
+		candidate.SourceRevision,
+	)
+	if err == nil {
+		if err := tx.Commit(); err != nil {
+			return domain.ProcurementRequest{}, false, repositoryFailure(err)
+		}
+		return existing, false, nil
+	}
+	if !errors.Is(err, usecase.ErrRepositoryNotFound) {
+		return domain.ProcurementRequest{}, false, err
+	}
+	var currentRevision int
+	var currentSHA string
+	var present bool
+	if err := tx.QueryRowContext(
+		ctx,
+		`SELECT item.revision, item.canonical_sha256, item.is_present
+		FROM freight_order_items AS item
+		JOIN freight_orders AS freight
+			ON freight.id = item.freight_order_id
+		WHERE freight.creator_subject = ? AND item.id = ?`,
+		candidate.CreatorSubject,
+		candidate.FreightOrderItemID,
+	).Scan(&currentRevision, &currentSHA, &present); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return domain.ProcurementRequest{}, false, usecase.ErrRepositoryNotFound
+		}
+		return domain.ProcurementRequest{}, false, repositoryFailure(err)
+	}
+	if !present || currentRevision != candidate.SourceRevision ||
+		currentSHA != candidate.SourceSHA256 {
+		return domain.ProcurementRequest{}, false,
+			usecase.ErrProcurementSourceChanged
+	}
+	_, err = tx.ExecContext(
+		ctx,
+		`INSERT INTO procurement_requests (
+			id, creator_subject, freight_order_item_id, source_revision,
+			source_sha256, title, product_spec, sku, quantity,
+			source_purchase_status, source_is_canceled,
+			procurement_confirmed_by_user_id, procurement_confirmed_at,
+			status, blocking_code, reference_asset_id, purchase_task_id,
+			created_at, updated_at
+		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?)`,
+		candidate.ID,
+		candidate.CreatorSubject,
+		candidate.FreightOrderItemID,
+		candidate.SourceRevision,
+		candidate.SourceSHA256,
+		candidate.Title,
+		candidate.ProductSpec,
+		candidate.SKU,
+		nullableFreightQuantity(candidate.Quantity),
+		nullableString(candidate.SourcePurchaseStatus),
+		nullableBool(candidate.SourceIsCanceled),
+		candidate.ProcurementConfirmedByUserID,
+		formatTimestamp(candidate.ProcurementConfirmedAt),
+		candidate.Status,
+		nullableString(candidate.BlockingCode),
+		formatTimestamp(candidate.CreatedAt),
+		formatTimestamp(candidate.UpdatedAt),
+	)
+	if err != nil {
+		return domain.ProcurementRequest{}, false, repositoryFailure(err)
+	}
+	if err := tx.Commit(); err != nil {
+		return domain.ProcurementRequest{}, false, repositoryFailure(err)
+	}
+	return candidate, true, nil
+}
+
+func (store *Store) ListProcurementRequestsForOrder(
+	ctx context.Context,
+	creatorSubject, orderID string,
+) ([]domain.ProcurementRequest, error) {
+	rows, err := store.db.QueryContext(
+		ctx,
+		procurementRequestSelect+`
+		WHERE request.creator_subject = ?
+			AND item.freight_order_id = ?
+		ORDER BY item.id, request.source_revision DESC, request.id DESC`,
+		creatorSubject,
+		orderID,
+	)
+	if err != nil {
+		return nil, repositoryFailure(err)
+	}
+	defer rows.Close()
+	requests := make([]domain.ProcurementRequest, 0)
+	for rows.Next() {
+		request, err := scanProcurementRequest(rows)
+		if err != nil {
+			return nil, repositoryFailure(err)
+		}
+		requests = append(requests, request)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, repositoryFailure(err)
+	}
+	return requests, nil
+}
+
+func (store *Store) GetProcurementRequest(
+	ctx context.Context,
+	creatorSubject, requestID string,
+) (domain.ProcurementRequest, error) {
+	return getProcurementRequest(ctx, store.db, creatorSubject, requestID)
+}
+
+func (store *Store) BindProcurementReference(
+	ctx context.Context,
+	creatorSubject, requestID, assetID string,
+	now time.Time,
+) (domain.ProcurementRequest, error) {
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return domain.ProcurementRequest{}, repositoryFailure(err)
+	}
+	defer tx.Rollback()
+	request, err := getProcurementRequest(
+		ctx,
+		tx,
+		creatorSubject,
+		requestID,
+	)
+	if err != nil {
+		return domain.ProcurementRequest{}, err
+	}
+	if request.SourceChanged {
+		if request.PurchaseTaskID == nil {
+			if err := markProcurementSourceChanged(ctx, tx, request.ID, now); err != nil {
+				return domain.ProcurementRequest{}, err
+			}
+		}
+		if err := tx.Commit(); err != nil {
+			return domain.ProcurementRequest{}, repositoryFailure(err)
+		}
+		return domain.ProcurementRequest{}, usecase.ErrProcurementSourceChanged
+	}
+	if request.ReferenceAssetID != nil &&
+		*request.ReferenceAssetID == assetID &&
+		(request.Status == domain.ProcurementReady ||
+			request.Status == domain.ProcurementTaskCreated) {
+		if err := tx.Commit(); err != nil {
+			return domain.ProcurementRequest{}, repositoryFailure(err)
+		}
+		return request, nil
+	}
+	if request.Status != domain.ProcurementNeedsImage {
+		return domain.ProcurementRequest{}, usecase.ErrProcurementStateConflict
+	}
+	var available int
+	if err := tx.QueryRowContext(
+		ctx,
+		`SELECT EXISTS (
+			SELECT 1
+			FROM assets AS asset
+			WHERE asset.id = ? AND asset.creator_subject = ?
+				AND asset.purpose = 'TASK_REFERENCE'
+				AND NOT EXISTS (
+					SELECT 1 FROM purchase_tasks AS task
+					WHERE task.image_asset_id = asset.id
+				)
+				AND NOT EXISTS (
+					SELECT 1 FROM procurement_requests AS other
+					WHERE other.reference_asset_id = asset.id
+						AND other.id != ?
+				)
+		)`,
+		assetID,
+		creatorSubject,
+		requestID,
+	).Scan(&available); err != nil {
+		return domain.ProcurementRequest{}, repositoryFailure(err)
+	}
+	if available != 1 {
+		return domain.ProcurementRequest{}, usecase.ErrAssetUnavailable
+	}
+	if _, err := tx.ExecContext(
+		ctx,
+		`UPDATE procurement_requests
+		SET reference_asset_id = ?, status = 'READY', updated_at = ?
+		WHERE id = ? AND status = 'NEEDS_IMAGE'`,
+		assetID,
+		formatTimestamp(now),
+		requestID,
+	); err != nil {
+		return domain.ProcurementRequest{}, repositoryFailure(err)
+	}
+	updated, err := getProcurementRequest(ctx, tx, creatorSubject, requestID)
+	if err != nil {
+		return domain.ProcurementRequest{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return domain.ProcurementRequest{}, repositoryFailure(err)
+	}
+	return updated, nil
+}
+
+func (store *Store) CreateProcurementTask(
+	ctx context.Context,
+	expected domain.ProcurementRequest,
+	task domain.PurchaseTask,
+	event domain.TaskEvent,
+	source domain.PurchaseTaskSource,
+	idempotencyKey, requestSHA256 string,
+) (domain.PurchaseTask, bool, error) {
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return domain.PurchaseTask{}, false, repositoryFailure(err)
+	}
+	defer tx.Rollback()
+	existingHash, resourceID, found, err := lookupIdempotency(
+		ctx,
+		tx,
+		task.CreatorSubject,
+		createProcurementTaskOperation,
+		idempotencyKey,
+	)
+	if err != nil {
+		return domain.PurchaseTask{}, false, err
+	}
+	if found {
+		if existingHash != requestSHA256 {
+			return domain.PurchaseTask{}, false, usecase.ErrIdempotencyConflict
+		}
+		existing, err := getTaskByID(
+			ctx,
+			tx,
+			task.CreatorSubject,
+			resourceID,
+		)
+		if err != nil {
+			return domain.PurchaseTask{}, false, err
+		}
+		if err := tx.Commit(); err != nil {
+			return domain.PurchaseTask{}, false, repositoryFailure(err)
+		}
+		return existing, false, nil
+	}
+	request, err := getProcurementRequest(
+		ctx,
+		tx,
+		task.CreatorSubject,
+		expected.ID,
+	)
+	if err != nil {
+		return domain.PurchaseTask{}, false, err
+	}
+	if request.Status == domain.ProcurementTaskCreated &&
+		request.PurchaseTaskID != nil {
+		existing, err := getTaskByID(
+			ctx,
+			tx,
+			task.CreatorSubject,
+			*request.PurchaseTaskID,
+		)
+		if err != nil {
+			return domain.PurchaseTask{}, false, err
+		}
+		if err := insertIdempotency(
+			ctx,
+			tx,
+			task.CreatorSubject,
+			createProcurementTaskOperation,
+			idempotencyKey,
+			requestSHA256,
+			"PURCHASE_TASK",
+			existing.ID,
+			task.CreatedAt,
+		); err != nil {
+			return domain.PurchaseTask{}, false, err
+		}
+		if err := tx.Commit(); err != nil {
+			return domain.PurchaseTask{}, false, repositoryFailure(err)
+		}
+		return existing, false, nil
+	}
+	if request.SourceChanged {
+		if request.PurchaseTaskID == nil {
+			if err := markProcurementSourceChanged(
+				ctx,
+				tx,
+				request.ID,
+				task.CreatedAt,
+			); err != nil {
+				return domain.PurchaseTask{}, false, err
+			}
+		}
+		if err := tx.Commit(); err != nil {
+			return domain.PurchaseTask{}, false, repositoryFailure(err)
+		}
+		return domain.PurchaseTask{}, false,
+			usecase.ErrProcurementSourceChanged
+	}
+	if request.Status != domain.ProcurementReady ||
+		request.ReferenceAssetID == nil ||
+		request.SourceSHA256 != expected.SourceSHA256 {
+		return domain.PurchaseTask{}, false,
+			usecase.ErrProcurementStateConflict
+	}
+	var assetAvailable int
+	if err := tx.QueryRowContext(
+		ctx,
+		`SELECT EXISTS (
+			SELECT 1 FROM assets AS asset
+			WHERE asset.id = ? AND asset.creator_subject = ?
+				AND asset.purpose = 'TASK_REFERENCE'
+				AND NOT EXISTS (
+					SELECT 1 FROM purchase_tasks AS other
+					WHERE other.image_asset_id = asset.id
+				)
+		)`,
+		*request.ReferenceAssetID,
+		task.CreatorSubject,
+	).Scan(&assetAvailable); err != nil {
+		return domain.PurchaseTask{}, false, repositoryFailure(err)
+	}
+	if assetAvailable != 1 {
+		return domain.PurchaseTask{}, false, usecase.ErrAssetUnavailable
+	}
+	_, err = tx.ExecContext(
+		ctx,
+		`INSERT INTO purchase_tasks (
+			id, creator_subject, created_by_user_id, source_ref, title,
+			description, sku, image_asset_id, quantity, max_budget_cents,
+			currency, status, version, cancel_reason, canceled_at,
+			created_at, updated_at
+		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, 'CNY', 'PENDING', 1,
+			NULL, NULL, ?, ?)`,
+		task.ID,
+		task.CreatorSubject,
+		nullableString(task.CreatedByUserID),
+		nullableString(task.SourceRef),
+		task.Title,
+		task.Description,
+		task.SKU,
+		task.ImageAssetID,
+		task.Quantity,
+		formatTimestamp(task.CreatedAt),
+		formatTimestamp(task.UpdatedAt),
+	)
+	if err != nil {
+		return domain.PurchaseTask{}, false, repositoryFailure(err)
+	}
+	if err := insertTaskEvent(ctx, tx, event); err != nil {
+		return domain.PurchaseTask{}, false, err
+	}
+	_, err = tx.ExecContext(
+		ctx,
+		`INSERT INTO purchase_task_sources (
+			task_id, procurement_request_id, freight_order_item_id,
+			source_revision, source_sha256, created_at
+		) VALUES (?, ?, ?, ?, ?, ?)`,
+		source.TaskID,
+		source.ProcurementRequestID,
+		source.FreightOrderItemID,
+		source.SourceRevision,
+		source.SourceSHA256,
+		formatTimestamp(source.CreatedAt),
+	)
+	if err != nil {
+		return domain.PurchaseTask{}, false, repositoryFailure(err)
+	}
+	result, err := tx.ExecContext(
+		ctx,
+		`UPDATE procurement_requests
+		SET status = 'TASK_CREATED', purchase_task_id = ?, updated_at = ?
+		WHERE id = ? AND status = 'READY'`,
+		task.ID,
+		formatTimestamp(task.CreatedAt),
+		request.ID,
+	)
+	if err != nil {
+		return domain.PurchaseTask{}, false, repositoryFailure(err)
+	}
+	changed, err := result.RowsAffected()
+	if err != nil {
+		return domain.PurchaseTask{}, false, repositoryFailure(err)
+	}
+	if changed != 1 {
+		return domain.PurchaseTask{}, false,
+			usecase.ErrProcurementStateConflict
+	}
+	if err := insertIdempotency(
+		ctx,
+		tx,
+		task.CreatorSubject,
+		createProcurementTaskOperation,
+		idempotencyKey,
+		requestSHA256,
+		"PURCHASE_TASK",
+		task.ID,
+		task.CreatedAt,
+	); err != nil {
+		return domain.PurchaseTask{}, false, err
+	}
+	if err := tx.Commit(); err != nil {
+		return domain.PurchaseTask{}, false, repositoryFailure(err)
+	}
+	return task, true, nil
+}
+
+const procurementRequestSelect = `SELECT
+	request.id, request.creator_subject, request.freight_order_item_id,
+	request.source_revision, request.source_sha256, request.title,
+	request.product_spec, request.sku, request.quantity,
+	request.source_purchase_status, request.source_is_canceled,
+	request.procurement_confirmed_by_user_id,
+	request.procurement_confirmed_at, request.status,
+	request.blocking_code, request.reference_asset_id,
+	request.purchase_task_id, request.created_at, request.updated_at,
+	CASE
+		WHEN item.revision != request.source_revision
+			OR item.canonical_sha256 != request.source_sha256
+			OR item.is_present != 1
+			OR COALESCE(freight.is_canceled, -1)
+				!= COALESCE(request.source_is_canceled, -1)
+		THEN 1 ELSE 0
+	END
+	FROM procurement_requests AS request
+	JOIN freight_order_items AS item
+		ON item.id = request.freight_order_item_id
+	JOIN freight_orders AS freight
+		ON freight.id = item.freight_order_id
+	`
+
+func getProcurementRequest(
+	ctx context.Context,
+	query queryRower,
+	creatorSubject, requestID string,
+) (domain.ProcurementRequest, error) {
+	request, err := scanProcurementRequest(query.QueryRowContext(
+		ctx,
+		procurementRequestSelect+`
+		WHERE request.creator_subject = ? AND request.id = ?`,
+		creatorSubject,
+		requestID,
+	))
+	if errors.Is(err, sql.ErrNoRows) {
+		return domain.ProcurementRequest{}, usecase.ErrRepositoryNotFound
+	}
+	if err != nil {
+		return domain.ProcurementRequest{}, repositoryFailure(err)
+	}
+	return request, nil
+}
+
+func getProcurementRequestBySource(
+	ctx context.Context,
+	query queryRower,
+	creatorSubject, itemID string,
+	revision int,
+) (domain.ProcurementRequest, error) {
+	request, err := scanProcurementRequest(query.QueryRowContext(
+		ctx,
+		procurementRequestSelect+`
+		WHERE request.creator_subject = ?
+			AND request.freight_order_item_id = ?
+			AND request.source_revision = ?`,
+		creatorSubject,
+		itemID,
+		revision,
+	))
+	if errors.Is(err, sql.ErrNoRows) {
+		return domain.ProcurementRequest{}, usecase.ErrRepositoryNotFound
+	}
+	if err != nil {
+		return domain.ProcurementRequest{}, repositoryFailure(err)
+	}
+	return request, nil
+}
+
+func scanProcurementRequest(
+	scanner rowScanner,
+) (domain.ProcurementRequest, error) {
+	var request domain.ProcurementRequest
+	var quantity sql.NullInt64
+	var purchaseStatus sql.NullString
+	var isCanceled sql.NullBool
+	var blockingCode, assetID, taskID sql.NullString
+	var confirmedAt, createdAt, updatedAt string
+	var sourceChanged bool
+	err := scanner.Scan(
+		&request.ID,
+		&request.CreatorSubject,
+		&request.FreightOrderItemID,
+		&request.SourceRevision,
+		&request.SourceSHA256,
+		&request.Title,
+		&request.ProductSpec,
+		&request.SKU,
+		&quantity,
+		&purchaseStatus,
+		&isCanceled,
+		&request.ProcurementConfirmedByUserID,
+		&confirmedAt,
+		&request.Status,
+		&blockingCode,
+		&assetID,
+		&taskID,
+		&createdAt,
+		&updatedAt,
+		&sourceChanged,
+	)
+	if err != nil {
+		return domain.ProcurementRequest{}, err
+	}
+	if quantity.Valid {
+		value := int(quantity.Int64)
+		request.Quantity = &value
+	}
+	request.SourcePurchaseStatus = optionalString(purchaseStatus)
+	if isCanceled.Valid {
+		request.SourceIsCanceled = &isCanceled.Bool
+	}
+	request.BlockingCode = optionalString(blockingCode)
+	request.ReferenceAssetID = optionalString(assetID)
+	request.PurchaseTaskID = optionalString(taskID)
+	request.SourceChanged = sourceChanged
+	request.ProcurementConfirmedAt, err = parseTimestamp(confirmedAt)
+	if err != nil {
+		return domain.ProcurementRequest{}, err
+	}
+	request.CreatedAt, err = parseTimestamp(createdAt)
+	if err != nil {
+		return domain.ProcurementRequest{}, err
+	}
+	request.UpdatedAt, err = parseTimestamp(updatedAt)
+	return request, err
+}
+
+func markProcurementSourceChanged(
+	ctx context.Context,
+	tx *sql.Tx,
+	requestID string,
+	now time.Time,
+) error {
+	_, err := tx.ExecContext(
+		ctx,
+		`UPDATE procurement_requests
+		SET status = 'SOURCE_CHANGED', blocking_code = 'SOURCE_CHANGED',
+			updated_at = ?
+		WHERE id = ? AND purchase_task_id IS NULL`,
+		formatTimestamp(now),
+		requestID,
+	)
+	if err != nil {
+		return repositoryFailure(err)
+	}
+	return nil
+}
+
+var _ usecase.ProcurementRepository = (*Store)(nil)