descrybe/apps/api/internal/jobs/heartbeat.go

package jobs

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

// ProcessingWorkerID is the durable heartbeat row for cmd/worker.
const ProcessingWorkerID = "processing"

// DefaultHeartbeatStaleAfter is how long /readyz tolerates a missing touch.
// Worker poll defaults to 250ms; 60s absorbs brief deploys without masking death.
const DefaultHeartbeatStaleAfter = 60 * time.Second

// TouchHeartbeat upserts last_seen_at for workerID (call from the worker poll loop).
func TouchHeartbeat(ctx context.Context, pool *pgxpool.Pool, workerID string) error {
	if pool == nil {
		return fmt.Errorf("jobs heartbeat: pool unavailable")
	}
	if workerID == "" {
		workerID = ProcessingWorkerID
	}
	_, err := pool.Exec(ctx, `
		INSERT INTO worker_heartbeats (worker_id, last_seen_at, updated_at)
		VALUES ($1, now(), now())
		ON CONFLICT (worker_id) DO UPDATE
		SET last_seen_at = now(), updated_at = now()`, workerID)
	return err
}

// WorkerProbe is the /readyz worker + queue snapshot (no driver detail in ErrMsg).
type WorkerProbe struct {
	OK           bool
	WorkerCheck  string // ok | missing | stale | fail | unavailable
	QueueCheck   string // ok | fail | unavailable
	ErrMsg       string // short stable code for clients
	Reason       string // optional operator remediation (safe, no secrets)
	PendingJobs  int64
	LastSeenAgeS int64 // seconds since last heartbeat; -1 when missing
}

// HeartbeatQuerier is satisfied by *pgxpool.Pool (and test stubs).
type HeartbeatQuerier interface {
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}

// ProbeWorkerReadiness checks the processing worker heartbeat and pending queue depth.
func ProbeWorkerReadiness(ctx context.Context, q HeartbeatQuerier, staleAfter time.Duration) WorkerProbe {
	out := WorkerProbe{
		WorkerCheck:  "unavailable",
		QueueCheck:   "unavailable",
		LastSeenAgeS: -1,
	}
	if q == nil {
		out.ErrMsg = "worker probe unavailable"
		out.Reason = "Database pool unavailable; cannot probe worker heartbeat."
		return out
	}
	if staleAfter <= 0 {
		staleAfter = DefaultHeartbeatStaleAfter
	}

	var pending int64
	if err := q.QueryRow(ctx, `
		SELECT COUNT(*)::bigint FROM processing_jobs WHERE status = 'pending'`).Scan(&pending); err != nil {
		out.WorkerCheck = "fail"
		out.QueueCheck = "fail"
		out.ErrMsg = "queue depth query failed"
		out.Reason = "Could not read processing job queue depth; check DATABASE_URL and Postgres."
		return out
	}
	out.PendingJobs = pending
	out.QueueCheck = "ok"

	var lastSeen time.Time
	err := q.QueryRow(ctx, `
		SELECT last_seen_at FROM worker_heartbeats WHERE worker_id = $1`, ProcessingWorkerID).Scan(&lastSeen)
	if errors.Is(err, pgx.ErrNoRows) {
		out.WorkerCheck = "missing"
		out.ErrMsg = "worker heartbeat missing"
		out.Reason = "No processing worker heartbeat. API-only readiness 503 is expected — start the worker (npm run dev includes it, or npm run dev:worker)."
		return out
	}
	if err != nil {
		out.WorkerCheck = "fail"
		out.ErrMsg = "worker heartbeat query failed"
		out.Reason = "Could not read worker_heartbeats; ensure goose migration 039_worker_heartbeats is applied."
		return out
	}

	age := time.Since(lastSeen)
	if age < 0 {
		age = 0
	}
	out.LastSeenAgeS = int64(age / time.Second)
	if age > staleAfter {
		out.WorkerCheck = "stale"
		out.ErrMsg = "worker heartbeat stale"
		out.Reason = fmt.Sprintf(
			"Processing worker heartbeat older than %s. API-only readiness 503 is expected — start or restart the worker (npm run dev includes it, or npm run dev:worker).",
			staleAfter,
		)
		return out
	}

	out.OK = true
	out.WorkerCheck = "ok"
	return out
}