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

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.original_unit_price_minor,
			item.original_currency, 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,
	auto *usecase.AutoProcurementReference,
) (domain.ProcurementRequest, bool, bool, error) {
	tx, err := store.db.BeginTx(ctx, nil)
	if err != nil {
		return domain.ProcurementRequest{}, false, false,
			repositoryFailure(err)
	}
	defer tx.Rollback()
	existing, err := getProcurementRequestBySource(
		ctx,
		tx,
		candidate.CreatorSubject,
		candidate.FreightOrderItemID,
		candidate.SourceRevision,
	)
	if err == nil {
		if auto != nil &&
			existing.Status == domain.ProcurementNeedsImage {
			updated, bindErr := bindAutomaticProcurementReference(
				ctx,
				tx,
				existing,
				*auto,
				candidate.UpdatedAt,
			)
			if bindErr != nil {
				return domain.ProcurementRequest{}, false, false, bindErr
			}
			if err := tx.Commit(); err != nil {
				return domain.ProcurementRequest{}, false, false,
					repositoryFailure(err)
			}
			return updated, false, true, nil
		}
		if err := tx.Commit(); err != nil {
			return domain.ProcurementRequest{}, false, false,
				repositoryFailure(err)
		}
		return existing, false, false, nil
	}
	if !errors.Is(err, usecase.ErrRepositoryNotFound) {
		return domain.ProcurementRequest{}, false, false, err
	}
	if err := validateProcurementSource(ctx, tx, candidate); err != nil {
		return domain.ProcurementRequest{}, false, false, err
	}
	referenceUsed := false
	if auto != nil {
		if candidate.Status != domain.ProcurementNeedsImage {
			return domain.ProcurementRequest{}, false, false,
				repositoryFailure(usecase.ErrRepositoryInvariant)
		}
		if err := validateAutomaticProcurementReference(
			ctx,
			tx,
			candidate,
			*auto,
		); err != nil {
			return domain.ProcurementRequest{}, false, false, err
		}
		if err := insertProcurementReferenceAsset(
			ctx,
			tx,
			auto.Asset,
		); err != nil {
			return domain.ProcurementRequest{}, false, false, err
		}
		candidate.ReferenceAssetID = &auto.Asset.ID
		candidate.Status = domain.ProcurementReady
		referenceUsed = true
	}
	_, 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, ?, ?)`,
		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),
		nullableString(candidate.ReferenceAssetID),
		formatTimestamp(candidate.CreatedAt),
		formatTimestamp(candidate.UpdatedAt),
	)
	if err != nil {
		return domain.ProcurementRequest{}, false, false,
			repositoryFailure(err)
	}
	if auto != nil {
		if err := saveProcurementReferenceSource(
			ctx,
			tx,
			candidate.ID,
			domain.ProcurementReferenceSource{
				Origin:            domain.ProcurementReferenceERP,
				ProductThumbRef:   &auto.ProductThumbRef,
				SourceImageSHA256: &auto.SourceImageSHA256,
				BoundAt:           candidate.UpdatedAt,
			},
		); err != nil {
			return domain.ProcurementRequest{}, false, false, err
		}
	}
	stored, err := getProcurementRequest(
		ctx,
		tx,
		candidate.CreatorSubject,
		candidate.ID,
	)
	if err != nil {
		return domain.ProcurementRequest{}, false, false, err
	}
	if err := tx.Commit(); err != nil {
		return domain.ProcurementRequest{}, false, false,
			repositoryFailure(err)
	}
	return stored, true, referenceUsed, 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, *string, error) {
	tx, err := store.db.BeginTx(ctx, nil)
	if err != nil {
		return domain.ProcurementRequest{}, nil, repositoryFailure(err)
	}
	defer tx.Rollback()
	request, err := getProcurementRequest(
		ctx,
		tx,
		creatorSubject,
		requestID,
	)
	if err != nil {
		return domain.ProcurementRequest{}, nil, err
	}
	if request.SourceChanged {
		if request.PurchaseTaskID == nil {
			if err := markProcurementSourceChanged(ctx, tx, request.ID, now); err != nil {
				return domain.ProcurementRequest{}, nil, err
			}
		}
		if err := tx.Commit(); err != nil {
			return domain.ProcurementRequest{}, nil, repositoryFailure(err)
		}
		return domain.ProcurementRequest{}, nil,
			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{}, nil, repositoryFailure(err)
		}
		return request, nil, nil
	}
	if request.Status != domain.ProcurementNeedsImage &&
		request.Status != domain.ProcurementReady {
		return domain.ProcurementRequest{}, nil,
			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{}, nil, repositoryFailure(err)
	}
	if available != 1 {
		return domain.ProcurementRequest{}, nil, usecase.ErrAssetUnavailable
	}
	var replacedStorageKey *string
	var replacedAssetID *string
	if request.ReferenceAssetID != nil &&
		request.ReferenceOrigin != nil &&
		*request.ReferenceOrigin == domain.ProcurementReferenceERP {
		replaced, getErr := getAssetByID(
			ctx,
			tx,
			creatorSubject,
			*request.ReferenceAssetID,
		)
		if getErr != nil {
			return domain.ProcurementRequest{}, nil, getErr
		}
		replacedStorageKey = &replaced.StorageKey
		replacedAssetID = &replaced.ID
	}
	if _, err := tx.ExecContext(
		ctx,
		`UPDATE procurement_requests
		SET reference_asset_id = ?, status = 'READY', blocking_code = NULL,
			updated_at = ?
		WHERE id = ? AND status IN ('NEEDS_IMAGE', 'READY')`,
		assetID,
		formatTimestamp(now),
		requestID,
	); err != nil {
		return domain.ProcurementRequest{}, nil, repositoryFailure(err)
	}
	if err := saveProcurementReferenceSource(
		ctx,
		tx,
		requestID,
		domain.ProcurementReferenceSource{
			Origin:  domain.ProcurementReferenceManual,
			BoundAt: now,
		},
	); err != nil {
		return domain.ProcurementRequest{}, nil, err
	}
	if replacedAssetID != nil {
		if _, err := tx.ExecContext(
			ctx,
			`DELETE FROM assets WHERE id = ?`,
			*replacedAssetID,
		); err != nil {
			return domain.ProcurementRequest{}, nil, repositoryFailure(err)
		}
	}
	updated, err := getProcurementRequest(ctx, tx, creatorSubject, requestID)
	if err != nil {
		return domain.ProcurementRequest{}, nil, err
	}
	if err := tx.Commit(); err != nil {
		return domain.ProcurementRequest{}, nil, repositoryFailure(err)
	}
	return updated, replacedStorageKey, 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,
	reference_source.origin, reference_source.product_thumb_ref,
	reference_source.source_image_sha256, reference_source.bound_at,
	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
	LEFT JOIN procurement_reference_sources AS reference_source
		ON reference_source.procurement_request_id = request.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, referenceOrigin sql.NullString
	var referenceThumb, referenceSHA, referenceBoundAt sql.NullString
	var 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,
		&referenceOrigin,
		&referenceThumb,
		&referenceSHA,
		&referenceBoundAt,
		&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)
	if referenceOrigin.Valid {
		origin := domain.ProcurementReferenceOrigin(referenceOrigin.String)
		request.ReferenceOrigin = &origin
	}
	request.ReferenceProductThumbRef = optionalString(referenceThumb)
	request.ReferenceSourceImageSHA256 = optionalString(referenceSHA)
	if referenceBoundAt.Valid {
		boundAt, parseErr := parseTimestamp(referenceBoundAt.String)
		if parseErr != nil {
			return domain.ProcurementRequest{}, parseErr
		}
		request.ReferenceBoundAt = &boundAt
	}
	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 validateProcurementSource(
	ctx context.Context,
	tx *sql.Tx,
	candidate domain.ProcurementRequest,
) error {
	var revision int
	var sourceSHA string
	var present bool
	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(&revision, &sourceSHA, &present)
	if errors.Is(err, sql.ErrNoRows) {
		return usecase.ErrRepositoryNotFound
	}
	if err != nil {
		return repositoryFailure(err)
	}
	if !present || revision != candidate.SourceRevision ||
		sourceSHA != candidate.SourceSHA256 {
		return usecase.ErrProcurementSourceChanged
	}
	return nil
}

func validateAutomaticProcurementReference(
	ctx context.Context,
	tx *sql.Tx,
	candidate domain.ProcurementRequest,
	auto usecase.AutoProcurementReference,
) error {
	if auto.Asset.CreatorSubject != candidate.CreatorSubject ||
		auto.Asset.Purpose != domain.AssetPurposeTaskReference ||
		auto.Asset.MediaType != domain.NormalizedImageMediaType ||
		auto.Asset.ID == "" || auto.Asset.StorageKey == "" ||
		auto.Asset.SizeBytes <= 0 || len(auto.Asset.SHA256) != 64 {
		return usecase.ErrRepositoryInvariant
	}
	var thumbRef, sourceSHA, storageKey string
	err := tx.QueryRowContext(
		ctx,
		`SELECT image.product_thumb_ref, image.sha256, image.storage_key
		FROM freight_item_images AS image
		JOIN freight_order_items AS item
			ON item.id = image.freight_order_item_id
		JOIN freight_orders AS freight
			ON freight.id = item.freight_order_id
		WHERE freight.creator_subject = ?
			AND item.id = ?
			AND item.is_present = 1
			AND item.product_thumb_ref = image.product_thumb_ref
			AND image.status = 'READY'`,
		candidate.CreatorSubject,
		candidate.FreightOrderItemID,
	).Scan(&thumbRef, &sourceSHA, &storageKey)
	if errors.Is(err, sql.ErrNoRows) {
		return usecase.ErrProcurementSourceChanged
	}
	if err != nil {
		return repositoryFailure(err)
	}
	if thumbRef != auto.ProductThumbRef ||
		sourceSHA != auto.SourceImageSHA256 ||
		storageKey != auto.SourceImageStorageKey {
		return usecase.ErrProcurementSourceChanged
	}
	return nil
}

func insertProcurementReferenceAsset(
	ctx context.Context,
	tx *sql.Tx,
	asset domain.Asset,
) error {
	_, err := tx.ExecContext(
		ctx,
		`INSERT INTO assets (
			id, creator_subject, purpose, media_type, size_bytes,
			sha256, storage_key, created_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
		asset.ID,
		asset.CreatorSubject,
		asset.Purpose,
		asset.MediaType,
		asset.SizeBytes,
		asset.SHA256,
		asset.StorageKey,
		formatTimestamp(asset.CreatedAt),
	)
	if err != nil {
		return repositoryFailure(err)
	}
	return nil
}

func saveProcurementReferenceSource(
	ctx context.Context,
	tx *sql.Tx,
	requestID string,
	source domain.ProcurementReferenceSource,
) error {
	_, err := tx.ExecContext(
		ctx,
		`INSERT INTO procurement_reference_sources (
			procurement_request_id, origin, product_thumb_ref,
			source_image_sha256, bound_at
		) VALUES (?, ?, ?, ?, ?)
		ON CONFLICT(procurement_request_id) DO UPDATE SET
			origin = excluded.origin,
			product_thumb_ref = excluded.product_thumb_ref,
			source_image_sha256 = excluded.source_image_sha256,
			bound_at = excluded.bound_at`,
		requestID,
		source.Origin,
		nullableString(source.ProductThumbRef),
		nullableString(source.SourceImageSHA256),
		formatTimestamp(source.BoundAt),
	)
	if err != nil {
		return repositoryFailure(err)
	}
	return nil
}

func bindAutomaticProcurementReference(
	ctx context.Context,
	tx *sql.Tx,
	request domain.ProcurementRequest,
	auto usecase.AutoProcurementReference,
	now time.Time,
) (domain.ProcurementRequest, error) {
	if request.SourceChanged {
		return domain.ProcurementRequest{},
			usecase.ErrProcurementSourceChanged
	}
	if err := validateProcurementSource(ctx, tx, request); err != nil {
		return domain.ProcurementRequest{}, err
	}
	if err := validateAutomaticProcurementReference(
		ctx,
		tx,
		request,
		auto,
	); err != nil {
		return domain.ProcurementRequest{}, err
	}
	if err := insertProcurementReferenceAsset(ctx, tx, auto.Asset); err != nil {
		return domain.ProcurementRequest{}, err
	}
	result, err := tx.ExecContext(
		ctx,
		`UPDATE procurement_requests
		SET reference_asset_id = ?, status = 'READY',
			blocking_code = NULL, updated_at = ?
		WHERE id = ? AND status = 'NEEDS_IMAGE'
			AND reference_asset_id IS NULL`,
		auto.Asset.ID,
		formatTimestamp(now),
		request.ID,
	)
	if err != nil {
		return domain.ProcurementRequest{}, repositoryFailure(err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return domain.ProcurementRequest{}, repositoryFailure(err)
	}
	if affected != 1 {
		return domain.ProcurementRequest{},
			usecase.ErrProcurementStateConflict
	}
	if err := saveProcurementReferenceSource(
		ctx,
		tx,
		request.ID,
		domain.ProcurementReferenceSource{
			Origin:            domain.ProcurementReferenceERP,
			ProductThumbRef:   &auto.ProductThumbRef,
			SourceImageSHA256: &auto.SourceImageSHA256,
			BoundAt:           now,
		},
	); err != nil {
		return domain.ProcurementRequest{}, err
	}
	return getProcurementRequest(
		ctx,
		tx,
		request.CreatorSubject,
		request.ID,
	)
}

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)