88 lines
2.8 KiB
Go
88 lines
2.8 KiB
Go
|
package processing
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"fmt"
|
||
|
|
|
||
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
|
)
|
||
|
|
|
||
|
|
// RetentionAgeInterval is the shared SQL age used by CleanupExpired /
|
||
|
|
// CleanupExpiredSyncJobs to prune terminal job history.
|
||
|
|
const RetentionAgeInterval = "30 days"
|
||
|
|
|
||
|
|
// RetentionBatchLimit caps rows deleted per cleanup call to avoid long locks.
|
||
|
|
const RetentionBatchLimit = 5000
|
||
|
|
|
||
|
|
// RetentionCleanupResult counts rows deleted by retention cleanups.
|
||
|
|
type RetentionCleanupResult struct {
|
||
|
|
JobsDeleted int64
|
||
|
|
SyncJobsDeleted int64
|
||
|
|
}
|
||
|
|
|
||
|
|
// CleanupExpired deletes terminal processing_jobs older than RetentionAgeInterval.
|
||
|
|
// Child processing_job_products rows are removed via ON DELETE CASCADE.
|
||
|
|
// Pending/running jobs are never deleted.
|
||
|
|
// Migrated history (ai_provider_mode='migrated') is retained indefinitely.
|
||
|
|
func CleanupExpired(ctx context.Context, pool *pgxpool.Pool) (RetentionCleanupResult, error) {
|
||
|
|
var out RetentionCleanupResult
|
||
|
|
if pool == nil {
|
||
|
|
return out, fmt.Errorf("cleanup expired: nil pool")
|
||
|
|
}
|
||
|
|
|
||
|
|
ct, err := pool.Exec(ctx, `
|
||
|
|
WITH doomed AS (
|
||
|
|
SELECT id
|
||
|
|
FROM processing_jobs
|
||
|
|
WHERE status IN ('completed', 'failed', 'cancelled')
|
||
|
|
AND COALESCE(ai_provider_mode, '') <> 'migrated'
|
||
|
|
AND COALESCE(completed_at, updated_at) < now() - interval '`+RetentionAgeInterval+`'
|
||
|
|
ORDER BY COALESCE(completed_at, updated_at) ASC
|
||
|
|
LIMIT $1
|
||
|
|
)
|
||
|
|
DELETE FROM processing_jobs
|
||
|
|
WHERE id IN (SELECT id FROM doomed)`, RetentionBatchLimit)
|
||
|
|
if err != nil {
|
||
|
|
return out, fmt.Errorf("cleanup expired jobs: %w", err)
|
||
|
|
}
|
||
|
|
out.JobsDeleted = ct.RowsAffected()
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// CleanupExpiredSyncJobs deletes terminal feed_sync_jobs older than RetentionAgeInterval.
|
||
|
|
// Keeps the newest completed job that still has a content_hash per feed so
|
||
|
|
// lastContentHash skip-unchanged continues to work after cleanup.
|
||
|
|
// raw_products.sync_job_id is ON DELETE SET NULL, so product rows are preserved.
|
||
|
|
func CleanupExpiredSyncJobs(ctx context.Context, pool *pgxpool.Pool) (RetentionCleanupResult, error) {
|
||
|
|
var out RetentionCleanupResult
|
||
|
|
if pool == nil {
|
||
|
|
return out, fmt.Errorf("cleanup expired sync jobs: nil pool")
|
||
|
|
}
|
||
|
|
|
||
|
|
ct, err := pool.Exec(ctx, `
|
||
|
|
WITH keep AS (
|
||
|
|
SELECT DISTINCT ON (feed_id) id
|
||
|
|
FROM feed_sync_jobs
|
||
|
|
WHERE status = 'completed'
|
||
|
|
AND content_hash IS NOT NULL
|
||
|
|
AND content_hash <> ''
|
||
|
|
ORDER BY feed_id, completed_at DESC NULLS LAST
|
||
|
|
),
|
||
|
|
doomed AS (
|
||
|
|
SELECT id
|
||
|
|
FROM feed_sync_jobs
|
||
|
|
WHERE status IN ('completed', 'failed')
|
||
|
|
AND COALESCE(completed_at, updated_at) < now() - interval '`+RetentionAgeInterval+`'
|
||
|
|
AND id NOT IN (SELECT id FROM keep)
|
||
|
|
ORDER BY COALESCE(completed_at, updated_at) ASC
|
||
|
|
LIMIT $1
|
||
|
|
)
|
||
|
|
DELETE FROM feed_sync_jobs
|
||
|
|
WHERE id IN (SELECT id FROM doomed)`, RetentionBatchLimit)
|
||
|
|
if err != nil {
|
||
|
|
return out, fmt.Errorf("cleanup expired sync jobs: %w", err)
|
||
|
|
}
|
||
|
|
out.SyncJobsDeleted = ct.RowsAffected()
|
||
|
|
return out, nil
|
||
|
|
}
|