310 lines
8.9 KiB
Go
310 lines
8.9 KiB
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 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)
|
||
|
|
}
|
||
|
|
}
|