diff --git a/.gitignore b/.gitignore index ea7d938..fafe4e0 100644 --- a/.gitignore +++ b/.gitignore @@ -37,4 +37,7 @@ go.work.sum # Stale branches / forks stale +# Taskharbor binaries +th + # .gitignore diff --git a/cmd/taskharbor/internal/app/app.go b/cmd/taskharbor/internal/app/app.go new file mode 100644 index 0000000..c706a2f --- /dev/null +++ b/cmd/taskharbor/internal/app/app.go @@ -0,0 +1,336 @@ +package app + +import ( + "flag" + "fmt" + "io" + "os" + "strings" + + th "github.com/ARJ2211/taskharbor/taskharbor" + "github.com/ARJ2211/taskharbor/taskharbor/driver" +) + +type GlobalFlags struct { + Driver string + Queue string + JSON bool + Verbose bool + + PostgresDSN string + RedisAddr string +} + +func envOr(key, fallback string) string { + if v := strings.TrimSpace(os.Getenv(key)); v != "" { + return v + } + return fallback +} + +func Run(argv []string, stdout, stderr io.Writer) int { + var g GlobalFlags + var help bool + var h bool + + fs := flag.NewFlagSet("taskharbor", flag.ContinueOnError) + fs.SetOutput(io.Discard) + + driverList := "" + for _, d := range driver.ImplementedDrivers { + driverList += d + "|" + } + driverList = driverList[:len(driverList)-1] + + fs.StringVar(&g.Driver, "driver", "memory", fmt.Sprintf("drivers: %s", driverList)) + fs.StringVar(&g.Queue, "queue", th.DefaultQueue, "queue name") + fs.BoolVar(&g.JSON, "json", false, "output JSON") + fs.BoolVar(&g.Verbose, "verbose", false, "verbose logs") + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + fs.StringVar(&g.PostgresDSN, "dsn", envOr("TH_PG_DSN", envOr("TH_POSTGRES_DSN", "")), "postgres DSN (for --driver postgres)") + fs.StringVar(&g.RedisAddr, "redis-addr", envOr("TH_REDIS_ADDR", ""), "redis addr host:port (for --driver redis)") + + if err := fs.Parse(argv); err != nil { + fmt.Println(stderr, "error: ", err) + printRootUsage(stderr) + return 2 + } + + args := fs.Args() + if help || h || len(argv) == 0 { + printRootUsage(stdout) + return 0 + } + + cmd := args[0] + cmdArgs := args[1:] + + switch cmd { + case "worker": + return runWorker(g, cmdArgs, stdout, stderr) + case "enqueue": + return runEnqueue(g, cmdArgs, stdout, stderr) + case "list": + return runList(g, cmdArgs, stdout, stderr) + case "inspect": + return runInspect(g, cmdArgs, stdout, stderr) + case "dlq": + return runDLQ(g, cmdArgs, stdout, stderr) + case "job": + return runJob(g, cmdArgs, stdout, stderr) + case "help": + printRootUsage(stdout) + return 0 + default: + fmt.Fprintln(stderr, "error: unknown command:", cmd) + printRootUsage(stderr) + return 2 + } +} + +func runWorker(g GlobalFlags, argv []string, stdout, stderr io.Writer) int { + var help bool + var h bool + + fs := flag.NewFlagSet("taskharbor worker", flag.ContinueOnError) + fs.SetOutput(io.Discard) + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + if err := fs.Parse(argv); err != nil { + fmt.Fprintln(stderr, "error:", err) + printWorkerUsage(stderr) + return 2 + } + + args := fs.Args() + if help || h || len(args) == 0 { + printWorkerUsage(stdout) + return 0 + } + + sub := args[0] + subArgs := args[1:] + + switch sub { + case "run": + return runWorkerRun(g, subArgs, stdout, stderr) + default: + fmt.Fprintln(stderr, "error: unknown subcommand: worker", sub) + printWorkerUsage(stderr) + return 2 + } +} + +func runList(_ GlobalFlags, argv []string, stdout, stderr io.Writer) int { + var help bool + var h bool + + fs := flag.NewFlagSet("taskharbor list", flag.ContinueOnError) + fs.SetOutput(io.Discard) + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + if err := fs.Parse(argv); err != nil { + fmt.Fprintln(stderr, "error:", err) + printListUsage(stderr) + return 2 + } + + if help || h { + printListUsage(stdout) + return 0 + } + + fmt.Fprintln(stderr, "not implemented yet (issue #119)") + return 1 +} + +func runInspect(_ GlobalFlags, argv []string, stdout, stderr io.Writer) int { + var help bool + var h bool + + fs := flag.NewFlagSet("taskharbor inspect", flag.ContinueOnError) + fs.SetOutput(io.Discard) + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + if err := fs.Parse(argv); err != nil { + fmt.Fprintln(stderr, "error:", err) + printInspectUsage(stderr) + return 2 + } + + if help || h { + printInspectUsage(stdout) + return 0 + } + + args := fs.Args() + if len(args) != 1 { + fmt.Fprintln(stderr, "error: inspect requires exactly 1 arg: ") + printInspectUsage(stderr) + return 2 + } + + fmt.Fprintln(stderr, "not implemented yet (issue #119)") + return 1 +} + +func runDLQ(g GlobalFlags, argv []string, stdout, stderr io.Writer) int { + var help bool + var h bool + + fs := flag.NewFlagSet("taskharbor dlq", flag.ContinueOnError) + fs.SetOutput(io.Discard) + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + if err := fs.Parse(argv); err != nil { + fmt.Fprintln(stderr, "error:", err) + printDLQUsage(stderr) + return 2 + } + + args := fs.Args() + if help || h || len(args) == 0 { + printDLQUsage(stdout) + return 0 + } + + sub := args[0] + subArgs := args[1:] + + switch sub { + case "list": + return runDLQList(g, subArgs, stdout, stderr) + case "requeue": + return runDLQRequeue(g, subArgs, stdout, stderr) + default: + fmt.Fprintln(stderr, "error: unknown subcommand: dlq", sub) + printDLQUsage(stderr) + return 2 + } +} + +func runDLQList(_ GlobalFlags, argv []string, stdout, stderr io.Writer) int { + var help bool + var h bool + + fs := flag.NewFlagSet("taskharbor dlq list", flag.ContinueOnError) + fs.SetOutput(io.Discard) + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + if err := fs.Parse(argv); err != nil { + fmt.Fprintln(stderr, "error:", err) + printDLQListUsage(stderr) + return 2 + } + + if help || h { + printDLQListUsage(stdout) + return 0 + } + + fmt.Fprintln(stderr, "not implemented yet (issue #119)") + return 1 +} + +func runDLQRequeue(_ GlobalFlags, argv []string, stdout, stderr io.Writer) int { + var help bool + var h bool + + fs := flag.NewFlagSet("taskharbor dlq requeue", flag.ContinueOnError) + fs.SetOutput(io.Discard) + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + if err := fs.Parse(argv); err != nil { + fmt.Fprintln(stderr, "error:", err) + printDLQRequeueUsage(stderr) + return 2 + } + + if help || h { + printDLQRequeueUsage(stdout) + return 0 + } + + args := fs.Args() + if len(args) != 1 { + fmt.Fprintln(stderr, "error: dlq requeue requires exactly 1 arg: ") + printDLQRequeueUsage(stderr) + return 2 + } + + fmt.Fprintln(stderr, "not implemented yet (issue #119)") + return 1 +} + +func runJob(g GlobalFlags, argv []string, stdout, stderr io.Writer) int { + var help bool + var h bool + + fs := flag.NewFlagSet("taskharbor job", flag.ContinueOnError) + fs.SetOutput(io.Discard) + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + if err := fs.Parse(argv); err != nil { + fmt.Fprintln(stderr, "error:", err) + printJobUsage(stderr) + return 2 + } + + args := fs.Args() + if help || h || len(args) == 0 { + printJobUsage(stdout) + return 0 + } + + sub := args[0] + subArgs := args[1:] + + switch sub { + case "retry": + return runJobRetry(g, subArgs, stdout, stderr) + default: + fmt.Fprintln(stderr, "error: unknown subcommand: job", sub) + printJobUsage(stderr) + return 2 + } +} + +func runJobRetry(_ GlobalFlags, argv []string, stdout, stderr io.Writer) int { + var help bool + var h bool + + fs := flag.NewFlagSet("taskharbor job retry", flag.ContinueOnError) + fs.SetOutput(io.Discard) + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + if err := fs.Parse(argv); err != nil { + fmt.Fprintln(stderr, "error:", err) + printJobRetryUsage(stderr) + return 2 + } + + if help || h { + printJobRetryUsage(stdout) + return 0 + } + + args := fs.Args() + if len(args) != 1 { + fmt.Fprintln(stderr, "error: job retry requires exactly 1 arg: ") + printJobRetryUsage(stderr) + return 2 + } + + fmt.Fprintln(stderr, "not implemented yet (issue #119)") + return 1 +} diff --git a/cmd/taskharbor/internal/app/app_test.go b/cmd/taskharbor/internal/app/app_test.go new file mode 100644 index 0000000..2510d71 --- /dev/null +++ b/cmd/taskharbor/internal/app/app_test.go @@ -0,0 +1,48 @@ +package app + +import ( + "bytes" + "strings" + "testing" +) + +func TestRootHelp(t *testing.T) { + var out, err bytes.Buffer + code := Run([]string{"--help"}, &out, &err) + if code != 0 { + t.Fatalf("expected 0, got %d (stderr=%q)", code, err.String()) + } + if !strings.Contains(out.String(), "worker run") { + t.Fatalf("expected help to mention worker run, got: %q", out.String()) + } +} + +func TestWorkerRunHelp(t *testing.T) { + var out, err bytes.Buffer + code := Run([]string{"worker", "run", "--help"}, &out, &err) + if code != 0 { + t.Fatalf("expected 0, got %d (stderr=%q)", code, err.String()) + } + if !strings.Contains(out.String(), "worker run") { + t.Fatalf("expected worker run usage, got: %q", out.String()) + } +} + +func TestUnknownCommand(t *testing.T) { + var out, err bytes.Buffer + code := Run([]string{"nope"}, &out, &err) + if code == 0 { + t.Fatalf("expected non-zero, got %d", code) + } +} + +func TestEnqueueMinimal(t *testing.T) { + var out, err bytes.Buffer + code := Run([]string{"enqueue", "--type", "echo", "--payload", "hi"}, &out, &err) + if code != 0 { + t.Fatalf("expected 0, got %d (stderr=%q)", code, err.String()) + } + if strings.TrimSpace(out.String()) == "" { + t.Fatalf("expected a job id, got empty output") + } +} diff --git a/cmd/taskharbor/internal/app/enqueue.go b/cmd/taskharbor/internal/app/enqueue.go new file mode 100644 index 0000000..68d16e9 --- /dev/null +++ b/cmd/taskharbor/internal/app/enqueue.go @@ -0,0 +1,204 @@ +package app + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "io" + "strconv" + "strings" + "time" + + "github.com/ARJ2211/taskharbor/cmd/taskharbor/internal/backend" + th "github.com/ARJ2211/taskharbor/taskharbor" +) + +func runEnqueue(g GlobalFlags, argv []string, stdout, stderr io.Writer) int { + var ( + help bool + h bool + + jobType string + queue string + runAt string + + timeout time.Duration + maxAttempts int + idKey string + + payloadStr string + payloadJSON string + ) + + fs := flag.NewFlagSet("taskharbor enqueue", flag.ContinueOnError) + fs.SetOutput(io.Discard) + + fs.StringVar(&jobType, "type", "", "job type (required)") + fs.StringVar(&queue, "queue", "", "queue name (default: global --queue or DefaultQueue)") + fs.StringVar(&runAt, "run-at", "", "schedule time (RFC3339/RFC3339Nano) or unix seconds (or unix ms)") + fs.DurationVar(&timeout, "timeout", 0, "job timeout (e.g. 30s)") + fs.IntVar(&maxAttempts, "max-attempts", 0, "max attempts before DLQ (0 means fail immediately on handler error)") + fs.StringVar(&idKey, "idempotency-key", "", "idempotency key") + + fs.StringVar(&payloadStr, "payload", "", "payload as string (will be JSON-encoded)") + fs.StringVar(&payloadJSON, "payload-json", "", "payload as raw JSON (not double-encoded)") + + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + if err := fs.Parse(argv); err != nil { + fmt.Fprintln(stderr, "error:", err) + printEnqueueUsage(stderr) + return 2 + } + if help || h { + printEnqueueUsage(stdout) + return 0 + } + if len(fs.Args()) != 0 { + fmt.Fprintln(stderr, "error: unexpected args:", strings.Join(fs.Args(), " ")) + printEnqueueUsage(stderr) + return 2 + } + + if strings.TrimSpace(jobType) == "" { + fmt.Fprintln(stderr, "error: --type is required") + printEnqueueUsage(stderr) + return 2 + } + if payloadStr != "" && payloadJSON != "" { + fmt.Fprintln(stderr, "error: use only one of --payload or --payload-json") + return 2 + } + + effectiveQueue := strings.TrimSpace(queue) + if effectiveQueue == "" { + effectiveQueue = strings.TrimSpace(g.Queue) + } + if effectiveQueue == "" { + effectiveQueue = th.DefaultQueue + } + + runAtTime, err := parseRunAt(runAt) + if err != nil { + fmt.Fprintln(stderr, "error:", err) + return 2 + } + + var payload any = nil + if payloadJSON != "" { + b := []byte(payloadJSON) + if !json.Valid(b) { + fmt.Fprintln(stderr, "error: --payload-json must be valid JSON") + return 2 + } + payload = json.RawMessage(b) + } else if payloadStr != "" { + payload = payloadStr + } + + ctx := context.Background() + hnd, err := backend.Open(ctx, backend.Config{ + Driver: g.Driver, + PostgresDSN: g.PostgresDSN, + RedisAddr: g.RedisAddr, + }) + if err != nil { + fmt.Fprintln(stderr, "error:", err) + return 1 + } + defer func() { _ = hnd.Close() }() + + client := th.NewClient(hnd.Driver) + + req := th.JobRequest{ + Type: jobType, + Payload: payload, + Queue: effectiveQueue, + RunAt: runAtTime, + Timeout: timeout, + IdempotencyKey: idKey, + MaxAttempts: maxAttempts, + } + + id, err := client.Enqueue(ctx, req) + if err != nil { + fmt.Fprintln(stderr, "error:", err) + return 1 + } + + if g.JSON { + out := map[string]any{ + "id": id, + "type": jobType, + "queue": effectiveQueue, + "max_attempts": maxAttempts, + "idempotency": idKey, + "timeout_millis": func() any { + if timeout <= 0 { + return nil + } + return timeout.Milliseconds() + }(), + "run_at": func() any { + if runAtTime.IsZero() { + return nil + } + return runAtTime.Format(time.RFC3339Nano) + }(), + } + enc := json.NewEncoder(stdout) + enc.SetEscapeHTML(false) + _ = enc.Encode(out) + return 0 + } + + fmt.Fprintln(stdout, id) + return 0 +} + +func parseRunAt(s string) (time.Time, error) { + s = strings.TrimSpace(s) + if s == "" { + return time.Time{}, nil + } + + // unix seconds or millis + if isDigits(s) { + n, err := strconv.ParseInt(s, 10, 64) + if err != nil { + return time.Time{}, fmt.Errorf("invalid --run-at: %w", err) + } + if n < 0 { + return time.Time{}, fmt.Errorf("invalid --run-at: must be >= 0") + } + // heuristic: 13+ digits -> ms + if n >= 1_000_000_000_000 { + return time.Unix(0, n*int64(time.Millisecond)).UTC(), nil + } + return time.Unix(n, 0).UTC(), nil + } + + // RFC3339 / RFC3339Nano + if t, err := time.Parse(time.RFC3339Nano, s); err == nil { + return t.UTC(), nil + } + if t, err := time.Parse(time.RFC3339, s); err == nil { + return t.UTC(), nil + } + + return time.Time{}, fmt.Errorf("invalid --run-at: use RFC3339 (with timezone) or unix seconds") +} + +func isDigits(s string) bool { + if s == "" { + return false + } + for i := 0; i < len(s); i++ { + if s[i] < '0' || s[i] > '9' { + return false + } + } + return true +} diff --git a/cmd/taskharbor/internal/app/flags.go b/cmd/taskharbor/internal/app/flags.go new file mode 100644 index 0000000..43c9280 --- /dev/null +++ b/cmd/taskharbor/internal/app/flags.go @@ -0,0 +1,11 @@ +package app + +import "strings" + +type multiString []string + +func (m *multiString) String() string { return strings.Join(*m, ",") } +func (m *multiString) Set(v string) error { + *m = append(*m, v) + return nil +} diff --git a/cmd/taskharbor/internal/app/usage.go b/cmd/taskharbor/internal/app/usage.go new file mode 100644 index 0000000..c96df89 --- /dev/null +++ b/cmd/taskharbor/internal/app/usage.go @@ -0,0 +1,149 @@ +package app + +import ( + "fmt" + "io" +) + +func printRootUsage(w io.Writer) { + fmt.Fprintln(w, `TaskHarbor CLI + +Usage: + taskharbor [global flags] [args] + +Global flags: + --driver memory|postgres|redis (default: memory) + --queue queue name (default: default) + --json JSON output + --verbose verbose logs + --help, -h show help + --dsn postgres DSN (for --driver postgres) + --redis-addr redis addr host:port (for --driver redis) + +Commands: + worker run + enqueue + list + inspect + dlq list + dlq requeue + job retry + +Examples: + taskharbor --help + taskharbor worker run --help + taskharbor enqueue --help`) +} + +func printWorkerUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + taskharbor [global flags] worker + +Subcommands: + run + +Example: + taskharbor worker run --help`) +} + +func printWorkerRunUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + taskharbor [global flags] worker run [flags] + +Flags: + --concurrency max concurrent jobs + --poll-interval e.g. 200ms + --lease-duration e.g. 30s + --heartbeat-interval e.g. 10s + --register repeatable mapping: jobType=builtin (builtins: echo, fail, sleep) + +Examples: + taskharbor worker run + taskharbor worker run --concurrency 8 + taskharbor worker run --register email=echo --register slow=sleep`) +} + +func printEnqueueUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + taskharbor [global flags] enqueue [flags] + +Flags: + --type required job type + --queue queue name (default: global --queue or "default") + --run-at RFC3339/RFC3339Nano or unix seconds (or unix ms) + --timeout e.g. 30s + --max-attempts retries before DLQ (0 means fail immediately on handler error) + --idempotency-key idempotency key + --payload string payload (JSON-encoded) + --payload-json raw JSON payload (not double-encoded) + +Examples: + taskharbor enqueue --type echo --payload hello + taskharbor enqueue --type echo --payload-json {"msg":"hi"} + taskharbor enqueue --type echo --run-at 1772000000 --payload hi + taskharbor --driver postgres --dsn $TH_PG_DSN enqueue --type echo --idempotency-key user:123 --payload hi`) +} + +func printListUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + taskharbor [global flags] list [flags] + +Notes: + This will be implemented in issue #119.`) +} + +func printInspectUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + taskharbor [global flags] inspect [flags] + +Notes: + This will be implemented in issue #119.`) +} + +func printDLQUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + taskharbor [global flags] dlq + +Subcommands: + list + requeue + +Examples: + taskharbor dlq list --help + taskharbor dlq requeue `) +} + +func printDLQListUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + taskharbor [global flags] dlq list [flags] + +Notes: + This will be implemented in issue #119.`) +} + +func printDLQRequeueUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + taskharbor [global flags] dlq requeue [flags] + +Notes: + This will be implemented in issue #119.`) +} + +func printJobUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + taskharbor [global flags] job + +Subcommands: + retry + +Example: + taskharbor job retry `) +} + +func printJobRetryUsage(w io.Writer) { + fmt.Fprintln(w, `Usage: + taskharbor [global flags] job retry [flags] + +Notes: + This will be implemented in issue #119.`) +} diff --git a/cmd/taskharbor/internal/app/worker_run.go b/cmd/taskharbor/internal/app/worker_run.go new file mode 100644 index 0000000..159dd3f --- /dev/null +++ b/cmd/taskharbor/internal/app/worker_run.go @@ -0,0 +1,216 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "os" + "os/signal" + "strconv" + "strings" + "syscall" + "time" + + "github.com/ARJ2211/taskharbor/cmd/taskharbor/internal/backend" + th "github.com/ARJ2211/taskharbor/taskharbor" +) + +func runWorkerRun(g GlobalFlags, argv []string, stdout, stderr io.Writer) int { + var ( + help bool + h bool + + concurrency int + poll time.Duration + lease time.Duration + heartbeat time.Duration + + reg multiString + ) + fs := flag.NewFlagSet("taskharbor worker run", flag.ContinueOnError) + fs.SetOutput(io.Discard) + + fs.IntVar(&concurrency, "concurrency", 0, "max concurrent jobs (default: taskharbor default)") + fs.DurationVar(&poll, "poll-interval", 0, "poll interval when no jobs (e.g. 200ms)") + fs.DurationVar(&lease, "lease-duration", 0, "lease duration (e.g. 30s)") + fs.DurationVar(&heartbeat, "heartbeat-interval", 0, "lease heartbeat interval (e.g. 10s)") + fs.Var(®, "register", "map jobType to builtin handler (repeatable). format: jobType=builtin. builtins: echo,fail,sleep") + + fs.BoolVar(&help, "help", false, "show help") + fs.BoolVar(&h, "h", false, "show help") + + if err := fs.Parse(argv); err != nil { + fmt.Fprintln(stderr, "error:", err) + printWorkerRunUsage(stderr) + return 2 + } + if help || h { + printWorkerRunUsage(stdout) + return 0 + } + if len(fs.Args()) != 0 { + fmt.Fprintln(stderr, "error: unexpected args:", strings.Join(fs.Args(), " ")) + printWorkerRunUsage(stderr) + return 2 + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Ctrl+C / SIGTERM -> graceful shutdown + sigc := make(chan os.Signal, 2) + signal.Notify(sigc, os.Interrupt, syscall.SIGTERM) + defer signal.Stop(sigc) + go func() { + <-sigc + cancel() + }() + + bh, err := backend.Open(ctx, backend.Config{ + Driver: g.Driver, + PostgresDSN: g.PostgresDSN, + RedisAddr: g.RedisAddr, + }) + if err != nil { + fmt.Fprintln(stderr, "error:", err) + return 1 + } + defer func() { _ = bh.Close() }() + + opts := make([]th.Option, 0, 6) + opts = append(opts, th.WithDefaultQueue(g.Queue)) + if concurrency > 0 { + opts = append(opts, th.WithConcurrency(concurrency)) + } + if poll > 0 { + opts = append(opts, th.WithPollInterval(poll)) + } + if lease > 0 { + opts = append(opts, th.WithLeaseDuration(lease)) + } + if heartbeat > 0 { + opts = append(opts, th.WithHeartbeatInterval(heartbeat)) + } + + worker := th.NewWorker(bh.Driver, opts...) + + // Example Builtins + builtins := map[string]th.Handler{ + "echo": echoHandler(stdout), + "fail": failHandler(), + "sleep": sleepHandler(stdout), + } + + // Always register the default names. + for name, fn := range builtins { + _ = worker.Register(name, fn) + } + + // Apply --register mappings (jobType=builtin) + for _, m := range reg { + jobType, builtin, ok := strings.Cut(m, "=") + jobType = strings.TrimSpace(jobType) + if !ok { + // allow shorthand + builtin = jobType + } + builtin = strings.TrimSpace(builtin) + + if jobType == "" || builtin == "" { + fmt.Fprintln(stderr, "error: invalid --register:", m) + return 2 + } + fn, ok := builtins[builtin] + if !ok { + fmt.Fprintln(stderr, "error: unknown builtin for --register:", builtin) + return 2 + } + if err := worker.Register(jobType, fn); err != nil { + fmt.Fprintln(stderr, "error:", err) + return 1 + } + } + + fmt.Fprintf(stdout, "worker started id=%s driver=%s queue=%s\n", worker.ID(), strings.ToLower(g.Driver), g.Queue) + + if err := worker.Run(ctx); err != nil { + fmt.Fprintln(stderr, "error:", err) + return 1 + } + + fmt.Fprintln(stdout, "worker stopped") + return 0 +} + +func echoHandler(w io.Writer) th.Handler { + return func(ctx context.Context, job th.Job) error { + _ = ctx + fmt.Fprintf(w, "echo id=%s type=%s queue=%s payload=%s\n", job.ID, job.Type, job.Queue, string(job.Payload)) + return nil + } +} + +func failHandler() th.Handler { + return func(ctx context.Context, job th.Job) error { + _ = ctx + _ = job + return errors.New("fail handler: requested failure") + } +} + +func sleepHandler(w io.Writer) th.Handler { + return func(ctx context.Context, job th.Job) error { + d := parseSleepDuration(job.Payload) + if d <= 0 { + d = 250 * time.Millisecond + } + fmt.Fprintf(w, "sleep id=%s duration=%s\n", job.ID, d) + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + return nil + } + } +} + +func parseSleepDuration(payload []byte) time.Duration { + s := strings.TrimSpace(string(payload)) + if s == "" { + return 0 + } + + // Try plain int millis: "1500" or `"1500"` + s = strings.Trim(s, "\"") + if n, err := strconv.Atoi(s); err == nil { + return time.Duration(n) * time.Millisecond + } + + // Try JSON number or JSON object: {"ms": 1500} or {"duration":"1.5s"} + var num int + if err := json.Unmarshal(payload, &num); err == nil { + return time.Duration(num) * time.Millisecond + } + + var obj struct { + MS int `json:"ms"` + Duration string `json:"duration"` + } + if err := json.Unmarshal(payload, &obj); err == nil { + if obj.Duration != "" { + if d, err := time.ParseDuration(obj.Duration); err == nil { + return d + } + } + if obj.MS > 0 { + return time.Duration(obj.MS) * time.Millisecond + } + } + + return 0 +} diff --git a/cmd/taskharbor/internal/backend/backend.go b/cmd/taskharbor/internal/backend/backend.go new file mode 100644 index 0000000..bcbbd9b --- /dev/null +++ b/cmd/taskharbor/internal/backend/backend.go @@ -0,0 +1,71 @@ +package backend + +import ( + "context" + "fmt" + "strings" + + "github.com/ARJ2211/taskharbor/taskharbor/driver" + "github.com/ARJ2211/taskharbor/taskharbor/driver/memory" + "github.com/ARJ2211/taskharbor/taskharbor/driver/postgres" + "github.com/ARJ2211/taskharbor/taskharbor/driver/redis" +) + +type Config struct { + Driver string + + PostgresDSN string + RedisAddr string +} + +type Handle struct { + Driver driver.Driver +} + +func (h *Handle) Close() error { + if h == nil || h.Driver == nil { + return nil + } + if err := h.Driver.Close(); err != nil { + return err + } + return nil +} + +/* +This function opens a new driver based on the configs and +arguments provided by the user. Default driver: memory. +*/ +func Open(ctx context.Context, cfg Config) (*Handle, error) { + ds := strings.ToLower(strings.TrimSpace(cfg.Driver)) + switch ds { + case "", "memory": + memDrvHnd := Handle{Driver: memory.New()} + return &memDrvHnd, nil + case "postgres": + if strings.TrimSpace(cfg.PostgresDSN) == "" { + return nil, fmt.Errorf("postgres requires --dsn (or TH_PG_DSN)") + } + d, err := postgres.New(ctx, cfg.PostgresDSN) + if err != nil { + return nil, err + } + psqlDrvHnd := Handle{Driver: d} + return &psqlDrvHnd, nil + case "redis": + if strings.TrimSpace(cfg.RedisAddr) == "" { + return nil, fmt.Errorf("redis requires --redis-addr (or TH_REDIS_ADDR)") + } + d, err := redis.New(ctx, cfg.RedisAddr) + if err != nil { + return nil, err + } + redisDrvHnd := Handle{Driver: d} + return &redisDrvHnd, nil + default: + return nil, fmt.Errorf( + "unknown driver: %s (expected memory|postgres|redis)", + cfg.Driver, + ) + } +} diff --git a/cmd/taskharbor/internal/backend/backend_test.go b/cmd/taskharbor/internal/backend/backend_test.go new file mode 100644 index 0000000..cf850f0 --- /dev/null +++ b/cmd/taskharbor/internal/backend/backend_test.go @@ -0,0 +1,65 @@ +package backend + +import ( + "context" + "os" + "testing" + + "github.com/ARJ2211/taskharbor/cmd/taskharbor/internal/envutil" +) + +func TestOpenMemory(t *testing.T) { + h, err := Open(context.Background(), Config{Driver: "memory"}) + if err != nil { + t.Fatalf("expected nil err, got %v", err) + } + if h == nil || h.Driver == nil { + t.Fatalf("expected non-nil handle + driver") + } + if err := h.Close(); err != nil { + t.Fatalf("expected nil close err, got %v", err) + } +} + +func TestOpenRedis(t *testing.T) { + cwd, _ := os.Getwd() + _ = envutil.LoadRepoDotenv(cwd) + + h, err := Open(context.Background(), Config{ + Driver: "redis", RedisAddr: os.Getenv("REDIS_ADDR"), + }) + if err != nil { + t.Fatalf("expected nil err, got %v", err) + } + if h == nil || h.Driver == nil { + t.Fatalf("expected non-nil handle + driver") + } + if err := h.Close(); err != nil { + t.Fatalf("expected nil close err, got %v", err) + } +} + +func TestOpenPostgres(t *testing.T) { + cwd, _ := os.Getwd() + _ = envutil.LoadRepoDotenv(cwd) + + h, err := Open(context.Background(), Config{ + Driver: "postgres", PostgresDSN: os.Getenv("TASKHARBOR_TEST_DSN"), + }) + if err != nil { + t.Fatalf("expected nil err, got %v", err) + } + if h == nil || h.Driver == nil { + t.Fatalf("expected non-nil handle + driver") + } + if err := h.Close(); err != nil { + t.Fatalf("expected nil close err, got %v", err) + } +} + +func TestOpenUnknownDriver(t *testing.T) { + _, err := Open(context.Background(), Config{Driver: "nope"}) + if err == nil { + t.Fatalf("expected error") + } +} diff --git a/cmd/taskharbor/internal/envutil/dotenv.go b/cmd/taskharbor/internal/envutil/dotenv.go new file mode 100644 index 0000000..25fa7e7 --- /dev/null +++ b/cmd/taskharbor/internal/envutil/dotenv.go @@ -0,0 +1,36 @@ +package envutil + +import ( + "os" + "path/filepath" + + "github.com/joho/godotenv" +) + +/* +LoadRepoDotenv walks up from startDir until it finds go.mod, then loads .env from that directory. + +If not found, it does nothing and returns nil. +*/ +func LoadRepoDotenv(startDir string) error { + dir, err := filepath.Abs(startDir) + if err != nil { + return err + } + + for { + if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { + envPath := filepath.Join(dir, ".env") + if _, err := os.Stat(envPath); err == nil { + return godotenv.Load(envPath) + } + return nil + } + + parent := filepath.Dir(dir) + if parent == dir { + return nil + } + dir = parent + } +} diff --git a/cmd/taskharbor/main.go b/cmd/taskharbor/main.go new file mode 100644 index 0000000..cdecef8 --- /dev/null +++ b/cmd/taskharbor/main.go @@ -0,0 +1,11 @@ +package main + +import ( + "os" + + "github.com/ARJ2211/taskharbor/cmd/taskharbor/internal/app" +) + +func main() { + os.Exit(app.Run(os.Args[1:], os.Stdout, os.Stderr)) +} diff --git a/taskharbor/driver/admin.go b/taskharbor/driver/admin.go new file mode 100644 index 0000000..a0d851f --- /dev/null +++ b/taskharbor/driver/admin.go @@ -0,0 +1,127 @@ +package driver + +import ( + "context" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "strings" + "time" +) + +type JobState string + +const ( + StateReady JobState = "ready" // runnable now (due) + StateScheduled JobState = "scheduled" // not due yet (run_at > now) + StateInflight JobState = "inflight" // leased + StateDone JobState = "done" // terminal success + StateDLQ JobState = "dlq" // terminal failure +) + +var ( + ErrAdminUnsupported = errors.New("admin interface not supported by driver") + ErrInvalidCursor = errors.New("invalid cursor") + ErrJobNotDLQ = errors.New("job is not in dlq") +) + +type LeaseInfo struct { + Token LeaseToken + ExpiresAt time.Time +} + +type DLQInfo struct { + Reason string + FailedAt time.Time +} + +type JobInfo struct { + Record JobRecord + State JobState + + Lease *LeaseInfo + DLQ *DLQInfo +} + +type JobSummary struct { + ID string + Type string + Queue string + State JobState + RunAt time.Time + CreatedAt time.Time + Timeout time.Duration + Attempts int + MaxAttempts int + LastError string + FailedAt time.Time + + LeaseExpiresAt time.Time + DLQReason string + DLQFailedAt time.Time +} + +type ListRequest struct { + Queue string // required + State JobState // optional; empty means "all" + Now time.Time + Limit int + Cursor string +} + +type ListPage struct { + Jobs []JobSummary + NextCursor string +} + +type RequeueOptions struct { + Queue string // optional guard; if set and mismatch, driver should error + RunAt time.Time // zero => immediate + ResetAttempts bool // if true: set attempts=0 and clear last_error/failed_at +} + +// Admin is optional and used by CLI/dev tooling. Worker + Client must not depend on this. +type Admin interface { + Inspect(ctx context.Context, id string, now time.Time) (JobInfo, error) + List(ctx context.Context, req ListRequest) (ListPage, error) + RequeueDLQ(ctx context.Context, id string, now time.Time, opt RequeueOptions) error +} + +// Cursor is an opaque pagination token with a standard encoding. +// A/B/ID meaning depends on State and the driver’s chosen ordering. +type Cursor struct { + V int `json:"v"` + State JobState `json:"s"` + A int64 `json:"a"` + B int64 `json:"b,omitempty"` + ID string `json:"id"` +} + +func EncodeCursor(c Cursor) string { + c.V = 1 + b, _ := json.Marshal(c) + return base64.RawURLEncoding.EncodeToString(b) +} + +func DecodeCursor(s string) (Cursor, error) { + s = strings.TrimSpace(s) + if s == "" { + return Cursor{}, nil + } + b, err := base64.RawURLEncoding.DecodeString(s) + if err != nil { + return Cursor{}, fmt.Errorf("%w: %v", ErrInvalidCursor, err) + } + var c Cursor + if err := json.Unmarshal(b, &c); err != nil { + return Cursor{}, fmt.Errorf("%w: %v", ErrInvalidCursor, err) + } + if c.V != 1 { + return Cursor{}, fmt.Errorf("%w: unsupported cursor version", ErrInvalidCursor) + } + if strings.TrimSpace(c.ID) == "" { + return Cursor{}, fmt.Errorf("%w: missing id", ErrInvalidCursor) + } + return c, nil +} diff --git a/taskharbor/driver/driver.go b/taskharbor/driver/driver.go index 7c29126..f2096b0 100644 --- a/taskharbor/driver/driver.go +++ b/taskharbor/driver/driver.go @@ -7,6 +7,9 @@ import ( "time" ) +// List of all implemented drivers +var ImplementedDrivers = []string{"memory", "redis", "postgres"} + /* This JobRecord is the driver-level representation of a Job. This is the struct that will be saved in some backend (Redis/SQL/etc...) diff --git a/taskharbor/driver/memory/admin.go b/taskharbor/driver/memory/admin.go new file mode 100644 index 0000000..7b349ac --- /dev/null +++ b/taskharbor/driver/memory/admin.go @@ -0,0 +1,328 @@ +package memory + +import ( + "container/heap" + "context" + "fmt" + "sort" + "strings" + "time" + + "github.com/ARJ2211/taskharbor/taskharbor/driver" +) + +var _ driver.Admin = (*Driver)(nil) + +func (d *Driver) Inspect(ctx context.Context, id string, now time.Time) (driver.JobInfo, error) { + if err := ctx.Err(); err != nil { + return driver.JobInfo{}, err + } + id = strings.TrimSpace(id) + if id == "" { + return driver.JobInfo{}, driver.ErrJobNotFound + } + if now.IsZero() { + now = time.Now().UTC() + } + + d.mu.Lock() + defer d.mu.Unlock() + + if d.closed { + return driver.JobInfo{}, ErrDriverClosed + } + + // inflight (fast path) + if q, ok := d.inflightIndex[id]; ok { + qs := d.queues[q] + if qs != nil { + if it, ok := qs.inflight[id]; ok { + return driver.JobInfo{ + Record: it.rec, + State: driver.StateInflight, + Lease: &driver.LeaseInfo{ + Token: it.lease.Token, + ExpiresAt: it.lease.ExpiresAt, + }, + }, nil + } + } + } + + // dlq (fast path) + if q, ok := d.dlqIndex[id]; ok { + qs := d.queues[q] + if qs != nil { + if item, ok := qs.dlq[id]; ok { + return driver.JobInfo{ + Record: item.Record, + State: driver.StateDLQ, + DLQ: &driver.DLQInfo{ + Reason: item.Reason, + FailedAt: item.FailedAt, + }, + }, nil + } + } + } + + // done (fast path) + if _, ok := d.doneIndex[id]; ok { + if rec, ok2 := d.doneRecords[id]; ok2 { + return driver.JobInfo{Record: rec, State: driver.StateDone}, nil + } + // Shouldn't happen after we store doneRecords, but keep a sane error. + return driver.JobInfo{}, driver.ErrJobNotFound + } + + // ready/scheduled scan (no index) + for _, qs := range d.queues { + if qs == nil { + continue + } + for _, rec := range qs.runnable { + if rec.ID == id { + return driver.JobInfo{Record: rec, State: stateFromRunAt(now, rec.RunAt)}, nil + } + } + for _, rec := range qs.scheduled { + if rec.ID == id { + return driver.JobInfo{Record: rec, State: stateFromRunAt(now, rec.RunAt)}, nil + } + } + } + + return driver.JobInfo{}, driver.ErrJobNotFound +} + +func (d *Driver) List(ctx context.Context, req driver.ListRequest) (driver.ListPage, error) { + if err := ctx.Err(); err != nil { + return driver.ListPage{}, err + } + req.Queue = strings.TrimSpace(req.Queue) + if req.Queue == "" { + return driver.ListPage{}, fmt.Errorf("queue is required") + } + if req.Now.IsZero() { + req.Now = time.Now().UTC() + } + if req.Limit <= 0 { + req.Limit = 50 + } + + var cur driver.Cursor + if strings.TrimSpace(req.Cursor) != "" { + c, err := driver.DecodeCursor(req.Cursor) + if err != nil { + return driver.ListPage{}, err + } + cur = c + } + + d.mu.Lock() + defer d.mu.Unlock() + + if d.closed { + return driver.ListPage{}, ErrDriverClosed + } + + qs := d.queues[req.Queue] + if qs == nil { + return driver.ListPage{Jobs: nil, NextCursor: ""}, nil + } + + jobs := make([]driver.JobSummary, 0) + + addIf := func(s driver.JobSummary) { + if req.State == "" || s.State == req.State { + jobs = append(jobs, s) + } + } + + // runnable + for _, rec := range qs.runnable { + addIf(summaryFromRecord(req.Now, rec, nil, nil)) + } + // scheduled heap contents + for _, rec := range qs.scheduled { + addIf(summaryFromRecord(req.Now, rec, nil, nil)) + } + // inflight + for _, it := range qs.inflight { + lease := &driver.LeaseInfo{Token: it.lease.Token, ExpiresAt: it.lease.ExpiresAt} + addIf(summaryFromRecord(req.Now, it.rec, lease, nil)) + } + // dlq + for _, item := range qs.dlq { + dlq := &driver.DLQInfo{Reason: item.Reason, FailedAt: item.FailedAt} + addIf(summaryFromRecord(req.Now, item.Record, nil, dlq)) + } + // done (global map, filter by queue) + for _, rec := range d.doneRecords { + if rec.Queue != req.Queue { + continue + } + addIf(summaryFromRecord(req.Now, rec, nil, nil)) + } + + // stable ordering: (created_at asc, id asc) + sort.Slice(jobs, func(i, j int) bool { + ai := jobs[i].CreatedAt.UnixNano() + aj := jobs[j].CreatedAt.UnixNano() + if ai != aj { + return ai < aj + } + return jobs[i].ID < jobs[j].ID + }) + + // apply cursor filter + if strings.TrimSpace(req.Cursor) != "" { + out := jobs[:0] + for _, s := range jobs { + a := s.CreatedAt.UnixNano() + if a > cur.A || (a == cur.A && s.ID > cur.ID) { + out = append(out, s) + } + } + jobs = out + } + + if len(jobs) == 0 { + return driver.ListPage{Jobs: nil, NextCursor: ""}, nil + } + + if len(jobs) <= req.Limit { + return driver.ListPage{Jobs: jobs, NextCursor: ""}, nil + } + + page := jobs[:req.Limit] + last := page[len(page)-1] + next := driver.EncodeCursor(driver.Cursor{ + State: req.State, + A: last.CreatedAt.UnixNano(), + ID: last.ID, + }) + + return driver.ListPage{Jobs: page, NextCursor: next}, nil +} + +func (d *Driver) RequeueDLQ(ctx context.Context, id string, now time.Time, opt driver.RequeueOptions) error { + if err := ctx.Err(); err != nil { + return err + } + id = strings.TrimSpace(id) + if id == "" { + return driver.ErrJobNotFound + } + if now.IsZero() { + now = time.Now().UTC() + } + + d.mu.Lock() + defer d.mu.Unlock() + + if d.closed { + return ErrDriverClosed + } + + q, ok := d.dlqIndex[id] + if !ok { + // If job exists elsewhere, be explicit. + if _, ok := d.inflightIndex[id]; ok { + return driver.ErrJobNotDLQ + } + if _, ok := d.doneIndex[id]; ok { + return driver.ErrJobNotDLQ + } + for _, qs := range d.queues { + if qs == nil { + continue + } + for _, rec := range qs.runnable { + if rec.ID == id { + return driver.ErrJobNotDLQ + } + } + for _, rec := range qs.scheduled { + if rec.ID == id { + return driver.ErrJobNotDLQ + } + } + } + return driver.ErrJobNotFound + } + + if opt.Queue != "" && opt.Queue != q { + return fmt.Errorf("queue mismatch: job is in %q, not %q", q, opt.Queue) + } + + qs := d.queues[q] + if qs == nil { + return driver.ErrJobNotFound + } + + item, ok := qs.dlq[id] + if !ok { + return driver.ErrJobNotFound + } + + delete(qs.dlq, id) + delete(d.dlqIndex, id) + + rec := item.Record + rec.Queue = q + rec.RunAt = opt.RunAt // zero => immediate + + if opt.ResetAttempts { + rec.Attempts = 0 + rec.LastError = "" + rec.FailedAt = time.Time{} + } + + if rec.RunAt.IsZero() || !rec.RunAt.After(now) { + rec.RunAt = time.Time{} + qs.runnable = append(qs.runnable, rec) + return nil + } + + heap.Push(&qs.scheduled, rec) + return nil +} + +func stateFromRunAt(now, runAt time.Time) driver.JobState { + if runAt.IsZero() || !runAt.After(now) { + return driver.StateReady + } + return driver.StateScheduled +} + +func summaryFromRecord(now time.Time, rec driver.JobRecord, lease *driver.LeaseInfo, dlq *driver.DLQInfo) driver.JobSummary { + s := driver.JobSummary{ + ID: rec.ID, + Type: rec.Type, + Queue: rec.Queue, + RunAt: rec.RunAt, + CreatedAt: rec.CreatedAt, + Timeout: rec.Timeout, + Attempts: rec.Attempts, + MaxAttempts: rec.MaxAttempts, + LastError: rec.LastError, + FailedAt: rec.FailedAt, + State: stateFromRunAt(now, rec.RunAt), + } + + if lease != nil { + s.State = driver.StateInflight + s.LeaseExpiresAt = lease.ExpiresAt + } + if dlq != nil { + s.State = driver.StateDLQ + s.DLQReason = dlq.Reason + s.DLQFailedAt = dlq.FailedAt + } + + // done is inferred later via state override in List when needed, + // but for memory we don’t have a separate “done container” per queue. + // Inspect sets done explicitly. + return s +} diff --git a/taskharbor/driver/memory/memory.go b/taskharbor/driver/memory/memory.go index dfb910d..fcf3983 100644 --- a/taskharbor/driver/memory/memory.go +++ b/taskharbor/driver/memory/memory.go @@ -29,8 +29,9 @@ type Driver struct { inflightIndex map[string]string idemIndex map[string]string - doneIndex map[string]struct{} - dlqIndex map[string]string + doneIndex map[string]struct{} + dlqIndex map[string]string + doneRecords map[string]driver.JobRecord closed bool } @@ -78,6 +79,11 @@ func (d *Driver) Reset() error { d.queues = make(map[string]*queueState) d.inflightIndex = make(map[string]string) d.idemIndex = make(map[string]string) + + d.doneIndex = make(map[string]struct{}) + d.doneRecords = make(map[string]driver.JobRecord) + d.dlqIndex = make(map[string]string) + d.closed = false return nil } @@ -123,6 +129,7 @@ func New() *Driver { idemIndex: make(map[string]string), doneIndex: make(map[string]struct{}), dlqIndex: make(map[string]string), + doneRecords: make(map[string]driver.JobRecord), } return &driver } @@ -415,6 +422,9 @@ func (d *Driver) Ack( delete(qs.inflight, id) delete(d.inflightIndex, id) + d.doneIndex[id] = struct{}{} + d.doneRecords[id] = it.rec + d.doneIndex[id] = struct{}{} return nil diff --git a/taskharbor/driver/memory/memory_admin_test.go b/taskharbor/driver/memory/memory_admin_test.go new file mode 100644 index 0000000..43c93be --- /dev/null +++ b/taskharbor/driver/memory/memory_admin_test.go @@ -0,0 +1,151 @@ +package memory + +import ( + "context" + "testing" + "time" + + "github.com/ARJ2211/taskharbor/taskharbor/driver" +) + +func TestAdmin_Inspect_List_RequeueDLQ(t *testing.T) { + d := New() + ctx := context.Background() + now := time.Now().UTC() + q := "q0" + + recReady := driver.JobRecord{ + ID: "job_ready", + Type: "echo", + Queue: q, + CreatedAt: now.Add(-3 * time.Second), + } + if _, _, err := d.Enqueue(ctx, recReady); err != nil { + t.Fatal(err) + } + + recSched := driver.JobRecord{ + ID: "job_sched", + Type: "echo", + Queue: q, + RunAt: now.Add(10 * time.Second), + CreatedAt: now.Add(-2 * time.Second), + } + if _, _, err := d.Enqueue(ctx, recSched); err != nil { + t.Fatal(err) + } + + // Inspect ready + ji, err := d.Inspect(ctx, "job_ready", now) + if err != nil { + t.Fatal(err) + } + if ji.State != driver.StateReady { + t.Fatalf("expected ready, got %s", ji.State) + } + + // Reserve -> inflight + r, lease, ok, err := d.Reserve(ctx, q, now, 5*time.Second) + if err != nil || !ok { + t.Fatalf("reserve err=%v ok=%v", err, ok) + } + if r.ID != "job_ready" { + t.Fatalf("expected job_ready reserved, got %s", r.ID) + } + + ji, err = d.Inspect(ctx, "job_ready", now) + if err != nil { + t.Fatal(err) + } + if ji.State != driver.StateInflight || ji.Lease == nil { + t.Fatalf("expected inflight w/ lease, got state=%s lease=%v", ji.State, ji.Lease) + } + + // Fail -> DLQ + if err := d.Fail(ctx, "job_ready", lease.Token, now, "boom"); err != nil { + t.Fatal(err) + } + ji, err = d.Inspect(ctx, "job_ready", now) + if err != nil { + t.Fatal(err) + } + if ji.State != driver.StateDLQ || ji.DLQ == nil { + t.Fatalf("expected dlq w/ info, got state=%s dlq=%v", ji.State, ji.DLQ) + } + + // Requeue DLQ -> ready + if err := d.RequeueDLQ(ctx, "job_ready", now, driver.RequeueOptions{Queue: q, ResetAttempts: true}); err != nil { + t.Fatal(err) + } + ji, err = d.Inspect(ctx, "job_ready", now) + if err != nil { + t.Fatal(err) + } + if ji.State != driver.StateReady { + t.Fatalf("expected ready after requeue, got %s", ji.State) + } + + // Reserve + Ack -> done + r2, lease2, ok, err := d.Reserve(ctx, q, now, 5*time.Second) + if err != nil || !ok { + t.Fatalf("reserve2 err=%v ok=%v", err, ok) + } + if r2.ID != "job_ready" { + t.Fatalf("expected job_ready reserved again, got %s", r2.ID) + } + if err := d.Ack(ctx, "job_ready", lease2.Token, now); err != nil { + t.Fatal(err) + } + + ji, err = d.Inspect(ctx, "job_ready", now) + if err != nil { + t.Fatal(err) + } + if ji.State != driver.StateDone { + t.Fatalf("expected done, got %s", ji.State) + } + + // List all states in queue + page, err := d.List(ctx, driver.ListRequest{Queue: q, Now: now, Limit: 10}) + if err != nil { + t.Fatal(err) + } + if len(page.Jobs) == 0 { + t.Fatalf("expected some jobs in list") + } +} + +func TestAdmin_List_Pagination(t *testing.T) { + d := New() + ctx := context.Background() + now := time.Now().UTC() + q := "q0" + + for i := 0; i < 3; i++ { + rec := driver.JobRecord{ + ID: "job_" + string(rune('a'+i)), + Type: "echo", + Queue: q, + CreatedAt: now.Add(time.Duration(i) * time.Second), + } + if _, _, err := d.Enqueue(ctx, rec); err != nil { + t.Fatal(err) + } + } + + p1, err := d.List(ctx, driver.ListRequest{Queue: q, Now: now, Limit: 2}) + if err != nil { + t.Fatal(err) + } + if len(p1.Jobs) != 2 || p1.NextCursor == "" { + t.Fatalf("expected 2 jobs + cursor, got %d cursor=%q", len(p1.Jobs), p1.NextCursor) + } + + p2, err := d.List(ctx, driver.ListRequest{Queue: q, Now: now, Limit: 2, Cursor: p1.NextCursor}) + if err != nil { + t.Fatal(err) + } + if len(p2.Jobs) != 1 { + t.Fatalf("expected 1 job on second page, got %d", len(p2.Jobs)) + } +} diff --git a/taskharbor/driver/postgres/admin.go b/taskharbor/driver/postgres/admin.go new file mode 100644 index 0000000..c7a219e --- /dev/null +++ b/taskharbor/driver/postgres/admin.go @@ -0,0 +1,468 @@ +package postgres + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "github.com/ARJ2211/taskharbor/taskharbor/driver" + "github.com/jackc/pgx/v5" +) + +var _ driver.Admin = (*Driver)(nil) + +const qAdminInspect = ` +SELECT + id, + type, + queue, + payload, + run_at, + timeout_nanos, + idempotency_key, + created_at, + attempts, + max_attempts, + last_error, + failed_at, + status, + lease_token, + lease_expires_at, + dlq_reason, + dlq_failed_at +FROM th_jobs +WHERE id = $1 +` + +const qAdminGetStatus = ` +SELECT queue, status +FROM th_jobs +WHERE id = $1 +` + +const qAdminRequeueDLQ = ` +UPDATE th_jobs +SET + status = 'ready', + run_at = $2, + dlq_reason = NULL, + dlq_failed_at = NULL, + lease_token = NULL, + lease_expires_at = NULL, + attempts = CASE WHEN $3 THEN 0 ELSE attempts END, + last_error = CASE WHEN $3 THEN '' ELSE last_error END, + failed_at = CASE WHEN $3 THEN NULL ELSE failed_at END +WHERE id = $1 + AND status = 'dlq' + AND ($4::text IS NULL OR queue = $4) +RETURNING id +` + +func (d *Driver) Inspect(ctx context.Context, id string, now time.Time) (driver.JobInfo, error) { + if err := ctx.Err(); err != nil { + return driver.JobInfo{}, err + } + if err := d.ensureOpen(); err != nil { + return driver.JobInfo{}, err + } + + id = strings.TrimSpace(id) + if id == "" { + return driver.JobInfo{}, driver.ErrJobNotFound + } + if now.IsZero() { + now = time.Now().UTC() + } else { + now = now.UTC() + } + + var ( + dbID string + typ string + queue string + payload []byte + runAtPtr *time.Time + timeoutNanos int64 + idemPtr *string + createdAt time.Time + attempts int + maxAttempts int + lastError string + failedAtPtr *time.Time + status string + leaseTokPtr *string + leaseExpPtr *time.Time + dlqReasonPtr *string + dlqFailedAtPtr *time.Time + ) + + err := d.pool.QueryRow(ctx, qAdminInspect, id).Scan( + &dbID, + &typ, + &queue, + &payload, + &runAtPtr, + &timeoutNanos, + &idemPtr, + &createdAt, + &attempts, + &maxAttempts, + &lastError, + &failedAtPtr, + &status, + &leaseTokPtr, + &leaseExpPtr, + &dlqReasonPtr, + &dlqFailedAtPtr, + ) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return driver.JobInfo{}, driver.ErrJobNotFound + } + return driver.JobInfo{}, err + } + + rec := driver.JobRecord{ + ID: dbID, + Type: typ, + Queue: queue, + Payload: payload, + RunAt: time.Time{}, + Timeout: time.Duration(timeoutNanos), + IdempotencyKey: "", + CreatedAt: createdAt.UTC(), + Attempts: attempts, + MaxAttempts: maxAttempts, + LastError: lastError, + FailedAt: time.Time{}, + } + + if runAtPtr != nil { + rec.RunAt = runAtPtr.UTC() + } + if failedAtPtr != nil { + rec.FailedAt = failedAtPtr.UTC() + } + if idemPtr != nil { + rec.IdempotencyKey = *idemPtr + } + + st := stateFromDB(status, runAtPtr, now) + + var lease *driver.LeaseInfo + if status == "inflight" && leaseTokPtr != nil && leaseExpPtr != nil { + lease = &driver.LeaseInfo{ + Token: driver.LeaseToken(*leaseTokPtr), + ExpiresAt: leaseExpPtr.UTC(), + } + } + + var dlq *driver.DLQInfo + if status == "dlq" && dlqFailedAtPtr != nil { + reason := "" + if dlqReasonPtr != nil { + reason = *dlqReasonPtr + } + dlq = &driver.DLQInfo{ + Reason: reason, + FailedAt: dlqFailedAtPtr.UTC(), + } + } + + return driver.JobInfo{ + Record: rec, + State: st, + Lease: lease, + DLQ: dlq, + }, nil +} + +func (d *Driver) List(ctx context.Context, req driver.ListRequest) (driver.ListPage, error) { + if err := ctx.Err(); err != nil { + return driver.ListPage{}, err + } + if err := d.ensureOpen(); err != nil { + return driver.ListPage{}, err + } + + req.Queue = strings.TrimSpace(req.Queue) + if req.Queue == "" { + return driver.ListPage{}, fmt.Errorf("queue is required") + } + if req.Now.IsZero() { + req.Now = time.Now().UTC() + } else { + req.Now = req.Now.UTC() + } + if req.Limit <= 0 { + req.Limit = 50 + } + + var cur driver.Cursor + if strings.TrimSpace(req.Cursor) != "" { + c, err := driver.DecodeCursor(req.Cursor) + if err != nil { + return driver.ListPage{}, err + } + cur = c + } + + sql, args := buildAdminListQuery(req, cur) + + rows, err := d.pool.Query(ctx, sql, args...) + if err != nil { + return driver.ListPage{}, err + } + defer rows.Close() + + out := make([]driver.JobSummary, 0, req.Limit+1) + for rows.Next() { + var ( + id string + typ string + queue string + runAtPtr *time.Time + timeoutNanos int64 + createdAt time.Time + attempts int + maxAttempts int + lastError string + failedAtPtr *time.Time + status string + leaseExpPtr *time.Time + dlqReasonPtr *string + dlqFailedPtr *time.Time + ) + + if err := rows.Scan( + &id, + &typ, + &queue, + &runAtPtr, + &timeoutNanos, + &createdAt, + &attempts, + &maxAttempts, + &lastError, + &failedAtPtr, + &status, + &leaseExpPtr, + &dlqReasonPtr, + &dlqFailedPtr, + ); err != nil { + return driver.ListPage{}, err + } + + s := driver.JobSummary{ + ID: id, + Type: typ, + Queue: queue, + RunAt: time.Time{}, + CreatedAt: createdAt.UTC(), + Timeout: time.Duration(timeoutNanos), + Attempts: attempts, + MaxAttempts: maxAttempts, + LastError: lastError, + FailedAt: time.Time{}, + State: stateFromDB(status, runAtPtr, req.Now), + } + + if runAtPtr != nil { + s.RunAt = runAtPtr.UTC() + } + if failedAtPtr != nil { + s.FailedAt = failedAtPtr.UTC() + } + if leaseExpPtr != nil { + s.LeaseExpiresAt = leaseExpPtr.UTC() + } + if dlqReasonPtr != nil { + s.DLQReason = *dlqReasonPtr + } + if dlqFailedPtr != nil { + s.DLQFailedAt = dlqFailedPtr.UTC() + } + + out = append(out, s) + } + if err := rows.Err(); err != nil { + return driver.ListPage{}, err + } + + if len(out) == 0 { + return driver.ListPage{Jobs: nil, NextCursor: ""}, nil + } + + if len(out) <= req.Limit { + return driver.ListPage{Jobs: out, NextCursor: ""}, nil + } + + page := out[:req.Limit] + last := page[len(page)-1] + next := driver.EncodeCursor(driver.Cursor{ + State: req.State, + A: last.CreatedAt.UnixNano(), + ID: last.ID, + }) + + return driver.ListPage{Jobs: page, NextCursor: next}, nil +} + +func (d *Driver) RequeueDLQ(ctx context.Context, id string, now time.Time, opt driver.RequeueOptions) error { + if err := ctx.Err(); err != nil { + return err + } + if err := d.ensureOpen(); err != nil { + return err + } + + id = strings.TrimSpace(id) + if id == "" { + return driver.ErrJobNotFound + } + if now.IsZero() { + now = time.Now().UTC() + } else { + now = now.UTC() + } + + var runAt any + if opt.RunAt.IsZero() || !opt.RunAt.After(now) { + runAt = nil + } else { + runAt = opt.RunAt.UTC() + } + + var queueGuard any + if strings.TrimSpace(opt.Queue) == "" { + queueGuard = nil + } else { + queueGuard = opt.Queue + } + + var ignored string + err := d.pool.QueryRow(ctx, qAdminRequeueDLQ, id, runAt, opt.ResetAttempts, queueGuard).Scan(&ignored) + if err == nil { + return nil + } + if !errors.Is(err, pgx.ErrNoRows) { + return err + } + + // classify + var q string + var status string + err2 := d.pool.QueryRow(ctx, qAdminGetStatus, id).Scan(&q, &status) + if err2 != nil { + if errors.Is(err2, pgx.ErrNoRows) { + return driver.ErrJobNotFound + } + return err2 + } + + if opt.Queue != "" && q != opt.Queue { + return fmt.Errorf("queue mismatch: job is in %q, not %q", q, opt.Queue) + } + + if status != "dlq" { + return driver.ErrJobNotDLQ + } + + return driver.ErrJobNotDLQ +} + +func stateFromDB(status string, runAt *time.Time, now time.Time) driver.JobState { + switch status { + case "ready": + if runAt == nil || !runAt.UTC().After(now) { + return driver.StateReady + } + return driver.StateScheduled + case "inflight": + return driver.StateInflight + case "dlq": + return driver.StateDLQ + case "done": + return driver.StateDone + default: + // safest fallback + if runAt == nil || !runAt.UTC().After(now) { + return driver.StateReady + } + return driver.StateScheduled + } +} + +func buildAdminListQuery(req driver.ListRequest, cur driver.Cursor) (string, []any) { + // We use stable ordering: created_at ASC, id ASC + // Cursor is (created_at, id) > (cursorTime, cursorID) + args := make([]any, 0, 6) + + where := "WHERE queue = $1" + args = append(args, req.Queue) + p := 2 + + needNow := false + + switch req.State { + case "": + // no filter + case driver.StateReady: + where += fmt.Sprintf(" AND status = 'ready' AND (run_at IS NULL OR run_at <= $%d)", p) + args = append(args, req.Now) + p++ + needNow = true + case driver.StateScheduled: + where += fmt.Sprintf(" AND status = 'ready' AND run_at > $%d", p) + args = append(args, req.Now) + p++ + needNow = true + case driver.StateInflight: + where += " AND status = 'inflight'" + case driver.StateDLQ: + where += " AND status = 'dlq'" + case driver.StateDone: + where += " AND status = 'done'" + default: + // unknown state filter: return empty page rather than surprise + where += " AND 1=0" + } + + if strings.TrimSpace(req.Cursor) != "" { + cursorTime := time.Unix(0, cur.A).UTC() + where += fmt.Sprintf(" AND (created_at > $%d OR (created_at = $%d AND id > $%d))", p, p, p+1) + args = append(args, cursorTime, cur.ID) + p += 2 + } + + limit := req.Limit + 1 + args = append(args, limit) + + // select fields needed for JobSummary + sql := ` +SELECT + id, + type, + queue, + run_at, + timeout_nanos, + created_at, + attempts, + max_attempts, + last_error, + failed_at, + status, + lease_expires_at, + dlq_reason, + dlq_failed_at +FROM th_jobs +` + where + ` +ORDER BY created_at ASC, id ASC +LIMIT $` + fmt.Sprintf("%d", p) + + _ = needNow // just to make intent obvious while editing + + return sql, args +} diff --git a/taskharbor/driver/postgres/postgres_admin_test.go b/taskharbor/driver/postgres/postgres_admin_test.go new file mode 100644 index 0000000..10d0d4e --- /dev/null +++ b/taskharbor/driver/postgres/postgres_admin_test.go @@ -0,0 +1,220 @@ +package postgres + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "github.com/ARJ2211/taskharbor/taskharbor/driver" + "github.com/ARJ2211/taskharbor/taskharbor/internal/envutil" + "github.com/jackc/pgx/v5/pgxpool" +) + +func newAdminPoolAndDriver(t *testing.T) (context.Context, *pgxpool.Pool, *Driver) { + t.Helper() + + wd, _ := os.Getwd() + _ = envutil.LoadRepoDotenv(wd) + + dsn := os.Getenv("TASKHARBOR_TEST_DSN") + if dsn == "" { + t.Skip("TASKHARBOR_TEST_DSN not set") + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + t.Cleanup(cancel) + + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatalf("pgxpool.New: %v", err) + } + t.Cleanup(pool.Close) + + if err := ApplyMigrations(ctx, pool); err != nil { + t.Fatalf("ApplyMigrations: %v", err) + } + if _, err := pool.Exec(ctx, `DELETE FROM th_jobs`); err != nil { + t.Fatalf("cleanup: %v", err) + } + + d, err := NewWithPool(pool) + if err != nil { + t.Fatalf("NewWithPool: %v", err) + } + return ctx, pool, d +} + +func TestPostgresAdmin_Inspect_List_RequeueDLQ(t *testing.T) { + ctx, pool, d := newAdminPoolAndDriver(t) + _ = pool + + now := time.Date(2026, 2, 23, 12, 0, 0, 0, time.UTC) + + // scheduled job + recSched := driver.JobRecord{ + ID: "admin_sched_1", + Type: "t", + Queue: "default", + Payload: []byte(`{"x":1}`), + RunAt: now.Add(30 * time.Second), + CreatedAt: now.Add(-3 * time.Second), + MaxAttempts: 3, + } + if _, _, err := d.Enqueue(ctx, recSched); err != nil { + t.Fatalf("Enqueue sched: %v", err) + } + + // job we'll push to DLQ with attempts>0, then requeue with reset + recDLQ := driver.JobRecord{ + ID: "admin_dlq_1", + Type: "t", + Queue: "default", + Payload: []byte(`{}`), + CreatedAt: now.Add(-2 * time.Second), + MaxAttempts: 3, + } + if _, _, err := d.Enqueue(ctx, recDLQ); err != nil { + t.Fatalf("Enqueue dlq: %v", err) + } + + // reserve + retry to bump attempts + _, lease1, ok, err := d.Reserve(ctx, "default", now, 5*time.Second) + if err != nil || !ok { + t.Fatalf("Reserve#1: ok=%v err=%v", ok, err) + } + if err := d.Retry(ctx, recDLQ.ID, lease1.Token, now.Add(1*time.Second), driver.RetryUpdate{ + Attempts: 1, + LastError: "boom", + FailedAt: now.Add(1 * time.Second), + RunAt: time.Time{}, + }); err != nil { + t.Fatalf("Retry: %v", err) + } + + // reserve again + fail to DLQ + _, lease2, ok, err := d.Reserve(ctx, "default", now.Add(2*time.Second), 5*time.Second) + if err != nil || !ok { + t.Fatalf("Reserve#2: ok=%v err=%v", ok, err) + } + if err := d.Fail(ctx, recDLQ.ID, lease2.Token, now.Add(3*time.Second), "max attempts reached"); err != nil { + t.Fatalf("Fail: %v", err) + } + + // Inspect scheduled + ji, err := d.Inspect(ctx, recSched.ID, now) + if err != nil { + t.Fatalf("Inspect sched: %v", err) + } + if ji.State != driver.StateScheduled { + t.Fatalf("expected scheduled, got %s", ji.State) + } + + // Inspect dlq + ji2, err := d.Inspect(ctx, recDLQ.ID, now.Add(4*time.Second)) + if err != nil { + t.Fatalf("Inspect dlq: %v", err) + } + if ji2.State != driver.StateDLQ || ji2.DLQ == nil { + t.Fatalf("expected dlq with info, got state=%s dlq=%v", ji2.State, ji2.DLQ) + } + + // List scheduled only + pageSched, err := d.List(ctx, driver.ListRequest{ + Queue: "default", + State: driver.StateScheduled, + Now: now, + Limit: 10, + }) + if err != nil { + t.Fatalf("List scheduled: %v", err) + } + if len(pageSched.Jobs) != 1 || pageSched.Jobs[0].ID != recSched.ID { + t.Fatalf("expected 1 scheduled job %s, got %#v", recSched.ID, pageSched.Jobs) + } + + // List dlq only + pageDLQ, err := d.List(ctx, driver.ListRequest{ + Queue: "default", + State: driver.StateDLQ, + Now: now, + Limit: 10, + }) + if err != nil { + t.Fatalf("List dlq: %v", err) + } + if len(pageDLQ.Jobs) != 1 || pageDLQ.Jobs[0].ID != recDLQ.ID { + t.Fatalf("expected 1 dlq job %s, got %#v", recDLQ.ID, pageDLQ.Jobs) + } + + // Requeue from DLQ with reset attempts + if err := d.RequeueDLQ(ctx, recDLQ.ID, now.Add(5*time.Second), driver.RequeueOptions{ + Queue: "default", + ResetAttempts: true, + }); err != nil { + t.Fatalf("RequeueDLQ: %v", err) + } + + ji3, err := d.Inspect(ctx, recDLQ.ID, now.Add(5*time.Second)) + if err != nil { + t.Fatalf("Inspect after requeue: %v", err) + } + if ji3.State != driver.StateReady { + t.Fatalf("expected ready after requeue, got %s", ji3.State) + } + if ji3.Record.Attempts != 0 || ji3.Record.LastError != "" { + t.Fatalf("expected attempts reset, got attempts=%d last_error=%q", ji3.Record.Attempts, ji3.Record.LastError) + } + + // Reservable again + _, lease3, ok, err := d.Reserve(ctx, "default", now.Add(6*time.Second), 5*time.Second) + if err != nil || !ok { + t.Fatalf("Reserve after requeue: ok=%v err=%v", ok, err) + } + if err := d.Ack(ctx, recDLQ.ID, lease3.Token, now.Add(7*time.Second)); err != nil { + t.Fatalf("Ack: %v", err) + } + + ji4, err := d.Inspect(ctx, recDLQ.ID, now.Add(8*time.Second)) + if err != nil { + t.Fatalf("Inspect done: %v", err) + } + if ji4.State != driver.StateDone { + t.Fatalf("expected done, got %s", ji4.State) + } +} + +func TestPostgresAdmin_List_Pagination(t *testing.T) { + ctx, _, d := newAdminPoolAndDriver(t) + now := time.Date(2026, 2, 23, 12, 0, 0, 0, time.UTC) + + for i := 0; i < 3; i++ { + rec := driver.JobRecord{ + ID: fmt.Sprintf("admin_page_%d", i), + Type: "t", + Queue: "default", + Payload: []byte(`{}`), + CreatedAt: now.Add(time.Duration(i) * time.Second), + } + if _, _, err := d.Enqueue(ctx, rec); err != nil { + t.Fatalf("Enqueue %d: %v", i, err) + } + } + + p1, err := d.List(ctx, driver.ListRequest{Queue: "default", Now: now, Limit: 2}) + if err != nil { + t.Fatalf("List p1: %v", err) + } + if len(p1.Jobs) != 2 || p1.NextCursor == "" { + t.Fatalf("expected 2 jobs + cursor, got %d cursor=%q", len(p1.Jobs), p1.NextCursor) + } + + p2, err := d.List(ctx, driver.ListRequest{Queue: "default", Now: now, Limit: 2, Cursor: p1.NextCursor}) + if err != nil { + t.Fatalf("List p2: %v", err) + } + if len(p2.Jobs) != 1 { + t.Fatalf("expected 1 job on second page, got %d", len(p2.Jobs)) + } +} diff --git a/taskharbor/driver/redis/admin.go b/taskharbor/driver/redis/admin.go new file mode 100644 index 0000000..f479e7d --- /dev/null +++ b/taskharbor/driver/redis/admin.go @@ -0,0 +1,694 @@ +package redis + +import ( + "context" + "fmt" + "sort" + "strconv" + "strings" + "time" + + "github.com/ARJ2211/taskharbor/taskharbor/driver" + "github.com/redis/go-redis/v9" +) + +type hgetAllCmd interface { + Result() (map[string]string, error) +} + +var _ driver.Admin = (*Driver)(nil) + +func (d *Driver) keyReady(queue string) string { return d.opts.prefix + ":queue:" + queue + ":ready" } +func (d *Driver) keyScheduled(queue string) string { + return d.opts.prefix + ":queue:" + queue + ":scheduled" +} +func (d *Driver) keyInflight(queue string) string { + return d.opts.prefix + ":queue:" + queue + ":inflight" +} +func (d *Driver) keyDLQ(queue string) string { return d.opts.prefix + ":queue:" + queue + ":dlq" } +func (d *Driver) keyDone(queue string) string { return d.opts.prefix + ":queue:" + queue + ":done" } + +type jobMeta struct { + rec driver.JobRecord + status string + leaseTok string + leaseExp int64 + dlqReason string + dlqFail int64 +} + +func (d *Driver) Inspect(ctx context.Context, id string, now time.Time) (driver.JobInfo, error) { + if err := ctx.Err(); err != nil { + return driver.JobInfo{}, err + } + if err := d.ensureOpen(); err != nil { + return driver.JobInfo{}, err + } + + id = strings.TrimSpace(id) + if id == "" { + return driver.JobInfo{}, driver.ErrJobNotFound + } + if now.IsZero() { + now = time.Now().UTC() + } else { + now = now.UTC() + } + + m, err := d.loadJobMeta(ctx, id) + if err != nil { + return driver.JobInfo{}, err + } + + state := stateFromRedis(now, m.status, m.rec.RunAt) + + var lease *driver.LeaseInfo + if state == driver.StateInflight && m.leaseTok != "" && m.leaseExp != 0 { + lease = &driver.LeaseInfo{ + Token: driver.LeaseToken(m.leaseTok), + ExpiresAt: time.Unix(0, m.leaseExp).UTC(), + } + } + + // Important: always return DLQ info when state is DLQ (even if dlq_failed_at_nano isn't stored). + var dlq *driver.DLQInfo + if state == driver.StateDLQ { + failedAt := time.Time{} + if m.dlqFail != 0 { + failedAt = time.Unix(0, m.dlqFail).UTC() + } else if !m.rec.FailedAt.IsZero() { + failedAt = m.rec.FailedAt.UTC() + } + dlq = &driver.DLQInfo{ + Reason: m.dlqReason, + FailedAt: failedAt, + } + } + + return driver.JobInfo{ + Record: m.rec, + State: state, + Lease: lease, + DLQ: dlq, + }, nil +} + +func (d *Driver) List(ctx context.Context, req driver.ListRequest) (driver.ListPage, error) { + if err := ctx.Err(); err != nil { + return driver.ListPage{}, err + } + if err := d.ensureOpen(); err != nil { + return driver.ListPage{}, err + } + + req.Queue = strings.TrimSpace(req.Queue) + if req.Queue == "" { + return driver.ListPage{}, fmt.Errorf("queue is required") + } + if req.Now.IsZero() { + req.Now = time.Now().UTC() + } else { + req.Now = req.Now.UTC() + } + if req.Limit <= 0 { + req.Limit = 50 + } + + var cur driver.Cursor + if strings.TrimSpace(req.Cursor) != "" { + c, err := driver.DecodeCursor(req.Cursor) + if err != nil { + return driver.ListPage{}, err + } + cur = c + } + + cap := req.Limit * 10 + if cap < 200 { + cap = 200 + } + if cap > 2000 { + cap = 2000 + } + + ids, err := d.collectIDsForList(ctx, req, cap) + if err != nil { + return driver.ListPage{}, err + } + if len(ids) == 0 { + return driver.ListPage{Jobs: nil, NextCursor: ""}, nil + } + + metas, err := d.loadManyJobMeta(ctx, ids) + if err != nil { + return driver.ListPage{}, err + } + + jobs := make([]driver.JobSummary, 0, len(ids)) + for _, id := range ids { + m, ok := metas[id] + if !ok { + continue + } + st := stateFromRedis(req.Now, m.status, m.rec.RunAt) + if req.State != "" && st != req.State { + continue + } + jobs = append(jobs, summaryFromMeta(req.Now, st, m)) + } + + sort.Slice(jobs, func(i, j int) bool { + ai := jobs[i].CreatedAt.UnixNano() + aj := jobs[j].CreatedAt.UnixNano() + if ai != aj { + return ai < aj + } + return jobs[i].ID < jobs[j].ID + }) + + if strings.TrimSpace(req.Cursor) != "" { + out := jobs[:0] + for _, s := range jobs { + a := s.CreatedAt.UnixNano() + if a > cur.A || (a == cur.A && s.ID > cur.ID) { + out = append(out, s) + } + } + jobs = out + } + + if len(jobs) == 0 { + return driver.ListPage{Jobs: nil, NextCursor: ""}, nil + } + if len(jobs) <= req.Limit { + return driver.ListPage{Jobs: jobs, NextCursor: ""}, nil + } + + page := jobs[:req.Limit] + last := page[len(page)-1] + next := driver.EncodeCursor(driver.Cursor{ + State: req.State, + A: last.CreatedAt.UnixNano(), + ID: last.ID, + }) + + return driver.ListPage{Jobs: page, NextCursor: next}, nil +} + +func (d *Driver) RequeueDLQ(ctx context.Context, id string, now time.Time, opt driver.RequeueOptions) error { + if err := ctx.Err(); err != nil { + return err + } + if err := d.ensureOpen(); err != nil { + return err + } + + id = strings.TrimSpace(id) + if id == "" { + return driver.ErrJobNotFound + } + if now.IsZero() { + now = time.Now().UTC() + } else { + now = now.UTC() + } + + runAtNano := int64(0) + runAtSec := int64(0) + runAtMember := "" + + if !opt.RunAt.IsZero() && opt.RunAt.UTC().After(now) { + ru := opt.RunAt.UTC() + runAtNano = ru.UnixNano() + runAtSec = ru.Unix() + runAtMember = schedMember(int64(ru.Nanosecond()), id) + } + + queueGuard := strings.TrimSpace(opt.Queue) + + code, err := d.runRequeueDLQScript( + ctx, + id, + now.UnixNano(), + queueGuard, + runAtNano, + runAtSec, + runAtMember, + opt.ResetAttempts, + ) + if err != nil { + return err + } + + switch code { + case 1: + return nil + case 0: + return driver.ErrJobNotFound + case -1: + return driver.ErrJobNotDLQ + case -2: + return fmt.Errorf("queue mismatch for job %s", id) + default: + return driver.ErrJobNotDLQ + } +} + +func stateFromRedis(now time.Time, status string, runAt time.Time) driver.JobState { + switch status { + case "inflight": + return driver.StateInflight + case "dlq": + return driver.StateDLQ + case "done": + return driver.StateDone + default: + // ready/scheduled stored status is not authoritative; run_at decides. + if runAt.IsZero() || !runAt.After(now) { + return driver.StateReady + } + return driver.StateScheduled + } +} + +func summaryFromMeta(now time.Time, st driver.JobState, m jobMeta) driver.JobSummary { + s := driver.JobSummary{ + ID: m.rec.ID, + Type: m.rec.Type, + Queue: m.rec.Queue, + State: st, + RunAt: m.rec.RunAt, + CreatedAt: m.rec.CreatedAt, + Timeout: m.rec.Timeout, + Attempts: m.rec.Attempts, + MaxAttempts: m.rec.MaxAttempts, + LastError: m.rec.LastError, + FailedAt: m.rec.FailedAt, + } + + if m.leaseExp != 0 { + s.LeaseExpiresAt = time.Unix(0, m.leaseExp).UTC() + } + if m.dlqReason != "" { + s.DLQReason = m.dlqReason + } + if m.dlqFail != 0 { + s.DLQFailedAt = time.Unix(0, m.dlqFail).UTC() + } + + return s +} + +func (d *Driver) loadJobMeta(ctx context.Context, id string) (jobMeta, error) { + jobKey := d.keyJob(id) + m, err := d.client.HGetAll(ctx, jobKey).Result() + if err != nil { + return jobMeta{}, err + } + if len(m) == 0 { + return jobMeta{}, driver.ErrJobNotFound + } + + rec := driver.JobRecord{ID: id} + rec.Type = m["type"] + rec.Queue = m["queue"] + if v, ok := m["payload"]; ok { + rec.Payload = []byte(v) + } + rec.IdempotencyKey = m["idempotency_key"] + rec.LastError = m["last_error"] + + if v := m["run_at_nano"]; v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil && n != 0 { + rec.RunAt = time.Unix(0, n).UTC() + } + } + if v := m["timeout_nano"]; v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil { + rec.Timeout = time.Duration(n) + } + } + if v := m["created_at_nano"]; v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil { + rec.CreatedAt = time.Unix(0, n).UTC() + } + } + if v := m["failed_at_nano"]; v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil && n != 0 { + rec.FailedAt = time.Unix(0, n).UTC() + } + } + if v := m["attempts"]; v != "" { + rec.Attempts, _ = strconv.Atoi(v) + } + if v := m["max_attempts"]; v != "" { + rec.MaxAttempts, _ = strconv.Atoi(v) + } + + meta := jobMeta{ + rec: rec, + status: m["status"], + leaseTok: m["lease_token"], + dlqReason: m["dlq_reason"], + } + + if v := m["lease_expires_at_nano"]; v != "" { + meta.leaseExp, _ = strconv.ParseInt(v, 10, 64) + } + if v := m["dlq_failed_at_nano"]; v != "" { + meta.dlqFail, _ = strconv.ParseInt(v, 10, 64) + } + + // Fallback for older/legacy Redis schema: DLQ time/reason might live in failed_at_nano/last_error. + if meta.status == "dlq" { + if meta.dlqReason == "" { + meta.dlqReason = rec.LastError + } + if meta.dlqFail == 0 && !rec.FailedAt.IsZero() { + meta.dlqFail = rec.FailedAt.UnixNano() + } + } + + return meta, nil +} + +func (d *Driver) loadManyJobMeta(ctx context.Context, ids []string) (map[string]jobMeta, error) { + pipe := d.client.Pipeline() + cmds := make(map[string]hgetAllCmd, len(ids)) + + for _, id := range ids { + cmds[id] = pipe.HGetAll(ctx, d.keyJob(id)) + } + + _, err := pipe.Exec(ctx) + if err != nil && err != redis.Nil { + return nil, err + } + + out := make(map[string]jobMeta, len(ids)) + for id, cmd := range cmds { + m, err := cmd.Result() + if err != nil || len(m) == 0 { + continue + } + + rec := driver.JobRecord{ID: id} + rec.Type = m["type"] + rec.Queue = m["queue"] + if v, ok := m["payload"]; ok { + rec.Payload = []byte(v) + } + rec.IdempotencyKey = m["idempotency_key"] + rec.LastError = m["last_error"] + + if v := m["run_at_nano"]; v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil && n != 0 { + rec.RunAt = time.Unix(0, n).UTC() + } + } + if v := m["timeout_nano"]; v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil { + rec.Timeout = time.Duration(n) + } + } + if v := m["created_at_nano"]; v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil { + rec.CreatedAt = time.Unix(0, n).UTC() + } + } + if v := m["failed_at_nano"]; v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil && n != 0 { + rec.FailedAt = time.Unix(0, n).UTC() + } + } + if v := m["attempts"]; v != "" { + rec.Attempts, _ = strconv.Atoi(v) + } + if v := m["max_attempts"]; v != "" { + rec.MaxAttempts, _ = strconv.Atoi(v) + } + + meta := jobMeta{ + rec: rec, + status: m["status"], + leaseTok: m["lease_token"], + dlqReason: m["dlq_reason"], + } + if v := m["lease_expires_at_nano"]; v != "" { + meta.leaseExp, _ = strconv.ParseInt(v, 10, 64) + } + if v := m["dlq_failed_at_nano"]; v != "" { + meta.dlqFail, _ = strconv.ParseInt(v, 10, 64) + } + + // same fallback in bulk path + if meta.status == "dlq" { + if meta.dlqReason == "" { + meta.dlqReason = rec.LastError + } + if meta.dlqFail == 0 && !rec.FailedAt.IsZero() { + meta.dlqFail = rec.FailedAt.UnixNano() + } + } + + out[id] = meta + } + + return out, nil +} + +func (d *Driver) collectIDsForList(ctx context.Context, req driver.ListRequest, cap int) ([]string, error) { + queue := req.Queue + + add := func(dst []string, ids ...string) []string { + for _, id := range ids { + if id != "" { + dst = append(dst, id) + } + } + return dst + } + + uniq := make(map[string]struct{}) + out := make([]string, 0, cap) + + pushUnique := func(id string) { + if id == "" { + return + } + if _, ok := uniq[id]; ok { + return + } + uniq[id] = struct{}{} + out = append(out, id) + } + + pushMany := func(ids []string) { + for _, id := range ids { + if len(out) >= cap { + return + } + pushUnique(id) + } + } + + switch req.State { + case driver.StateReady: + ids, err := d.client.LRange(ctx, d.keyReady(queue), 0, int64(cap-1)).Result() + if err != nil && err != redis.Nil { + return nil, err + } + pushMany(ids) + + due, err := d.dueScheduledIDs(ctx, queue, req.Now, cap-len(out)) + if err != nil { + return nil, err + } + pushMany(due) + + case driver.StateScheduled: + fut, err := d.futureScheduledIDs(ctx, queue, req.Now, cap) + if err != nil { + return nil, err + } + pushMany(fut) + + case driver.StateInflight: + ids, err := d.client.ZRange(ctx, d.keyInflight(queue), 0, int64(cap-1)).Result() + if err != nil && err != redis.Nil { + return nil, err + } + pushMany(ids) + + case driver.StateDLQ: + ids, err := d.client.LRange(ctx, d.keyDLQ(queue), 0, int64(cap-1)).Result() + if err != nil && err != redis.Nil { + return nil, err + } + pushMany(ids) + + case driver.StateDone: + ids, err := d.client.LRange(ctx, d.keyDone(queue), 0, int64(cap-1)).Result() + if err != nil && err != redis.Nil { + return nil, err + } + pushMany(ids) + + case "": + ids, err := d.client.LRange(ctx, d.keyReady(queue), 0, int64(cap-1)).Result() + if err != nil && err != redis.Nil { + return nil, err + } + pushMany(ids) + + sched, err := d.client.ZRange(ctx, d.keyScheduled(queue), 0, int64(cap-1)).Result() + if err != nil && err != redis.Nil { + return nil, err + } + for _, m := range sched { + if len(out) >= cap { + break + } + if _, id, ok := strings.Cut(m, ":"); ok { + pushUnique(id) + } + } + + inflight, err := d.client.ZRange(ctx, d.keyInflight(queue), 0, int64(cap-1)).Result() + if err != nil && err != redis.Nil { + return nil, err + } + pushMany(inflight) + + dlq, err := d.client.LRange(ctx, d.keyDLQ(queue), 0, int64(cap-1)).Result() + if err != nil && err != redis.Nil { + return nil, err + } + pushMany(dlq) + + done, err := d.client.LRange(ctx, d.keyDone(queue), 0, int64(cap-1)).Result() + if err != nil && err != redis.Nil { + return nil, err + } + pushMany(done) + + default: + out = add(out) + } + + return out, nil +} + +func (d *Driver) dueScheduledIDs(ctx context.Context, queue string, now time.Time, cap int) ([]string, error) { + if cap <= 0 { + return nil, nil + } + nowSec := now.Unix() + nowSub := int64(now.Nanosecond()) + skey := d.keyScheduled(queue) + + out := make([]string, 0, cap) + + // < nowSec + if nowSec-1 >= 0 { + members, err := d.client.ZRangeByScore(ctx, skey, &redis.ZRangeBy{ + Min: "0", + Max: strconv.FormatInt(nowSec-1, 10), + Count: int64(cap), + }).Result() + if err != nil && err != redis.Nil { + return nil, err + } + for _, m := range members { + if len(out) >= cap { + break + } + if _, id, ok := strings.Cut(m, ":"); ok { + out = append(out, id) + } + } + } + + // == nowSec (filter by sub) + if len(out) < cap { + members, err := d.client.ZRangeByScore(ctx, skey, &redis.ZRangeBy{ + Min: strconv.FormatInt(nowSec, 10), + Max: strconv.FormatInt(nowSec, 10), + }).Result() + if err != nil && err != redis.Nil { + return nil, err + } + for _, m := range members { + if len(out) >= cap { + break + } + if len(m) < 10 { + continue + } + sub, _ := strconv.ParseInt(m[:9], 10, 64) + if sub <= nowSub { + if _, id, ok := strings.Cut(m, ":"); ok { + out = append(out, id) + } + } + } + } + + return out, nil +} + +func (d *Driver) futureScheduledIDs(ctx context.Context, queue string, now time.Time, cap int) ([]string, error) { + if cap <= 0 { + return nil, nil + } + nowSec := now.Unix() + nowSub := int64(now.Nanosecond()) + skey := d.keyScheduled(queue) + + out := make([]string, 0, cap) + + // > nowSec + members, err := d.client.ZRangeByScore(ctx, skey, &redis.ZRangeBy{ + Min: strconv.FormatInt(nowSec+1, 10), + Max: "+inf", + Count: int64(cap), + }).Result() + if err != nil && err != redis.Nil { + return nil, err + } + for _, m := range members { + if len(out) >= cap { + break + } + if _, id, ok := strings.Cut(m, ":"); ok { + out = append(out, id) + } + } + + // == nowSec (filter by sub > nowSub) + if len(out) < cap { + members, err := d.client.ZRangeByScore(ctx, skey, &redis.ZRangeBy{ + Min: strconv.FormatInt(nowSec, 10), + Max: strconv.FormatInt(nowSec, 10), + }).Result() + if err != nil && err != redis.Nil { + return nil, err + } + for _, m := range members { + if len(out) >= cap { + break + } + if len(m) < 10 { + continue + } + sub, _ := strconv.ParseInt(m[:9], 10, 64) + if sub > nowSub { + if _, id, ok := strings.Cut(m, ":"); ok { + out = append(out, id) + } + } + } + } + + return out, nil +} diff --git a/taskharbor/driver/redis/redis_admin_test.go b/taskharbor/driver/redis/redis_admin_test.go new file mode 100644 index 0000000..049d7ee --- /dev/null +++ b/taskharbor/driver/redis/redis_admin_test.go @@ -0,0 +1,184 @@ +package redis + +import ( + "context" + "os" + "testing" + "time" + + "github.com/ARJ2211/taskharbor/taskharbor/driver" + "github.com/ARJ2211/taskharbor/taskharbor/internal/envutil" +) + +func TestRedisAdmin_Inspect_List_RequeueDLQ(t *testing.T) { + cwd, _ := os.Getwd() + _ = envutil.LoadRepoDotenv(cwd) + + addr := os.Getenv("REDIS_ADDR") + if addr == "" { + t.Skip("REDIS_ADDR not set") + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + d, err := New(ctx, addr) + if err != nil { + t.Fatalf("New: %v", err) + } + defer func() { _ = d.Close() }() + + // isolate keys + d.opts.prefix = "taskharbor_admin_test:" + time.Now().UTC().Format("20060102150405.000000000") + + now := time.Now().UTC() + q := "q0" + + recSched := driver.JobRecord{ + ID: "admin_sched_1", + Type: "t", + Queue: q, + Payload: []byte(`{"x":1}`), + RunAt: now.Add(10 * time.Second), + CreatedAt: now.Add(-3 * time.Second), + MaxAttempts: 3, + } + if _, _, err := d.Enqueue(ctx, recSched); err != nil { + t.Fatalf("Enqueue sched: %v", err) + } + + recDLQ := driver.JobRecord{ + ID: "admin_dlq_1", + Type: "t", + Queue: q, + Payload: []byte(`{}`), + CreatedAt: now.Add(-2 * time.Second), + MaxAttempts: 3, + } + if _, _, err := d.Enqueue(ctx, recDLQ); err != nil { + t.Fatalf("Enqueue dlq: %v", err) + } + + // Reserve + Fail -> DLQ + _, lease, ok, err := d.Reserve(ctx, q, now, 5*time.Second) + if err != nil || !ok { + t.Fatalf("Reserve: ok=%v err=%v", ok, err) + } + if err := d.Fail(ctx, recDLQ.ID, lease.Token, now.Add(1*time.Second), "boom"); err != nil { + t.Fatalf("Fail: %v", err) + } + + jiSched, err := d.Inspect(ctx, recSched.ID, now) + if err != nil { + t.Fatalf("Inspect sched: %v", err) + } + if jiSched.State != driver.StateScheduled { + t.Fatalf("expected scheduled, got %s", jiSched.State) + } + + jiDLQ, err := d.Inspect(ctx, recDLQ.ID, now) + if err != nil { + t.Fatalf("Inspect dlq: %v", err) + } + if jiDLQ.State != driver.StateDLQ || jiDLQ.DLQ == nil { + t.Fatalf("expected dlq with info, got state=%s dlq=%v", jiDLQ.State, jiDLQ.DLQ) + } + + pageDLQ, err := d.List(ctx, driver.ListRequest{Queue: q, State: driver.StateDLQ, Now: now, Limit: 10}) + if err != nil { + t.Fatalf("List dlq: %v", err) + } + if len(pageDLQ.Jobs) != 1 || pageDLQ.Jobs[0].ID != recDLQ.ID { + t.Fatalf("expected dlq job %s, got %#v", recDLQ.ID, pageDLQ.Jobs) + } + + // Requeue DLQ -> ready + if err := d.RequeueDLQ(ctx, recDLQ.ID, now, driver.RequeueOptions{Queue: q, ResetAttempts: true}); err != nil { + t.Fatalf("RequeueDLQ: %v", err) + } + jiReady, err := d.Inspect(ctx, recDLQ.ID, now) + if err != nil { + t.Fatalf("Inspect after requeue: %v", err) + } + if jiReady.State != driver.StateReady { + t.Fatalf("expected ready, got %s", jiReady.State) + } + + // Reserve + Ack -> done + _, lease2, ok, err := d.Reserve(ctx, q, now.Add(2*time.Second), 5*time.Second) + if err != nil || !ok { + t.Fatalf("Reserve2: ok=%v err=%v", ok, err) + } + if err := d.Ack(ctx, recDLQ.ID, lease2.Token, now.Add(3*time.Second)); err != nil { + t.Fatalf("Ack: %v", err) + } + + jiDone, err := d.Inspect(ctx, recDLQ.ID, now.Add(3*time.Second)) + if err != nil { + t.Fatalf("Inspect done: %v", err) + } + if jiDone.State != driver.StateDone { + t.Fatalf("expected done, got %s", jiDone.State) + } + + pageDone, err := d.List(ctx, driver.ListRequest{Queue: q, State: driver.StateDone, Now: now, Limit: 10}) + if err != nil { + t.Fatalf("List done: %v", err) + } + if len(pageDone.Jobs) != 1 || pageDone.Jobs[0].ID != recDLQ.ID { + t.Fatalf("expected done job %s, got %#v", recDLQ.ID, pageDone.Jobs) + } +} + +func TestRedisAdmin_List_Pagination(t *testing.T) { + cwd, _ := os.Getwd() + _ = envutil.LoadRepoDotenv(cwd) + + addr := os.Getenv("REDIS_ADDR") + if addr == "" { + t.Skip("REDIS_ADDR not set") + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + d, err := New(ctx, addr) + if err != nil { + t.Fatalf("New: %v", err) + } + defer func() { _ = d.Close() }() + + d.opts.prefix = "taskharbor_admin_page_test:" + time.Now().UTC().Format("20060102150405.000000000") + + now := time.Now().UTC() + q := "q0" + + for i := 0; i < 3; i++ { + rec := driver.JobRecord{ + ID: "admin_page_" + string(rune('a'+i)), + Type: "t", + Queue: q, + Payload: []byte(`{}`), + CreatedAt: now.Add(time.Duration(i) * time.Second), + } + if _, _, err := d.Enqueue(ctx, rec); err != nil { + t.Fatalf("Enqueue %d: %v", i, err) + } + } + + p1, err := d.List(ctx, driver.ListRequest{Queue: q, State: driver.StateReady, Now: now, Limit: 2}) + if err != nil { + t.Fatalf("List p1: %v", err) + } + if len(p1.Jobs) != 2 || p1.NextCursor == "" { + t.Fatalf("expected 2 jobs + cursor, got %d cursor=%q", len(p1.Jobs), p1.NextCursor) + } + + p2, err := d.List(ctx, driver.ListRequest{Queue: q, State: driver.StateReady, Now: now, Limit: 2, Cursor: p1.NextCursor}) + if err != nil { + t.Fatalf("List p2: %v", err) + } + if len(p2.Jobs) != 1 { + t.Fatalf("expected 1 job on second page, got %d", len(p2.Jobs)) + } +} diff --git a/taskharbor/driver/redis/scripts.go b/taskharbor/driver/redis/scripts.go index 56ea7b0..0a3e2c7 100644 --- a/taskharbor/driver/redis/scripts.go +++ b/taskharbor/driver/redis/scripts.go @@ -386,6 +386,7 @@ if tonumber(db_exp) <= now then end local queue = redis.call('HGET', job_key, 'queue') redis.call('HSET', job_key, 'status', 'done', 'lease_token', '', 'lease_expires_at_nano', '0') +redis.call('RPUSH', prefix .. ":queue:" .. queue .. ":done", id) local inflight_key = prefix .. ":queue:" .. queue .. ":inflight" redis.call('ZREM', inflight_key, id) return 1 @@ -548,5 +549,110 @@ func (d *Driver) runFailScript(ctx context.Context, id, token string, nowNano in return n == 1, nil } +const scriptRequeueDLQ = ` +local prefix = KEYS[1] +local id = ARGV[1] +local now_nano = tonumber(ARGV[2]) + +local queue_guard = ARGV[3] -- '' means ignore +local run_at_nano = ARGV[4] -- string +local run_at_sec = tonumber(ARGV[5]) +local run_at_member = ARGV[6] +local reset = tonumber(ARGV[7]) -- 0/1 + +local job_key = prefix .. ":job:" .. id +if redis.call('EXISTS', job_key) == 0 then + return 0 +end + +local status = redis.call('HGET', job_key, 'status') +if status ~= 'dlq' then + return -1 +end + +local queue = redis.call('HGET', job_key, 'queue') +if queue_guard ~= nil and queue_guard ~= '' and queue_guard ~= queue then + return -2 +end + +local dlq_key = prefix .. ":queue:" .. queue .. ":dlq" +redis.call('LREM', dlq_key, 0, id) + +local new_status = 'ready' +if run_at_nano ~= nil and run_at_nano ~= '' and run_at_nano ~= '0' then + if tonumber(run_at_nano) > now_nano then + new_status = 'scheduled' + else + run_at_nano = '0' + end +else + run_at_nano = '0' +end + +if reset == 1 then + redis.call('HSET', job_key, + 'status', new_status, + 'run_at_nano', run_at_nano, + 'attempts', '0', + 'last_error', '', + 'failed_at_nano', '0', + 'dlq_reason', '', + 'dlq_failed_at_nano', '0', + 'lease_token', '', + 'lease_expires_at_nano', '0' + ) +else + redis.call('HSET', job_key, + 'status', new_status, + 'run_at_nano', run_at_nano, + 'dlq_reason', '', + 'dlq_failed_at_nano', '0', + 'lease_token', '', + 'lease_expires_at_nano', '0' + ) +end + +if new_status == 'ready' then + redis.call('RPUSH', prefix .. ":queue:" .. queue .. ":ready", id) +else + redis.call('ZADD', prefix .. ":queue:" .. queue .. ":scheduled", run_at_sec, run_at_member) +end + +return 1 +` + +func (d *Driver) runRequeueDLQScript( + ctx context.Context, + id string, + nowNano int64, + queueGuard string, + runAtNano int64, + runAtSec int64, + runAtMember string, + resetAttempts bool, +) (int64, error) { + keys := []string{d.opts.prefix} + reset := int64(0) + if resetAttempts { + reset = 1 + } + args := []any{ + id, + strconv.FormatInt(nowNano, 10), + queueGuard, + strconv.FormatInt(runAtNano, 10), + strconv.FormatInt(runAtSec, 10), + runAtMember, + strconv.FormatInt(reset, 10), + } + + v, err := d.client.Eval(ctx, scriptRequeueDLQ, keys, args...).Result() + if err != nil { + return 0, err + } + n, _ := toInt64(v) + return n, nil +} + // Silence unused import warnings if you temporarily comment-out scripts during refactors. var _ = errors.New