feat: support scheduled job availability

README.md

@@ -8,6 +8,7 @@ ForgeQueue is a Go and PostgreSQL job queue focused on durable execution, concur
 
 Implemented:
 - Durable job submission and lookup API with idempotency keys
+- Scheduled job availability using an optional `available_at` timestamp
 - Transactional job claiming with `FOR UPDATE SKIP LOCKED`
 - Worker identity, execution leases, and lease renewal
 - Expired-lease recovery and attempt-count increments
@@ -54,6 +55,16 @@ curl -X POST http://localhost:8080/jobs \
   -d '{"queue_name":"default","job_type":"noop","payload":{"message":"hello ForgeQueue"},"priority":10}'
 ```
 
+### Schedule a job
+
+Include `available_at` when submitting a job to make it ineligible for workers until that timestamp. The value must be an RFC 3339 timestamp; omit it to make the job available immediately. The worker claim query enforces this timestamp in PostgreSQL.
+
+```sh
+curl -X POST http://localhost:8080/jobs \
+  -H 'Content-Type: application/json' \
+  -d '{"queue_name":"default","job_type":"report","payload":{},"available_at":"2026-11-02T09:00:00Z"}'
+```
+
 ### Retrieve a job
 
 ```sh
@@ -161,6 +172,6 @@ Publish benchmark numbers only alongside the environment and exact command used.
 - **Milestone 2 — Durable Job API:** complete
 - **Milestone 3 — Queue + Worker Runtime:** complete
 - **Milestone 4 — Reliability + Recovery:** final CI verification in progress
-- **Milestone 5 — Scheduling + Dependencies:** planned
+- **Milestone 5 — Scheduling + Dependencies:** scheduled job availability implemented; job dependencies planned
 - **Milestone 6 — Observability + Load Testing:** initial queue statistics endpoint added; broader metrics and load testing planned
 - **Milestone 7 — Infrastructure + Production Operations:** planned

internal/job/model.go

@@ -38,4 +38,5 @@ type CreateRequest struct {
 	Priority       int             `json:"priority"`
 	MaxAttempts    int             `json:"max_attempts"`
 	IdempotencyKey string          `json:"idempotency_key,omitempty"`
+	AvailableAt    *time.Time      `json:"available_at,omitempty"`
 }

internal/job/postgres_cancel_integration_test.go

@@ -89,16 +89,21 @@ func TestPostgresRepositoryCancelPendingJob(t *testing.T) {
 	}
 
 	repo := NewPostgresRepository(pool)
+	scheduledAt := time.Now().UTC().Add(2 * time.Hour).Truncate(time.Microsecond)
 	pending, err := repo.Create(ctx, CreateRequest{
 		IdempotencyKey: fmt.Sprintf("cancel-pending-%d", time.Now().UnixNano()),
 		QueueName:      "cancel-test",
 		JobType:        "test",
 		Payload:        []byte(`{}`),
 		MaxAttempts:    3,
+		AvailableAt:    &scheduledAt,
 	})
 	if err != nil {
 		t.Fatalf("create pending job: %v", err)
 	}
+	if !pending.AvailableAt.Equal(scheduledAt) {
+		t.Fatalf("available_at = %s; want %s", pending.AvailableAt, scheduledAt)
+	}
 
 	cancelled, err := repo.Cancel(ctx, pending.ID)
 	if err != nil {

internal/job/postgres_repository.go

@@ -25,8 +25,8 @@ func scanJob(row pgx.Row) (Job, error) {
 }
 
 func (r *PostgresRepository) Create(ctx context.Context, req CreateRequest) (Job, error) {
-	q := "INSERT INTO jobs(idempotency_key,queue_name,job_type,payload,status,priority,max_attempts) VALUES($1,$2,$3,$4,'pending',$5,$6) ON CONFLICT(idempotency_key) DO UPDATE SET idempotency_key=EXCLUDED.idempotency_key RETURNING " + jobColumns
-	return scanJob(r.db.QueryRow(ctx, q, req.IdempotencyKey, req.QueueName, req.JobType, req.Payload, req.Priority, req.MaxAttempts))
+	q := "INSERT INTO jobs(idempotency_key,queue_name,job_type,payload,status,priority,max_attempts,available_at) VALUES($1,$2,$3,$4,'pending',$5,$6,COALESCE($7::timestamptz,NOW())) ON CONFLICT(idempotency_key) DO UPDATE SET idempotency_key=EXCLUDED.idempotency_key RETURNING " + jobColumns
+	return scanJob(r.db.QueryRow(ctx, q, req.IdempotencyKey, req.QueueName, req.JobType, req.Payload, req.Priority, req.MaxAttempts, req.AvailableAt))
 }
 
 func (r *PostgresRepository) Cancel(ctx context.Context, id string) (Job, error) {

internal/job/service.go

@@ -1,7 +1,61 @@
 package job
-import("context";"encoding/json";"errors";"strings")
-type Service struct{repo Repository}
-func NewService(repo Repository)*Service{return &Service{repo:repo}}
-func(s *Service)Create(ctx context.Context,req CreateRequest)(Job,error){req.QueueName=strings.TrimSpace(req.QueueName);req.JobType=strings.TrimSpace(req.JobType);if req.QueueName==""{return Job{},errors.New("queue_name is required")};if req.JobType==""{return Job{},errors.New("job_type is required")};if len(req.Payload)==0||!json.Valid(req.Payload){return Job{},errors.New("payload must be valid JSON")};if req.Priority < -100||req.Priority>100{return Job{},errors.New("priority must be between -100 and 100")};if req.MaxAttempts==0{req.MaxAttempts=3};if req.MaxAttempts<1||req.MaxAttempts>20{return Job{},errors.New("max_attempts must be between 1 and 20")};return s.repo.Create(ctx,req)}
-func(s *Service)Get(ctx context.Context,id string)(Job,error){id=strings.TrimSpace(id);if id==""{return Job{},errors.New("job id is required")};return s.repo.Get(ctx,id)}
-func(s *Service)Cancel(ctx context.Context,id string)(Job,error){id=strings.TrimSpace(id);if id==""{return Job{},errors.New("job id is required")};return s.repo.Cancel(ctx,id)}
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+)
+
+type Service struct {
+	repo Repository
+}
+
+func NewService(repo Repository) *Service {
+	return &Service{repo: repo}
+}
+
+func (s *Service) Create(ctx context.Context, req CreateRequest) (Job, error) {
+	req.QueueName = strings.TrimSpace(req.QueueName)
+	req.JobType = strings.TrimSpace(req.JobType)
+
+	if req.QueueName == "" {
+		return Job{}, errors.New("queue_name is required")
+	}
+	if req.JobType == "" {
+		return Job{}, errors.New("job_type is required")
+	}
+	if len(req.Payload) == 0 || !json.Valid(req.Payload) {
+		return Job{}, errors.New("payload must be valid JSON")
+	}
+	if req.Priority < -100 || req.Priority > 100 {
+		return Job{}, errors.New("priority must be between -100 and 100")
+	}
+	if req.MaxAttempts == 0 {
+		req.MaxAttempts = 3
+	}
+	if req.MaxAttempts < 1 || req.MaxAttempts > 20 {
+		return Job{}, errors.New("max_attempts must be between 1 and 20")
+	}
+	if req.AvailableAt != nil && req.AvailableAt.IsZero() {
+		return Job{}, errors.New("available_at must be a valid timestamp")
+	}
+
+	return s.repo.Create(ctx, req)
+}
+
+func (s *Service) Get(ctx context.Context, id string) (Job, error) {
+	id = strings.TrimSpace(id)
+	if id == "" {
+		return Job{}, errors.New("job id is required")
+	}
+	return s.repo.Get(ctx, id)
+}
+
+func (s *Service) Cancel(ctx context.Context, id string) (Job, error) {
+	id = strings.TrimSpace(id)
+	if id == "" {
+		return Job{}, errors.New("job id is required")
+	}
+	return s.repo.Cancel(ctx, id)
+}

internal/job/service_schedule_test.go

@@ -0,0 +1,70 @@
+package job
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+)
+
+type scheduledRequestRepository struct {
+	request CreateRequest
+	called  bool
+}
+
+func (r *scheduledRequestRepository) Create(_ context.Context, req CreateRequest) (Job, error) {
+	r.called = true
+	r.request = req
+	return Job{}, nil
+}
+
+func (r *scheduledRequestRepository) Get(context.Context, string) (Job, error) {
+	return Job{}, nil
+}
+
+func (r *scheduledRequestRepository) Cancel(context.Context, string) (Job, error) {
+	return Job{}, nil
+}
+
+func TestCreatePassesAvailableAtToRepository(t *testing.T) {
+	availableAt := time.Date(2026, time.November, 2, 9, 0, 0, 0, time.UTC)
+	repo := &scheduledRequestRepository{}
+	service := NewService(repo)
+
+	_, err := service.Create(context.Background(), CreateRequest{
+		QueueName:   "reports",
+		JobType:     "daily-report",
+		Payload:     json.RawMessage(`{}`),
+		MaxAttempts: 3,
+		AvailableAt: &availableAt,
+	})
+	if err != nil {
+		t.Fatalf("create scheduled job: %v", err)
+	}
+	if !repo.called {
+		t.Fatal("expected repository to be called")
+	}
+	if repo.request.AvailableAt == nil || !repo.request.AvailableAt.Equal(availableAt) {
+		t.Fatalf("repository available_at = %v; want %s", repo.request.AvailableAt, availableAt)
+	}
+}
+
+func TestCreateRejectsZeroAvailableAt(t *testing.T) {
+	repo := &scheduledRequestRepository{}
+	service := NewService(repo)
+	zero := time.Time{}
+
+	_, err := service.Create(context.Background(), CreateRequest{
+		QueueName:   "reports",
+		JobType:     "daily-report",
+		Payload:     json.RawMessage(`{}`),
+		MaxAttempts: 3,
+		AvailableAt: &zero,
+	})
+	if err == nil {
+		t.Fatal("expected zero available_at to be rejected")
+	}
+	if repo.called {
+		t.Fatal("repository must not be called for an invalid available_at")
+	}
+}

internal/queue/postgres_scheduling_integration_test.go

@@ -0,0 +1,45 @@
+package queue
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/nooby6/forgequeue/internal/job"
+)
+
+func TestPostgresQueueDoesNotClaimScheduledJobBeforeAvailableAt(t *testing.T) {
+	pool := integrationPool(t)
+	ctx := context.Background()
+	queueName := fmt.Sprintf("scheduled-%d", time.Now().UnixNano())
+	availableAt := time.Now().UTC().Add(time.Hour)
+
+	created, err := job.NewPostgresRepository(pool).Create(ctx, job.CreateRequest{
+		QueueName:   queueName,
+		JobType:     "scheduled-test",
+		Payload:     []byte(`{}`),
+		MaxAttempts: 3,
+		AvailableAt: &availableAt,
+	})
+	if err != nil {
+		t.Fatalf("create scheduled job: %v", err)
+	}
+
+	q := NewPostgresQueue(pool)
+	if _, err := q.ClaimNext(ctx, queueName, "scheduled-worker", time.Minute); !errors.Is(err, ErrNoJob) {
+		t.Fatalf("claim before available_at error = %v; want ErrNoJob", err)
+	}
+
+	if _, err := pool.Exec(ctx, "UPDATE jobs SET available_at = NOW() WHERE id = $1", created.ID); err != nil {
+		t.Fatalf("make scheduled job available: %v", err)
+	}
+	claimed, err := q.ClaimNext(ctx, queueName, "scheduled-worker", time.Minute)
+	if err != nil {
+		t.Fatalf("claim job after available_at: %v", err)
+	}
+	if claimed.ID != created.ID {
+		t.Fatalf("claimed job = %s; want %s", claimed.ID, created.ID)
+	}
+}