diff --git a/internal/repository/pg_task.go b/internal/repository/pg_task.go index d027932e..b974b73a 100644 --- a/internal/repository/pg_task.go +++ b/internal/repository/pg_task.go @@ -104,6 +104,10 @@ type TaskRepository interface { GetLastEmailTime(ctx context.Context, accountID uuid.UUID) (*time.Time, error) GetScheduledTasksForAccount(ctx context.Context, accountID uuid.UUID, date time.Time) ([]Task, error) GetScheduledTasksToday(ctx context.Context, accountID uuid.UUID) ([]Task, error) + // ListDuePendingTaskIDs returns pending tasks whose scheduled_at has passed, + // oldest first, capped at limit. Drives the in-process (TASKS_PROVIDER=local) + // dispatcher, which fires each due task by id. + ListDuePendingTaskIDs(ctx context.Context, limit int) ([]uuid.UUID, error) // Update operations UpdateTaskStatus(ctx context.Context, taskID uuid.UUID, status string) error @@ -578,6 +582,35 @@ func (r *taskRepository) CancelOverduePendingTasks(ctx context.Context, taskType return tag.RowsAffected(), nil } +// ListDuePendingTaskIDs returns pending tasks whose slot has arrived, oldest +// first. The in-process dispatcher fires each; rows stay pending until their +// handler flips the status, so this is safe to call on a short interval. +func (r *taskRepository) ListDuePendingTaskIDs(ctx context.Context, limit int) ([]uuid.UUID, error) { + query := ` + SELECT id + FROM tasks + WHERE status = 'pending' + AND scheduled_at <= NOW() + ORDER BY scheduled_at ASC + LIMIT $1 + ` + rows, err := r.db.Query(ctx, query, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var ids []uuid.UUID + for rows.Next() { + var id uuid.UUID + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + // UpdateTaskScheduledAt updates the scheduled time and cloud task name func (r *taskRepository) UpdateTaskScheduledAt(ctx context.Context, taskID uuid.UUID, scheduledAt time.Time, cloudTaskName string) error { query := ` diff --git a/internal/tasksched/local.go b/internal/tasksched/local.go new file mode 100644 index 00000000..32112b22 --- /dev/null +++ b/internal/tasksched/local.go @@ -0,0 +1,87 @@ +package tasksched + +import ( + "context" + "sync" + "time" + + "github.com/google/uuid" + "github.com/rs/zerolog/log" + + "github.com/warmbly/warmbly/internal/tasks/proto" +) + +// DueTaskLister returns the ids of pending tasks whose scheduled_at has passed. +// repository.TaskRepository satisfies it. +type DueTaskLister interface { + ListDuePendingTaskIDs(ctx context.Context, limit int) ([]uuid.UUID, error) +} + +// Local is the no-cloud Scheduler. Task rows already exist in Postgres (the +// caller inserts them before CreateTask), and Run polls for due rows and fires +// them in-process. Postgres is both the store and the clock. +type Local struct { + repo DueTaskLister + interval time.Duration + batch int + inflight sync.Map // taskID string -> struct{}, dedups across overlapping ticks +} + +// NewLocal builds a Local poller. interval<=0 defaults to 1s; batch<=0 to 200. +func NewLocal(repo DueTaskLister, interval time.Duration, batch int) *Local { + if interval <= 0 { + interval = time.Second + } + if batch <= 0 { + batch = 200 + } + return &Local{repo: repo, interval: interval, batch: batch} +} + +// CreateTask is a no-op enqueue: the row already exists with status=pending and +// scheduled_at, which the poller picks up at its slot. The returned handle is a +// marker; the local provider never needs to look a task up by it. +func (l *Local) CreateTask(_ context.Context, taskData *proto.ProcessTask, _ time.Time) (string, error) { + return "local:" + taskData.TaskId, nil +} + +// DeleteTask is a no-op: cancellation flips the DB row's status first, and the +// poller only fires rows that are still pending. +func (l *Local) DeleteTask(_ context.Context, _ string) error { return nil } + +// Run polls for due tasks until ctx is cancelled, invoking handle(taskID) for +// each. handle must be idempotent: rows stay pending until handle flips their +// status, and the per-type handlers short-circuit on a non-pending status, so a +// row re-selected before handle finished is a no-op. The in-flight guard keeps +// a single node from re-dispatching a task it is already handling. +func (l *Local) Run(ctx context.Context, handle func(taskID string)) { + ticker := time.NewTicker(l.interval) + defer ticker.Stop() + log.Info().Dur("interval", l.interval).Int("batch", l.batch).Msg("tasksched: local dispatcher started") + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + l.tick(ctx, handle) + } + } +} + +func (l *Local) tick(ctx context.Context, handle func(taskID string)) { + ids, err := l.repo.ListDuePendingTaskIDs(ctx, l.batch) + if err != nil { + log.Error().Err(err).Msg("tasksched: list due tasks") + return + } + for _, id := range ids { + key := id.String() + if _, busy := l.inflight.LoadOrStore(key, struct{}{}); busy { + continue + } + go func(taskID string) { + defer l.inflight.Delete(taskID) + handle(taskID) + }(key) + } +} diff --git a/internal/tasksched/scheduler.go b/internal/tasksched/scheduler.go new file mode 100644 index 00000000..7c469b22 --- /dev/null +++ b/internal/tasksched/scheduler.go @@ -0,0 +1,33 @@ +// Package tasksched abstracts delayed task scheduling so the platform can run +// with no cloud account. A task's schedule already lives in the Postgres +// `tasks` table (status + scheduled_at); the Scheduler only decides how a due +// task gets fired. +// +// - Local (default, TASKS_PROVIDER=local): an in-process poller reads due +// rows from Postgres and dispatches them directly. No external queue, no +// webhook, no GCP. +// - gtasks.Client (TASKS_PROVIDER=gcloud): Google Cloud Tasks POSTs a webhook +// back at scheduled_at. The historical behavior, still available. +package tasksched + +import ( + "context" + "time" + + "github.com/warmbly/warmbly/internal/tasks/proto" +) + +// Scheduler enqueues a task to fire at scheduleTime and can cancel it. It is +// satisfied structurally by both *gtasks.Client and *Local, so the app services +// depend only on this interface. +type Scheduler interface { + // CreateTask registers taskData to fire at scheduleTime and returns an + // opaque handle used by DeleteTask. The caller has already written the + // `tasks` row; the local provider treats this as a no-op enqueue. + CreateTask(ctx context.Context, taskData *proto.ProcessTask, scheduleTime time.Time) (string, error) + + // DeleteTask cancels a previously created task by its handle. Cancellation + // is best-effort: callers flip the DB row's status first, so the local + // provider treats this as a no-op. + DeleteTask(ctx context.Context, name string) error +}