summercms/modules/conga/commands.go

package conga

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"slices"
	"strconv"
	"strings"
	"syscall"
	"time"

	"git.golem15.com/golem15/summercms/modules/backpack"
	"git.golem15.com/golem15/summercms/modules/bonfire"
	"git.golem15.com/golem15/summercms/modules/diary"
	"git.golem15.com/golem15/summercms/modules/lagoon"
	"git.golem15.com/golem15/summercms/modules/pact"
	"git.golem15.com/golem15/summercms/modules/party"
	"github.com/riverqueue/river"
	"github.com/riverqueue/river/rivertype"
	"gorm.io/gorm"
)

// clearBatch is the number of jobs one JobDeleteMany pass removes.
const clearBatch = 10000

// RuntimeCommands returns the queue:work, schedule:run and queue:clear
// commands for the application binary.
func RuntimeCommands(app *backpack.App, plugins []party.Plugin) []bonfire.Command {
	return []bonfire.Command{
		{
			Name:        "queue:work",
			Description: "Run background job workers in the foreground",
			Long: `Starts one River worker on the shared database pool for every known queue, or
for the queues named with --queue, and runs jobs until Ctrl+C or SIGTERM. It
prints a startup summary (environment, queues with their concurrency, database
host, whether the scheduled queue runs here) on a terminal, and logs it as a
"startup summary" record of component queue otherwise.

On a signal it stops fetching jobs and waits up to 10 seconds for running
ones; jobs still running then are cancelled and River retries them later.
It prints a shutdown report (jobs done, failed, handed back to the queue,
uptime) and exits 0, or 1 when the worker fails to stop.

serve runs the same worker in-process unless queue.work_in_serve is false;
run queue:work as a separate process when it is.`,
			Examples: []bonfire.Example{
				{Command: "queue:work", Description: "Work every known queue"},
				{Command: "queue:work --queue mail --queue default", Description: "Work only the mail and default queues"},
			},
			Flags: []bonfire.Flag{{
				Name:        "queue",
				Description: "Queue to work (repeatable; default all known queues)",
				Repeatable:  true,
			}},
			Run: func(ctx context.Context, in bonfire.Input, out bonfire.Output) error {
				return withDB(ctx, app, func() error {
					return work(ctx, app, plugins, in.Flags("queue"), out)
				})
			},
		},
		{
			Name:        "schedule:run",
			Description: "Run the scheduler in the foreground (or once for system cron)",
			Long: `Without --once it runs a worker on the scheduled queue that also enqueues every
plugin's schedule entries, prints a startup summary (entry count, the next
due entry) and runs until Ctrl+C or SIGTERM, then prints a report (scheduled
commands run and failed, uptime) and exits 0.

With --once it runs the entries due in the current minute in app.timezone,
one after another, ends with "Ran N due entries (M failed)" and exits 1 when
any of them failed. Use it from system cron when no worker runs the
scheduled queue.`,
			Examples: []bonfire.Example{
				{Command: "schedule:run", Description: "Run the scheduler in the foreground"},
				{Command: "schedule:run --once", Description: "From cron, every minute: run the entries due now and exit"},
			},
			Flags: []bonfire.Flag{{
				Name:        "once",
				Description: "Run the entries due this minute and exit (for system cron)",
				Bare:        true,
			}},
			Run: func(ctx context.Context, in bonfire.Input, out bonfire.Output) error {
				if v, ok := in.Flag("once"); ok {
					if once, _ := strconv.ParseBool(v); once {
						return scheduleRunOnce(ctx, app, plugins, scheduleNow(), out)
					}
				}
				return withDB(ctx, app, func() error {
					return runScheduler(ctx, app, plugins, out)
				})
			},
		},
		{
			Name:        "queue:clear",
			Description: "Clear pending jobs from a queue",
			Long: `Deletes the available, scheduled and retryable jobs of one queue (default
"default") in batches until none are left, and prints how many it deleted.
Running jobs are never touched. In production, on an interactive terminal, it
asks first; --force skips the question. Off a terminal it never asks.`,
			Args: []bonfire.Arg{{
				Name:        "queue",
				Description: "The name of the queue to clear (default \"default\")",
			}},
			Flags: []bonfire.Flag{{
				Name:        "force",
				Description: "Skip the confirmation asked in production on a terminal",
				Bare:        true,
			}},
			Examples: []bonfire.Example{
				{Command: "queue:clear", Description: "Clear the default queue"},
				{Command: "queue:clear imports --force", Description: "Clear the imports queue without the production question"},
			},
			Run: func(ctx context.Context, in bonfire.Input, out bonfire.Output) error {
				queue, _ := in.Argument("queue")
				queue = strings.TrimSpace(queue)
				if queue == "" {
					queue = defaultQueue
				}
				if err := confirmProduction(app, in, out, fmt.Sprintf("Delete every pending job of queue %q in production?", queue)); err != nil {
					return err
				}
				return withDB(ctx, app, func() error {
					return clearQueue(ctx, app, queue, out)
				})
			},
		},
	}
}

// conga's console loggers: the startup summaries and shutdown reports of
// queue:work and schedule:run.
var (
	queueLog    = diary.For("queue")
	scheduleLog = diary.For("schedule")
)

// announce prints rows as a block on a terminal (not under -q) and logs them
// as one record msg; the record is hidden from a terminal console when the
// block was printed (diary.HideOnTTY).
func announce(ctx context.Context, out bonfire.Output, log *slog.Logger, level slog.Level, title, msg string, rows []bonfire.Detail, attrs []slog.Attr) {
	if bonfire.IsTerminal(out) && !bonfire.IsQuiet(out) {
		bonfire.Details(out, title, rows)
		ctx = diary.HideOnTTY(ctx)
	}
	log.LogAttrs(ctx, level, msg, attrs...)
}

// workerReport is the shutdown report of a worker: its job counters, the
// uptime and the stop error.
func workerReport(w *Worker, started time.Time, stopErr error) ([]bonfire.Detail, []slog.Attr, slog.Level) {
	st := w.Stats()
	errText := "none"
	level := slog.LevelInfo
	if stopErr != nil {
		errText = stopErr.Error()
		level = slog.LevelError
	}
	up := time.Since(started).Round(time.Second)
	rows := []bonfire.Detail{
		{Label: "Jobs", Value: fmt.Sprintf("%d done, %d failed, %d handed back to the queue", st.Done, st.Failed, st.HandedBack+st.InFlight), Warn: st.Failed > 0},
		{Label: "Uptime", Value: up.String()},
		{Label: "Errors", Value: errText, Warn: stopErr != nil},
	}
	attrs := []slog.Attr{
		slog.Int64("jobs_done", st.Done),
		slog.Int64("jobs_failed", st.Failed),
		slog.Int64("jobs_handed_back", st.HandedBack+st.InFlight),
		slog.Int64("uptime_ms", time.Since(started).Milliseconds()),
	}
	if stopErr != nil {
		attrs = append(attrs, slog.String("errors", errText))
	}
	return rows, attrs, level
}

// queueList is "default (10), mail (2)": each queue with its MaxWorkers.
func queueList(w *Worker) string {
	parts := make([]string, 0, len(w.queues))
	for _, q := range w.queues {
		parts = append(parts, fmt.Sprintf("%s (%d)", q, w.concurrency[q]))
	}
	return strings.Join(parts, ", ")
}

func work(ctx context.Context, app *backpack.App, plugins []party.Plugin, queues []string, out bonfire.Output) error {
	started := time.Now()
	ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
	defer stop()
	w, err := StartWorker(ctx, app, plugins, WorkerOptions{Queues: queues})
	if err != nil {
		return err
	}
	scheduler := "not in this worker (the " + QueueScheduled + " queue is not selected)"
	if slices.Contains(w.queues, QueueScheduled) {
		scheduler = "included (the " + QueueScheduled + " queue runs scheduled commands)"
	}
	env := "unknown"
	if app.Config != nil && app.Config.Environment() != "" {
		env = app.Config.Environment()
	}
	db := lagoon.DatabaseLabel(lagoon.DSN(app.Config))
	announce(ctx, out, queueLog, slog.LevelInfo, "Worker started", "startup summary", []bonfire.Detail{
		{Label: "Environment", Value: env},
		{Label: "Queues", Value: queueList(w)},
		{Label: "Database", Value: db},
		{Label: "Scheduler", Value: scheduler},
	}, []slog.Attr{
		slog.String("environment", env),
		slog.String("queues", strings.Join(w.queues, ",")),
		slog.String("database", db),
		slog.String("scheduler", scheduler),
	})
	<-ctx.Done()
	stopCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	stopErr := w.Stop(stopCtx)
	rows, attrs, level := workerReport(w, started, stopErr)
	announce(context.WithoutCancel(ctx), out, queueLog, level, "Shutdown report", "shutdown report", rows, attrs)
	return stopErr
}

// nextDue is the earliest next run of entries after now, and its command.
func nextDue(entries []scheduleEntry, now time.Time) (time.Time, string) {
	var at time.Time
	var cmd string
	for _, e := range entries {
		if e.schedule == nil {
			continue
		}
		if n := e.schedule.Next(now); at.IsZero() || n.Before(at) {
			at, cmd = n, e.cmd.Command
		}
	}
	return at, cmd
}

// runScheduler runs a scheduler-only worker (the scheduled queue plus every
// plugin's periodic jobs) until SIGINT or SIGTERM.
func runScheduler(ctx context.Context, app *backpack.App, plugins []party.Plugin, out bonfire.Output) error {
	started := time.Now()
	ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
	defer stop()
	w, err := StartWorker(ctx, app, plugins, WorkerOptions{Queues: []string{QueueScheduled}})
	if err != nil {
		return err
	}
	entries, _ := scheduleEntries(app, plugins)
	next := "none"
	if at, cmd := nextDue(entries, time.Now()); cmd != "" {
		next = cmd + " at " + at.Format("2006-01-02 15:04 MST")
	}
	announce(ctx, out, scheduleLog, slog.LevelInfo, "Scheduler started", "startup summary", []bonfire.Detail{
		{Label: "Entries", Value: fmt.Sprint(len(entries))},
		{Label: "Next", Value: next},
	}, []slog.Attr{
		slog.Int("entries", len(entries)),
		slog.String("next", next),
	})
	<-ctx.Done()
	stopCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	stopErr := w.Stop(stopCtx)
	st := w.Stats()
	rows, attrs, level := workerReport(w, started, stopErr)
	rows[0] = bonfire.Detail{Label: "Runs", Value: fmt.Sprintf("%d scheduled commands (%d failed)", st.Done+st.Failed, st.Failed), Warn: st.Failed > 0}
	announce(context.WithoutCancel(ctx), out, scheduleLog, level, "Shutdown report", "shutdown report", rows, attrs)
	return stopErr
}

// scheduleRunOnce runs, without River, every schedule entry due in the
// minute of now in app.timezone (Laravel schedule:run for system cron). An
// unregistered command is warned about and skipped; the returned error is the
// first command error, after every due entry ran. It ends with "Ran N due
// entries (M failed)".
func scheduleRunOnce(ctx context.Context, app *backpack.App, plugins []party.Plugin, now time.Time, out bonfire.Output) error {
	entries, err := scheduleEntries(app, plugins)
	if err != nil {
		return err
	}
	loc, err := appLocation(app)
	if err != nil {
		return err
	}
	local := now.In(loc)
	var due []scheduleEntry
	for _, e := range entries {
		if dueAt(e.cmd.Cadence, local) {
			due = append(due, e)
		}
	}
	if len(due) == 0 {
		out.Println("No scheduled commands are ready to run.")
		return nil
	}
	return withAppDB(ctx, app, func() error {
		log := loggerFromApp(app)
		var first error
		ran, failed := 0, 0
		for _, e := range due {
			if _, ok := scheduledCatalog(app, log, e.cmd.Command); !ok {
				out.Warning(fmt.Sprintf("Scheduled command %s is not registered; skipping", e.cmd.Command))
				continue
			}
			out.Printf("Running scheduled command: %s\n", strings.Join(append([]string{e.cmd.Command}, e.cmd.Args...), " "))
			ran++
			if _, err := callScheduled(ctx, app, log, e.cmd.Command, e.cmd.Args, out); err != nil {
				failed++
				if first == nil {
					first = err
				}
			}
		}
		out.Info(fmt.Sprintf("Ran %d due entries (%d failed)", ran, failed))
		return first
	})
}

// dueAt reports whether a cadence is due in the wall-clock minute of t: a
// daily cadence at its hour and minute, Every(d) of a minute or less on every
// run, and a longer Every(d) when the minutes since midnight are a multiple
// of d.
func dueAt(c pact.Cadence, t time.Time) bool {
	if h, m, ok := c.At(); ok {
		return t.Hour() == h && t.Minute() == m
	}
	d := c.Interval()
	if d <= 0 {
		return false
	}
	if d <= time.Minute {
		return true
	}
	sinceMidnight := time.Duration(t.Hour()*60+t.Minute()) * time.Minute
	return sinceMidnight%d == 0
}

// withAppDB runs fn with the app's database: the published one when there is
// one, otherwise one opened for the command the way withDB does.
func withAppDB(ctx context.Context, app *backpack.App, fn func() error) error {
	if gdb, ok := app.Lookup[*gorm.DB](); ok && gdb != nil {
		return fn()
	}
	return withDB(ctx, app, fn)
}

// clearQueue deletes the available, scheduled and retryable jobs of queue in
// batches until a pass deletes none. River never deletes running jobs.
func clearQueue(ctx context.Context, app *backpack.App, queue string, out bonfire.Output) error {
	m, err := From(app)
	if err != nil {
		return err
	}
	client, err := m.insertClient()
	if err != nil {
		return err
	}
	out.Info(fmt.Sprintf("Clearing queue %q", queue))
	total := 0
	for {
		res, err := client.JobDeleteMany(ctx, river.NewJobDeleteManyParams().
			Queues(queue).
			States(rivertype.JobStateAvailable, rivertype.JobStateScheduled, rivertype.JobStateRetryable).
			First(clearBatch))
		if err != nil {
			return fmt.Errorf("conga: clear queue %q: %w", queue, err)
		}
		if len(res.Jobs) == 0 {
			break
		}
		total += len(res.Jobs)
	}
	out.Success(fmt.Sprintf("Cleared %d jobs from %s", total, queue))
	return nil
}

// confirmProduction asks question before a destructive step when the
// environment is production and out is an interactive terminal, unless
// --force is set (D-28). Off a terminal, under -n or -q and in
// bonfire.Call it never asks. A "no" returns a Reported error (exit 1).
func confirmProduction(app *backpack.App, in bonfire.Input, out bonfire.Output, question string) error {
	if in != nil {
		if v, ok := in.Flag("force"); ok {
			if force, _ := strconv.ParseBool(v); force {
				return nil
			}
		}
	}
	if app == nil || app.Config == nil || app.Config.Environment() != "production" {
		return nil
	}
	ok, err := bonfire.ConfirmProceed(out, question)
	if err != nil {
		return err
	}
	if !ok {
		out.Warning("Cancelled.")
		return bonfire.Reported(errors.New("queue:clear cancelled"))
	}
	return nil
}

// withDB opens and publishes the shared pool for a CLI command and closes it
// when fn returns, the way lagoon's own commands do.
func withDB(ctx context.Context, app *backpack.App, fn func() error) error {
	sqlDB, gdb, err := lagoon.OpenFromApp(ctx, app)
	if err != nil {
		return err
	}
	defer sqlDB.Close()
	if err := lagoon.Publish(app, sqlDB, gdb); err != nil {
		return err
	}
	return fn()
}