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

package processing

import (
	"context"
	"os"
	"testing"
	"time"

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

func TestCleanupStuckResetsJobsAndProducts(t *testing.T) {
	dsn := os.Getenv("DATABASE_URL")
	if dsn == "" {
		t.Skip("DATABASE_URL not set")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()

	pg, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer pg.Close()

	var companyID uuid.UUID
	err = pg.QueryRow(ctx, `
		SELECT company_id
		FROM raw_products
		WHERE company_id IS NOT NULL
		ORDER BY updated_at DESC
		LIMIT 1`).Scan(&companyID)
	if errorsIsNoRows(err) {
		t.Skip("no raw_products rows available")
	}
	if err != nil {
		t.Fatal(err)
	}

	var userID uuid.UUID
	err = pg.QueryRow(ctx, `SELECT id FROM users ORDER BY created_at DESC LIMIT 1`).Scan(&userID)
	if errorsIsNoRows(err) {
		t.Skip("no users rows available")
	}
	if err != nil {
		t.Fatal(err)
	}

	var rawID uuid.UUID
	err = pg.QueryRow(ctx, `
		SELECT id
		FROM raw_products
		WHERE company_id = $1
		ORDER BY updated_at DESC
		LIMIT 1`, companyID).Scan(&rawID)
	if errorsIsNoRows(err) {
		t.Skip("no raw_products for selected company")
	}
	if err != nil {
		t.Fatal(err)
	}

	var jobID uuid.UUID
	err = pg.QueryRow(ctx, `
		INSERT INTO processing_jobs (
			company_id, user_id, status, total_products, processed_products,
			processing_type, started_at, updated_at
		) VALUES ($1, $2, 'running', 1, 0, 'full', now() - interval '3 hours', now() - interval '3 hours')
		RETURNING id`,
		companyID, userID,
	).Scan(&jobID)
	if err != nil {
		t.Fatal(err)
	}
	defer func() {
		_, _ = pg.Exec(context.Background(), `DELETE FROM processing_job_products WHERE job_id = $1`, jobID)
		_, _ = pg.Exec(context.Background(), `DELETE FROM processing_jobs WHERE id = $1`, jobID)
	}()

	var freshJobID uuid.UUID
	err = pg.QueryRow(ctx, `
		INSERT INTO processing_jobs (
			company_id, user_id, status, total_products, processed_products,
			processing_type, started_at, updated_at
		) VALUES ($1, $2, 'running', 1, 0, 'full', now(), now())
		RETURNING id`,
		companyID, userID,
	).Scan(&freshJobID)
	if err != nil {
		t.Fatal(err)
	}
	defer func() {
		_, _ = pg.Exec(context.Background(), `DELETE FROM processing_job_products WHERE job_id = $1`, freshJobID)
		_, _ = pg.Exec(context.Background(), `DELETE FROM processing_jobs WHERE id = $1`, freshJobID)
	}()

	if _, err := pg.Exec(ctx, `
		INSERT INTO processing_job_products (job_id, raw_product_id, status, updated_at)
		VALUES ($1, $2, 'processing', now() - interval '3 hours')`, jobID, rawID); err != nil {
		t.Fatal(err)
	}
	if _, err := pg.Exec(ctx, `
		INSERT INTO processing_job_products (job_id, raw_product_id, status, updated_at)
		VALUES ($1, $2, 'processing', now())`, freshJobID, rawID); err != nil {
		t.Fatal(err)
	}

	res, err := CleanupStuck(ctx, pg)
	if err != nil {
		t.Fatal(err)
	}
	if res.JobsMarkedFailed < 1 {
		t.Fatalf("jobs_marked_failed=%d want >= 1", res.JobsMarkedFailed)
	}
	if res.ProductsReset < 1 {
		t.Fatalf("products_reset=%d want >= 1", res.ProductsReset)
	}

	var stuckJobStatus, stuckProductStatus string
	if err := pg.QueryRow(ctx, `SELECT status FROM processing_jobs WHERE id = $1`, jobID).Scan(&stuckJobStatus); err != nil {
		t.Fatal(err)
	}
	if stuckJobStatus != "failed" {
		t.Fatalf("stuck job status=%q want failed", stuckJobStatus)
	}
	if err := pg.QueryRow(ctx, `
		SELECT status FROM processing_job_products WHERE job_id = $1`, jobID).Scan(&stuckProductStatus); err != nil {
		t.Fatal(err)
	}
	if stuckProductStatus != "pending" {
		t.Fatalf("stuck product status=%q want pending", stuckProductStatus)
	}

	var freshJobStatus, freshProductStatus string
	if err := pg.QueryRow(ctx, `SELECT status FROM processing_jobs WHERE id = $1`, freshJobID).Scan(&freshJobStatus); err != nil {
		t.Fatal(err)
	}
	if freshJobStatus != "running" {
		t.Fatalf("fresh job status=%q want running", freshJobStatus)
	}
	if err := pg.QueryRow(ctx, `
		SELECT status FROM processing_job_products WHERE job_id = $1`, freshJobID).Scan(&freshProductStatus); err != nil {
		t.Fatal(err)
	}
	if freshProductStatus != "processing" {
		t.Fatalf("fresh product status=%q want processing", freshProductStatus)
	}

	var syncFeedID uuid.UUID
	err = pg.QueryRow(ctx, `
		INSERT INTO input_feeds (company_id, name, url, feed_type, status, sync_interval_minutes, options)
		VALUES ($1, 'stuck-sync-feed', 'https://example.com/stuck.csv', 'csv', 'active', 60, '{}'::jsonb)
		RETURNING id`, companyID).Scan(&syncFeedID)
	if err != nil {
		t.Fatal(err)
	}
	defer func() {
		_, _ = pg.Exec(context.Background(), `DELETE FROM feed_sync_jobs WHERE feed_id = $1`, syncFeedID)
		_, _ = pg.Exec(context.Background(), `DELETE FROM input_feeds WHERE id = $1`, syncFeedID)
	}()

	var stuckSyncID, freshSyncID uuid.UUID
	err = pg.QueryRow(ctx, `
		INSERT INTO feed_sync_jobs (feed_id, company_id, status, started_at, updated_at)
		VALUES ($1, $2, 'running', now() - interval '3 hours', now() - interval '3 hours')
		RETURNING id`, syncFeedID, companyID).Scan(&stuckSyncID)
	if err != nil {
		t.Fatal(err)
	}
	err = pg.QueryRow(ctx, `
		INSERT INTO feed_sync_jobs (feed_id, company_id, status, started_at, updated_at)
		VALUES ($1, $2, 'running', now(), now())
		RETURNING id`, syncFeedID, companyID).Scan(&freshSyncID)
	if err != nil {
		t.Fatal(err)
	}

	res2, err := CleanupStuck(ctx, pg)
	if err != nil {
		t.Fatal(err)
	}
	if res2.SyncJobsMarkedFailed < 1 {
		t.Fatalf("sync_jobs_marked_failed=%d want >= 1", res2.SyncJobsMarkedFailed)
	}

	var stuckSyncStatus, freshSyncStatus string
	if err := pg.QueryRow(ctx, `SELECT status FROM feed_sync_jobs WHERE id = $1`, stuckSyncID).Scan(&stuckSyncStatus); err != nil {
		t.Fatal(err)
	}
	if stuckSyncStatus != "failed" {
		t.Fatalf("stuck sync status=%q want failed", stuckSyncStatus)
	}
	if err := pg.QueryRow(ctx, `SELECT status FROM feed_sync_jobs WHERE id = $1`, freshSyncID).Scan(&freshSyncStatus); err != nil {
		t.Fatal(err)
	}
	if freshSyncStatus != "running" {
		t.Fatalf("fresh sync status=%q want running", freshSyncStatus)
	}
}

func TestReclaimOrphanedRunningRequeuesImmediately(t *testing.T) {
	dsn := os.Getenv("DATABASE_URL")
	if dsn == "" {
		t.Skip("DATABASE_URL not set")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()

	pg, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer pg.Close()

	var companyID uuid.UUID
	err = pg.QueryRow(ctx, `
		SELECT company_id
		FROM raw_products
		WHERE company_id IS NOT NULL
		ORDER BY updated_at DESC
		LIMIT 1`).Scan(&companyID)
	if errorsIsNoRows(err) {
		t.Skip("no raw_products rows available")
	}
	if err != nil {
		t.Fatal(err)
	}

	var userID uuid.UUID
	err = pg.QueryRow(ctx, `SELECT id FROM users ORDER BY created_at DESC LIMIT 1`).Scan(&userID)
	if errorsIsNoRows(err) {
		t.Skip("no users rows available")
	}
	if err != nil {
		t.Fatal(err)
	}

	var rawID uuid.UUID
	err = pg.QueryRow(ctx, `
		SELECT id
		FROM raw_products
		WHERE company_id = $1
		ORDER BY updated_at DESC
		LIMIT 1`, companyID).Scan(&rawID)
	if errorsIsNoRows(err) {
		t.Skip("no raw_products for selected company")
	}
	if err != nil {
		t.Fatal(err)
	}

	var jobID uuid.UUID
	err = pg.QueryRow(ctx, `
		INSERT INTO processing_jobs (
			company_id, user_id, status, total_products, processed_products,
			processing_type, started_at, updated_at
		) VALUES ($1, $2, 'running', 1, 0, 'full', now(), now())
		RETURNING id`,
		companyID, userID,
	).Scan(&jobID)
	if err != nil {
		t.Fatal(err)
	}
	defer func() {
		_, _ = pg.Exec(context.Background(), `DELETE FROM processing_job_products WHERE job_id = $1`, jobID)
		_, _ = pg.Exec(context.Background(), `DELETE FROM processing_jobs WHERE id = $1`, jobID)
	}()

	if _, err := pg.Exec(ctx, `
		INSERT INTO processing_job_products (job_id, raw_product_id, status, updated_at)
		VALUES ($1, $2, 'processing', now())`, jobID, rawID); err != nil {
		t.Fatal(err)
	}

	var syncFeedID uuid.UUID
	err = pg.QueryRow(ctx, `
		INSERT INTO input_feeds (company_id, name, url, feed_type, status, sync_interval_minutes, options)
		VALUES ($1, 'orphan-reclaim-feed', 'https://example.com/orphan.csv', 'csv', 'active', 60, '{}'::jsonb)
		RETURNING id`, companyID).Scan(&syncFeedID)
	if err != nil {
		t.Fatal(err)
	}
	defer func() {
		_, _ = pg.Exec(context.Background(), `DELETE FROM feed_sync_jobs WHERE feed_id = $1`, syncFeedID)
		_, _ = pg.Exec(context.Background(), `DELETE FROM input_feeds WHERE id = $1`, syncFeedID)
	}()

	var syncID uuid.UUID
	err = pg.QueryRow(ctx, `
		INSERT INTO feed_sync_jobs (feed_id, company_id, status, started_at, updated_at)
		VALUES ($1, $2, 'running', now(), now())
		RETURNING id`, syncFeedID, companyID).Scan(&syncID)
	if err != nil {
		t.Fatal(err)
	}

	res, err := ReclaimOrphanedRunning(ctx, pg)
	if err != nil {
		t.Fatal(err)
	}
	if res.JobsRequeued < 1 {
		t.Fatalf("jobs_requeued=%d want >= 1", res.JobsRequeued)
	}
	if res.ProductsReset < 1 {
		t.Fatalf("products_reset=%d want >= 1", res.ProductsReset)
	}
	if res.SyncJobsRequeued < 1 {
		t.Fatalf("sync_requeued=%d want >= 1", res.SyncJobsRequeued)
	}

	var jobStatus, productStatus, syncStatus string
	if err := pg.QueryRow(ctx, `SELECT status FROM processing_jobs WHERE id = $1`, jobID).Scan(&jobStatus); err != nil {
		t.Fatal(err)
	}
	if jobStatus != "pending" {
		t.Fatalf("job status=%q want pending", jobStatus)
	}
	if err := pg.QueryRow(ctx, `SELECT status FROM processing_job_products WHERE job_id = $1`, jobID).Scan(&productStatus); err != nil {
		t.Fatal(err)
	}
	if productStatus != "pending" {
		t.Fatalf("product status=%q want pending", productStatus)
	}
	if err := pg.QueryRow(ctx, `SELECT status FROM feed_sync_jobs WHERE id = $1`, syncID).Scan(&syncStatus); err != nil {
		t.Fatal(err)
	}
	if syncStatus != "pending" {
		t.Fatalf("sync status=%q want pending", syncStatus)
	}
}

func TestLoadPendingItemsReclaimsAgedProcessing(t *testing.T) {
	dsn := os.Getenv("DATABASE_URL")
	if dsn == "" {
		t.Skip("DATABASE_URL not set")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()

	pg, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer pg.Close()

	var companyID uuid.UUID
	err = pg.QueryRow(ctx, `
		SELECT company_id
		FROM raw_products
		WHERE company_id IS NOT NULL
		ORDER BY updated_at DESC
		LIMIT 1`).Scan(&companyID)
	if errorsIsNoRows(err) {
		t.Skip("no raw_products rows available")
	}
	if err != nil {
		t.Fatal(err)
	}

	var userID uuid.UUID
	err = pg.QueryRow(ctx, `SELECT id FROM users ORDER BY created_at DESC LIMIT 1`).Scan(&userID)
	if errorsIsNoRows(err) {
		t.Skip("no users rows available")
	}
	if err != nil {
		t.Fatal(err)
	}

	var rawAged, rawFresh uuid.UUID
	err = pg.QueryRow(ctx, `
		SELECT id FROM raw_products WHERE company_id = $1 ORDER BY updated_at DESC LIMIT 1`, companyID).Scan(&rawAged)
	if errorsIsNoRows(err) {
		t.Skip("no raw_products for selected company")
	}
	if err != nil {
		t.Fatal(err)
	}
	err = pg.QueryRow(ctx, `
		SELECT id FROM raw_products WHERE company_id = $1 AND id <> $2 ORDER BY updated_at DESC LIMIT 1`,
		companyID, rawAged).Scan(&rawFresh)
	if errorsIsNoRows(err) {
		t.Skip("need two distinct raw_products for reclaim vs fresh")
	}
	if err != nil {
		t.Fatal(err)
	}

	var jobID uuid.UUID
	err = pg.QueryRow(ctx, `
		INSERT INTO processing_jobs (
			company_id, user_id, status, total_products, processed_products,
			processing_type, started_at, updated_at
		) VALUES ($1, $2, 'running', 2, 0, 'full', now(), now())
		RETURNING id`,
		companyID, userID,
	).Scan(&jobID)
	if err != nil {
		t.Fatal(err)
	}
	defer func() {
		_, _ = pg.Exec(context.Background(), `DELETE FROM processing_job_products WHERE job_id = $1`, jobID)
		_, _ = pg.Exec(context.Background(), `DELETE FROM processing_jobs WHERE id = $1`, jobID)
	}()

	if _, err := pg.Exec(ctx, `
		INSERT INTO processing_job_products (job_id, raw_product_id, status, updated_at)
		VALUES ($1, $2, 'processing', now() - interval '3 hours')`, jobID, rawAged); err != nil {
		t.Fatal(err)
	}
	if _, err := pg.Exec(ctx, `
		INSERT INTO processing_job_products (job_id, raw_product_id, status, updated_at)
		VALUES ($1, $2, 'processing', now())`, jobID, rawFresh); err != nil {
		t.Fatal(err)
	}

	p := &Pipeline{Pool: pg}
	items, err := p.loadPendingItems(ctx, jobID, 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(items) != 1 {
		t.Fatalf("reclaimed=%d want 1 (aged only)", len(items))
	}
	if items[0].RawID != rawAged {
		t.Fatalf("reclaimed raw=%s want aged=%s", items[0].RawID, rawAged)
	}

	var freshStatus string
	if err := pg.QueryRow(ctx, `
		SELECT status FROM processing_job_products
		WHERE job_id = $1 AND raw_product_id = $2`, jobID, rawFresh).Scan(&freshStatus); err != nil {
		t.Fatal(err)
	}
	if freshStatus != "processing" {
		t.Fatalf("fresh status=%q want processing (not reclaimed)", freshStatus)
	}
}