feat(backend): implement task creation and admin web
This commit is contained in:
QiuSW
@@ -0,0 +1,383 @@
|
||||
package sqlite
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"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, 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.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, source_ref, title, description, sku,
|
||||
image_asset_id, quantity, max_budget_cents, currency, status,
|
||||
version, cancel_reason, 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, 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 occurredAt string
|
||||
if err := rows.Scan(
|
||||
&event.ID,
|
||||
&event.TaskID,
|
||||
&event.Type,
|
||||
&event.Message,
|
||||
&occurredAt,
|
||||
); err != nil {
|
||||
return domain.TaskDetail{}, repositoryFailure(err)
|
||||
}
|
||||
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)
|
||||
}
|
||||
detail := domain.TaskDetail{
|
||||
Task: task,
|
||||
Asset: asset,
|
||||
Events: events,
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return domain.TaskDetail{}, repositoryFailure(err)
|
||||
}
|
||||
return detail, nil
|
||||
}
|
||||
|
||||
func (s *Store) CancelPendingTask(
|
||||
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
|
||||
}
|
||||
result, err := tx.ExecContext(
|
||||
ctx,
|
||||
`UPDATE purchase_tasks
|
||||
SET status = 'CANCELED',
|
||||
version = version + 1,
|
||||
cancel_reason = NULLIF(?, ''),
|
||||
canceled_at = ?,
|
||||
updated_at = ?
|
||||
WHERE id = ?
|
||||
AND creator_subject = ?
|
||||
AND status = 'PENDING'
|
||||
AND version = ?`,
|
||||
reason,
|
||||
formatTimestamp(canceledAt),
|
||||
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, event_type, message, occurred_at
|
||||
) VALUES (?, ?, ?, ?, ?)`,
|
||||
event.ID,
|
||||
event.TaskID,
|
||||
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)
|
||||
Reference in New Issue
Block a user