feat: add an in-process postgres task scheduler so delayed sends need no cloud tasks

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 := `

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)
+	}
+}

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
+}