descrybe/apps/api/internal/processing/stuck_cleanup.go

package processing

import (
	"context"
	"fmt"

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

// StuckAgeInterval is the shared SQL age used by CleanupStuck and claim-time
// reclaim in loadPendingItems for stranded processing_job_products.
const StuckAgeInterval = "2 hours"

// StuckCleanupResult counts rows touched by CleanupStuck.
type StuckCleanupResult struct {
	JobsMarkedFailed     int64
	ProductsReset        int64
	SyncJobsMarkedFailed int64
}

// OrphanReclaimResult counts rows touched by ReclaimOrphanedRunning.
type OrphanReclaimResult struct {
	JobsRequeued     int64
	ProductsReset    int64
	SyncJobsRequeued int64
}

// ReclaimOrphanedRunning resets in-flight work left by a dead worker process so
// ClaimNext / feed sync claim can pick it up again. Call on every worker process
// start (before TouchHeartbeat) under the single-worker architecture — a live
// heartbeat from a just-killed process must not skip reclaim. Never call while
// another live worker may own the rows.
//
// Unlike CleanupStuck (age-gated fail), this requeues immediately: running → pending
// for jobs/sync jobs, and processing → pending for job products without a result.
func ReclaimOrphanedRunning(ctx context.Context, pool *pgxpool.Pool) (OrphanReclaimResult, error) {
	var out OrphanReclaimResult
	if pool == nil {
		return out, fmt.Errorf("reclaim orphaned running: nil pool")
	}

	ctJobs, err := pool.Exec(ctx, `
		UPDATE processing_jobs
		SET status = 'pending', started_at = NULL, error = NULL, updated_at = now()
		WHERE status = 'running'`)
	if err != nil {
		return out, fmt.Errorf("reclaim orphaned jobs: %w", err)
	}
	out.JobsRequeued = ctJobs.RowsAffected()

	ctProd, err := pool.Exec(ctx, `
		UPDATE processing_job_products
		SET status = 'pending', error = NULL, updated_at = now()
		WHERE status = 'processing' AND processed_product_id IS NULL`)
	if err != nil {
		return out, fmt.Errorf("reclaim orphaned products: %w", err)
	}
	out.ProductsReset = ctProd.RowsAffected()

	ctSync, err := pool.Exec(ctx, `
		UPDATE feed_sync_jobs
		SET status = 'pending', started_at = NULL, error = NULL, completed_at = NULL, updated_at = now()
		WHERE status = 'running'`)
	if err != nil {
		return out, fmt.Errorf("reclaim orphaned sync jobs: %w", err)
	}
	out.SyncJobsRequeued = ctSync.RowsAffected()

	return out, nil
}

// CleanupStuck aligns worker and admin stuck-cleanup semantics:
// mark long-running jobs failed, reset stranded processing_job_products
// from 'processing' back to 'pending', and fail aged running feed_sync_jobs
// so they are not left unclaimable forever.
func CleanupStuck(ctx context.Context, pool *pgxpool.Pool) (StuckCleanupResult, error) {
	var out StuckCleanupResult
	if pool == nil {
		return out, fmt.Errorf("cleanup stuck: nil pool")
	}

	ctJobs, err := pool.Exec(ctx, `
		UPDATE processing_jobs
		SET status = 'failed', error = 'stuck cleanup', completed_at = now(), updated_at = now()
		WHERE status = 'running' AND updated_at < now() - interval '`+StuckAgeInterval+`'`)
	if err != nil {
		return out, fmt.Errorf("cleanup stuck jobs: %w", err)
	}
	out.JobsMarkedFailed = ctJobs.RowsAffected()

	ctProd, err := pool.Exec(ctx, `
		UPDATE processing_job_products
		SET status = 'pending', updated_at = now()
		WHERE status = 'processing' AND updated_at < now() - interval '`+StuckAgeInterval+`'`)
	if err != nil {
		return out, fmt.Errorf("cleanup stuck products: %w", err)
	}
	out.ProductsReset = ctProd.RowsAffected()

	ctSync, err := pool.Exec(ctx, `
		UPDATE feed_sync_jobs
		SET status = 'failed', error = 'stuck cleanup', completed_at = now(), updated_at = now()
		WHERE status = 'running' AND updated_at < now() - interval '`+StuckAgeInterval+`'`)
	if err != nil {
		return out, fmt.Errorf("cleanup stuck sync jobs: %w", err)
	}
	out.SyncJobsMarkedFailed = ctSync.RowsAffected()

	return out, nil
}