fix(tasks): canonicalize triggered_by so execution labels resolve

Unknown values such as http and inproc_cron made the admin UI call
t(undefined). Dispatch sites now write system/manual/retry/schedule,
the list API maps legacy rows, and the table skips missing i18n keys.
This commit is contained in:
ryan
2026-09-02 18:04:32 +08:00
parent 1f1f4efec7
commit df7ad453cc
9 changed files with 90 additions and 16 deletions
+8
View File
@@ -71,6 +71,14 @@ type TaskExecutionDTO struct {
UpdatedAt time.Time `json:"updated_at"` UpdatedAt time.Time `json:"updated_at"`
} }
// Canonical triggered_by values persisted on task executions and shown in admin UI.
const (
TaskTriggerSystem = "system"
TaskTriggerManual = "manual"
TaskTriggerRetry = "retry"
TaskTriggerSchedule = "schedule"
)
// TaskService defines the unified contract for dispatching and tracking background tasks. // TaskService defines the unified contract for dispatching and tracking background tasks.
type TaskService interface { type TaskService interface {
Dispatch(ctx context.Context, taskType string, payload []byte, triggeredBy string) (string, error) Dispatch(ctx context.Context, taskType string, payload []byte, triggeredBy string) (string, error)
+17 -2
View File
@@ -41,7 +41,7 @@ func DispatchTask(ctx context.Context, req model.DispatchTaskRequest) (string, e
return "", err return "", err
} }
taskID, err := taskSvc.Dispatch(ctx, req.TaskType, validated, "manual") taskID, err := taskSvc.Dispatch(ctx, req.TaskType, validated, contracts.TaskTriggerManual)
if err != nil { if err != nil {
return "", fmt.Errorf("%s: %w", errs.TaskDispatchFailed, err) return "", fmt.Errorf("%s: %w", errs.TaskDispatchFailed, err)
} }
@@ -110,6 +110,21 @@ func TaskExecution(ctx context.Context, id uint64) (*model.TaskExecution, error)
return &row, nil return &row, nil
} }
func normalizeTaskTrigger(v string) string {
switch v {
case contracts.TaskTriggerManual, contracts.TaskTriggerSystem, contracts.TaskTriggerRetry, contracts.TaskTriggerSchedule:
return v
case "inproc_cron", "cron":
return contracts.TaskTriggerSchedule
case "http":
return contracts.TaskTriggerSystem
case "":
return contracts.TaskTriggerSystem
default:
return v
}
}
func executionFromDTO(dto contracts.TaskExecutionDTO) model.TaskExecution { func executionFromDTO(dto contracts.TaskExecutionDTO) model.TaskExecution {
return model.TaskExecution{ return model.TaskExecution{
ID: dto.ID, ID: dto.ID,
@@ -127,7 +142,7 @@ func executionFromDTO(dto contracts.TaskExecutionDTO) model.TaskExecution {
FinishedAt: dto.FinishedAt, FinishedAt: dto.FinishedAt,
Duration: dto.Duration, Duration: dto.Duration,
Payload: dto.Payload, Payload: dto.Payload,
TriggeredBy: dto.TriggeredBy, TriggeredBy: normalizeTaskTrigger(dto.TriggeredBy),
CreatedAt: dto.CreatedAt, CreatedAt: dto.CreatedAt,
UpdatedAt: dto.UpdatedAt, UpdatedAt: dto.UpdatedAt,
} }
@@ -0,0 +1,32 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package service
import (
"Wavelet/core/contracts"
"testing"
)
func TestNormalizeTaskTrigger(t *testing.T) {
tests := []struct {
in string
want string
}{
{in: contracts.TaskTriggerManual, want: contracts.TaskTriggerManual},
{in: contracts.TaskTriggerSystem, want: contracts.TaskTriggerSystem},
{in: contracts.TaskTriggerRetry, want: contracts.TaskTriggerRetry},
{in: contracts.TaskTriggerSchedule, want: contracts.TaskTriggerSchedule},
{in: "http", want: contracts.TaskTriggerSystem},
{in: "inproc_cron", want: contracts.TaskTriggerSchedule},
{in: "cron", want: contracts.TaskTriggerSchedule},
{in: "", want: contracts.TaskTriggerSystem},
{in: "custom", want: "custom"},
}
for _, tt := range tests {
got := normalizeTaskTrigger(tt.in)
if got != tt.want {
t.Errorf("normalizeTaskTrigger(%q) = %q, want %q", tt.in, got, tt.want)
}
}
}
@@ -697,7 +697,7 @@ func EnqueuePushTask(ctx context.Context, payload model.SendPayload) error {
return err return err
} }
if taskSvc := GetTaskService(ctx); taskSvc != nil { if taskSvc := GetTaskService(ctx); taskSvc != nil {
_, err = taskSvc.Dispatch(ctx, "send_notification", payloadBytes, "system") _, err = taskSvc.Dispatch(ctx, "send_notification", payloadBytes, contracts.TaskTriggerSystem)
return err return err
} }
return errors.New(errs.ErrTaskServiceUnavailable) return errors.New(errs.ErrTaskServiceUnavailable)
+1 -1
View File
@@ -209,7 +209,7 @@ func SendEmailCode(c *gin.Context) {
} }
ctx := c.Request.Context() ctx := c.Request.Context()
if taskSvc := getTaskService(ctx); taskSvc != nil { if taskSvc := getTaskService(ctx); taskSvc != nil {
if _, err := taskSvc.Dispatch(ctx, TaskTypeSendEmailCode, payload, "http"); err != nil { if _, err := taskSvc.Dispatch(ctx, TaskTypeSendEmailCode, payload, contracts.TaskTriggerSystem); err != nil {
logger.ErrorF(ctx, "dispatch send_email_code failed: %v", err) logger.ErrorF(ctx, "dispatch send_email_code failed: %v", err)
response.AbortInternal(c, errSendEmailFailed) response.AbortInternal(c, errSendEmailFailed)
return return
@@ -4,6 +4,7 @@
package driver_asynq_worker package driver_asynq_worker
import ( import (
"Wavelet/core/contracts"
"Wavelet/pkg/idgen" "Wavelet/pkg/idgen"
"Wavelet/pkg/logger" "Wavelet/pkg/logger"
"Wavelet/pkg/util" "Wavelet/pkg/util"
@@ -208,7 +209,7 @@ func RetryTask(ctx context.Context, id uint64) (string, error) {
MaxRetry: execution.MaxRetry, MaxRetry: execution.MaxRetry,
RetryCount: execution.RetryCount + 1, RetryCount: execution.RetryCount + 1,
Payload: execution.Payload, Payload: execution.Payload,
TriggeredBy: "retry", TriggeredBy: contracts.TaskTriggerRetry,
} }
if err := createTaskExecution(ctx, newExecution); err != nil { if err := createTaskExecution(ctx, newExecution); err != nil {
@@ -393,7 +394,7 @@ func getOrCreateTaskExecution(ctx context.Context, taskID string, t *asynq.Task,
MaxRetry: meta.MaxRetry, MaxRetry: meta.MaxRetry,
RetryCount: 0, RetryCount: 0,
Payload: string(payload), Payload: string(payload),
TriggeredBy: "schedule", TriggeredBy: contracts.TaskTriggerSchedule,
StartedAt: &now, StartedAt: &now,
} }
@@ -99,7 +99,7 @@ func (s *inprocScheduler) registerJob(ctx context.Context, def extpoints.Schedul
_, err := s.cronRunner.AddFunc(cronSpec, func() { _, err := s.cronRunner.AddFunc(cronSpec, func() {
if s.taskSvc != nil { if s.taskSvc != nil {
if _, dispatchErr := s.taskSvc.Dispatch(ctx, taskType, payloadBytes, "inproc_cron"); dispatchErr != nil { if _, dispatchErr := s.taskSvc.Dispatch(ctx, taskType, payloadBytes, contracts.TaskTriggerSchedule); dispatchErr != nil {
logger.ErrorF(ctx, "driver_inproc_cron: dispatch task %q failed: %v", taskType, dispatchErr) logger.ErrorF(ctx, "driver_inproc_cron: dispatch task %q failed: %v", taskType, dispatchErr)
} }
return return
@@ -76,7 +76,7 @@ func (q *InprocQueue) Enqueue(ctx context.Context, taskType string, payload []by
} }
if source == "" { if source == "" {
source = "manual" source = contracts.TaskTriggerManual
} }
idType := td.Type idType := td.Type
if idType == "" { if idType == "" {
@@ -85,6 +85,16 @@ function statusVariant(status: TaskExecutionStatus) {
return 'outline'; return 'outline';
} }
function mappedLabel(
t: (key: string) => string,
keys: Record<string, string>,
value: string,
): string {
const key = keys[value];
if (!key) return value || '-';
return t(key);
}
export function TaskExecutionsManager() { export function TaskExecutionsManager() {
const t = useTranslations('admin.tasks'); const t = useTranslations('admin.tasks');
const queryClient = useQueryClient(); const queryClient = useQueryClient();
@@ -281,14 +291,16 @@ export function TaskExecutionsManager() {
</TableCell> </TableCell>
<TableCell> <TableCell>
<Badge variant={statusVariant(execution.status)}> <Badge variant={statusVariant(execution.status)}>
{t(STATUS_LABELS_KEYS[execution.status]) || {mappedLabel(t, STATUS_LABELS_KEYS, execution.status)}
execution.status}
</Badge> </Badge>
</TableCell> </TableCell>
<TableCell> <TableCell>
<Badge variant='outline'> <Badge variant='outline'>
{t(TRIGGER_LABELS_KEYS[execution.triggered_by]) || {mappedLabel(
execution.triggered_by} t,
TRIGGER_LABELS_KEYS,
execution.triggered_by,
)}
</Badge> </Badge>
</TableCell> </TableCell>
<TableCell className='font-mono text-xs text-muted-foreground'> <TableCell className='font-mono text-xs text-muted-foreground'>
@@ -368,8 +380,11 @@ export function TaskExecutionsManager() {
</div> </div>
<div className='mt-2'> <div className='mt-2'>
<Badge variant={statusVariant(selectedExecution.status)}> <Badge variant={statusVariant(selectedExecution.status)}>
{t(STATUS_LABELS_KEYS[selectedExecution.status]) || {mappedLabel(
selectedExecution.status} t,
STATUS_LABELS_KEYS,
selectedExecution.status,
)}
</Badge> </Badge>
</div> </div>
</div> </div>
@@ -378,8 +393,11 @@ export function TaskExecutionsManager() {
{t('detailTrigger')} {t('detailTrigger')}
</div> </div>
<div className='mt-2 text-sm font-medium'> <div className='mt-2 text-sm font-medium'>
{t(TRIGGER_LABELS_KEYS[selectedExecution.triggered_by]) || {mappedLabel(
selectedExecution.triggered_by} t,
TRIGGER_LABELS_KEYS,
selectedExecution.triggered_by,
)}
</div> </div>
</div> </div>
<div className='rounded-lg border p-3'> <div className='rounded-lg border p-3'>