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

package sqlite

import (
	"context"
	"database/sql"
	"errors"
	"strings"
	"time"

	"cmroubao/backend-api/internal/domain"
	"cmroubao/backend-api/internal/usecase"
)

const createTaskOperation = "CREATE_PURCHASE_TASK"

func (s *Store) CreateTaskIdempotent(
	ctx context.Context,
	candidate domain.PurchaseTask,
	event domain.TaskEvent,
	idempotencyKey string,
	requestHash string,
) (domain.PurchaseTask, bool, error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return domain.PurchaseTask{}, false, repositoryFailure(err)
	}
	defer func() { _ = tx.Rollback() }()

	existingHash, resourceID, found, err := lookupIdempotency(
		ctx,
		tx,
		candidate.CreatorSubject,
		createTaskOperation,
		idempotencyKey,
	)
	if err != nil {
		return domain.PurchaseTask{}, false, err
	}
	if found {
		if existingHash != requestHash {
			return domain.PurchaseTask{}, false, usecase.ErrIdempotencyConflict
		}
		existing, err := getTaskByID(
			ctx,
			tx,
			candidate.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
	}

	var available int
	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
				)
		)`,
		candidate.ImageAssetID,
		candidate.CreatorSubject,
	).Scan(&available)
	if err != nil {
		return domain.PurchaseTask{}, false, repositoryFailure(err)
	}
	if available != 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, NULL, ?, ?)`,
		candidate.ID,
		candidate.CreatorSubject,
		nullableString(candidate.CreatedByUserID),
		nullableString(candidate.SourceRef),
		candidate.Title,
		candidate.Description,
		candidate.SKU,
		candidate.ImageAssetID,
		candidate.Quantity,
		nullableInt64(candidate.MaxBudgetCents),
		candidate.Currency,
		candidate.Status,
		candidate.Version,
		formatTimestamp(candidate.CreatedAt),
		formatTimestamp(candidate.UpdatedAt),
	)
	if err != nil {
		switch {
		case isUniqueConstraint(
			err,
			"purchase_tasks.creator_subject, purchase_tasks.source_ref",
		):
			return domain.PurchaseTask{}, false, usecase.ErrSourceReferenceConflict
		case isUniqueConstraint(err, "purchase_tasks.image_asset_id"):
			return domain.PurchaseTask{}, false, usecase.ErrAssetUnavailable
		default:
			return domain.PurchaseTask{}, false, repositoryFailure(err)
		}
	}
	if err := insertTaskEvent(ctx, tx, event); err != nil {
		return domain.PurchaseTask{}, false, err
	}
	if err := insertIdempotency(
		ctx,
		tx,
		candidate.CreatorSubject,
		createTaskOperation,
		idempotencyKey,
		requestHash,
		"PURCHASE_TASK",
		candidate.ID,
		candidate.CreatedAt,
	); err != nil {
		return domain.PurchaseTask{}, false, err
	}
	if err := tx.Commit(); err != nil {
		return domain.PurchaseTask{}, false, repositoryFailure(err)
	}
	return candidate, true, nil
}

func (s *Store) ListTasks(
	ctx context.Context,
	filter usecase.TaskListFilter,
) ([]domain.PurchaseTask, error) {
	var query strings.Builder
	query.WriteString(`SELECT
		id, creator_subject, created_by_user_id, source_ref, title, description, sku,
		image_asset_id, quantity, max_budget_cents, currency, status,
		version, claimed_by_user_id, claimed_by_device_id, claim_generation,
		claim_token_hash, claim_issued_at, claim_expires_at, cancel_reason,
		cancel_requested_at, cancel_requested_by_user_id, canceled_at,
		created_at, updated_at
	FROM purchase_tasks
	WHERE creator_subject = ?`)
	arguments := []any{filter.CreatorSubject}
	if filter.Status != nil {
		query.WriteString(" AND status = ?")
		arguments = append(arguments, *filter.Status)
	}
	if filter.Query != "" {
		query.WriteString(` AND (
			id LIKE ? ESCAPE '\'
			OR COALESCE(source_ref, '') LIKE ? ESCAPE '\'
			OR title LIKE ? ESCAPE '\'
			OR sku LIKE ? ESCAPE '\'
		)`)
		pattern := "%" + escapeLike(filter.Query) + "%"
		arguments = append(
			arguments,
			pattern,
			pattern,
			pattern,
			pattern,
		)
	}
	if filter.CreatedFrom != nil {
		query.WriteString(" AND created_at >= ?")
		arguments = append(
			arguments,
			formatTimestamp(*filter.CreatedFrom),
		)
	}
	if filter.CreatedTo != nil {
		query.WriteString(" AND created_at <= ?")
		arguments = append(
			arguments,
			formatTimestamp(*filter.CreatedTo),
		)
	}
	if filter.After != nil {
		query.WriteString(
			" AND (created_at < ? OR (created_at = ? AND id < ?))",
		)
		createdAt := formatTimestamp(filter.After.CreatedAt)
		arguments = append(
			arguments,
			createdAt,
			createdAt,
			filter.After.ID,
		)
	}
	query.WriteString(" ORDER BY created_at DESC, id DESC LIMIT ?")
	arguments = append(arguments, filter.Limit)

	rows, err := s.db.QueryContext(ctx, query.String(), arguments...)
	if err != nil {
		return nil, repositoryFailure(err)
	}
	defer rows.Close()
	tasks := make([]domain.PurchaseTask, 0)
	for rows.Next() {
		task, err := scanTask(rows)
		if err != nil {
			return nil, repositoryFailure(err)
		}
		tasks = append(tasks, task)
	}
	if err := rows.Err(); err != nil {
		return nil, repositoryFailure(err)
	}
	return tasks, nil
}

func (s *Store) GetTaskDetail(
	ctx context.Context,
	creatorSubject string,
	taskID string,
) (domain.TaskDetail, error) {
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
	if err != nil {
		return domain.TaskDetail{}, repositoryFailure(err)
	}
	defer func() { _ = tx.Rollback() }()
	task, err := getTaskByID(ctx, tx, creatorSubject, taskID)
	if err != nil {
		return domain.TaskDetail{}, err
	}
	asset, err := getAssetByID(
		ctx,
		tx,
		creatorSubject,
		task.ImageAssetID,
	)
	if err != nil {
		return domain.TaskDetail{}, err
	}
	rows, err := tx.QueryContext(
		ctx,
		`SELECT
			id, task_id, actor_user_id, actor_device_id,
			event_type, message, occurred_at
		FROM task_events
		WHERE task_id = ?
		ORDER BY occurred_at ASC, id ASC`,
		taskID,
	)
	if err != nil {
		return domain.TaskDetail{}, repositoryFailure(err)
	}
	defer rows.Close()
	events := make([]domain.TaskEvent, 0)
	for rows.Next() {
		var event domain.TaskEvent
		var actorUserID sql.NullString
		var actorDeviceID sql.NullString
		var occurredAt string
		if err := rows.Scan(
			&event.ID,
			&event.TaskID,
			&actorUserID,
			&actorDeviceID,
			&event.Type,
			&event.Message,
			&occurredAt,
		); err != nil {
			return domain.TaskDetail{}, repositoryFailure(err)
		}
		if actorUserID.Valid {
			event.ActorUserID = &actorUserID.String
		}
		if actorDeviceID.Valid {
			event.ActorDeviceID = &actorDeviceID.String
		}
		event.OccurredAt, err = parseTimestamp(occurredAt)
		if err != nil {
			return domain.TaskDetail{}, err
		}
		events = append(events, event)
	}
	if err := rows.Err(); err != nil {
		return domain.TaskDetail{}, repositoryFailure(err)
	}
	if err := rows.Close(); err != nil {
		return domain.TaskDetail{}, repositoryFailure(err)
	}
	execution, err := scanExecution(tx.QueryRowContext(
		ctx,
		`SELECT
			id, task_id, attempt_no, claim_generation, user_id, device_id,
			current_step, order_submitted, started_at, last_heartbeat_at,
			finished_at
		FROM task_executions
		WHERE task_id = ?
		ORDER BY attempt_no DESC
		LIMIT 1`,
		taskID,
	))
	var executionPointer *domain.TaskExecution
	if errors.Is(err, sql.ErrNoRows) {
		executionPointer = nil
	} else if err != nil {
		return domain.TaskDetail{}, repositoryFailure(err)
	} else {
		executionPointer = &execution
	}
	var report *domain.ExecutionReport
	if executionPointer != nil {
		report, err = getExecutionReport(ctx, tx, taskID, executionPointer.ID)
		if err != nil {
			return domain.TaskDetail{}, err
		}
	}
	detail := domain.TaskDetail{
		Task:      task,
		Asset:     asset,
		Execution: executionPointer,
		Events:    events,
		Report:    report,
	}
	if err := tx.Commit(); err != nil {
		return domain.TaskDetail{}, repositoryFailure(err)
	}
	return detail, nil
}

func (s *Store) CancelTask(
	ctx context.Context,
	creatorSubject string,
	taskID string,
	reason string,
	canceledAt time.Time,
	event domain.TaskEvent,
) (domain.PurchaseTask, error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return domain.PurchaseTask{}, repositoryFailure(err)
	}
	defer func() { _ = tx.Rollback() }()
	task, err := getTaskByID(ctx, tx, creatorSubject, taskID)
	if err != nil {
		return domain.PurchaseTask{}, err
	}
	if !domain.CanCancel(task.Status) {
		return domain.PurchaseTask{}, usecase.ErrTaskStateConflict
	}
	if domain.CanRequestCancel(task.Status) &&
		task.CancelRequestedAt != nil {
		if err := tx.Commit(); err != nil {
			return domain.PurchaseTask{}, repositoryFailure(err)
		}
		return task, nil
	}
	var result sql.Result
	if domain.CanAdminCancelImmediately(task.Status) {
		result, err = tx.ExecContext(
			ctx,
			`UPDATE purchase_tasks
			SET status = 'CANCELED',
				version = version + 1,
				claimed_by_user_id = NULL,
				claimed_by_device_id = NULL,
				claim_token_hash = NULL,
				claim_issued_at = NULL,
				claim_expires_at = NULL,
				cancel_reason = NULLIF(?, ''),
				cancel_requested_at = NULL,
				cancel_requested_by_user_id = NULL,
				canceled_at = ?,
				updated_at = ?
			WHERE id = ?
				AND creator_subject = ?
				AND status IN ('PENDING', 'CLAIMED')
				AND version = ?`,
			reason,
			formatTimestamp(canceledAt),
			formatTimestamp(canceledAt),
			taskID,
			creatorSubject,
			task.Version,
		)
	} else {
		event.Type = "TASK_CANCEL_REQUESTED"
		event.Message = "task cancellation requested"
		result, err = tx.ExecContext(
			ctx,
			`UPDATE purchase_tasks
			SET version = version + 1,
				cancel_reason = NULLIF(?, ''),
				cancel_requested_at = ?,
				cancel_requested_by_user_id = ?,
				updated_at = ?
			WHERE id = ?
				AND creator_subject = ?
				AND status IN ('RUNNING', 'WAITING_CONFIRMATION')
				AND cancel_requested_at IS NULL
				AND version = ?`,
			reason,
			formatTimestamp(canceledAt),
			nullableString(event.ActorUserID),
			formatTimestamp(canceledAt),
			taskID,
			creatorSubject,
			task.Version,
		)
	}
	if err != nil {
		return domain.PurchaseTask{}, repositoryFailure(err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return domain.PurchaseTask{}, repositoryFailure(err)
	}
	if affected != 1 {
		return domain.PurchaseTask{}, usecase.ErrTaskStateConflict
	}
	if err := insertTaskEvent(ctx, tx, event); err != nil {
		return domain.PurchaseTask{}, err
	}
	updated, err := getTaskByID(ctx, tx, creatorSubject, taskID)
	if err != nil {
		return domain.PurchaseTask{}, err
	}
	if err := tx.Commit(); err != nil {
		return domain.PurchaseTask{}, repositoryFailure(err)
	}
	return updated, nil
}

func insertTaskEvent(
	ctx context.Context,
	tx *sql.Tx,
	event domain.TaskEvent,
) error {
	_, err := tx.ExecContext(
		ctx,
		`INSERT INTO task_events (
			id, task_id, actor_user_id, actor_device_id,
			event_type, message, occurred_at
		) VALUES (?, ?, ?, ?, ?, ?, ?)`,
		event.ID,
		event.TaskID,
		nullableString(event.ActorUserID),
		nullableString(event.ActorDeviceID),
		event.Type,
		event.Message,
		formatTimestamp(event.OccurredAt),
	)
	if err != nil {
		return repositoryFailure(err)
	}
	return nil
}

func escapeLike(value string) string {
	replacer := strings.NewReplacer(
		`\`, `\\`,
		`%`, `\%`,
		`_`, `\_`,
	)
	return replacer.Replace(value)
}

var _ usecase.TaskRepository = (*Store)(nil)