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.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)