feat: externalize filter configuration and add global structured logging and job reprocessing functionality

internal/db/tasks.go

@@ -24,24 +24,21 @@ func (db *DB) EnqueueTask(taskType string, payload interface{}) error {
 
 // ClaimNextTask atomically finds and claims the next pending task
 func (db *DB) ClaimNextTask() (*Task, error) {
-	tx, err := db.Begin()
-	if err != nil {
-		return nil, err
-	}
-	defer func() { _ = tx.Rollback() }()
-
-	// Find the oldest PENDING task that is ready to be retried
 	query := `
-	SELECT id, task_type, payload, status, error_message, retry_count, next_retry_at, created_at, updated_at
-	FROM tasks
-	WHERE status = 'PENDING' AND next_retry_at <= CURRENT_TIMESTAMP
-	ORDER BY created_at ASC
-	LIMIT 1
+	UPDATE tasks 
+	SET status = 'RUNNING', updated_at = CURRENT_TIMESTAMP 
+	WHERE id = (
+		SELECT id FROM tasks 
+		WHERE status = 'PENDING' AND next_retry_at <= CURRENT_TIMESTAMP 
+		ORDER BY created_at ASC 
+		LIMIT 1
+	)
+	RETURNING id, task_type, payload, status, error_message, retry_count, next_retry_at, created_at, updated_at
 	`
 	
 	var task Task
 	var errMsg sql.NullString
-	err = tx.QueryRow(query).Scan(
+	err := db.QueryRow(query).Scan(
 		&task.ID, &task.TaskType, &task.Payload, &task.Status, &errMsg,
 		&task.RetryCount, &task.NextRetryAt, &task.CreatedAt, &task.UpdatedAt,
 	)
@@ -57,17 +54,6 @@ func (db *DB) ClaimNextTask() (*Task, error) {
 		task.ErrorMessage = errMsg.String
 	}
 
-	// Claim the task
-	updateQuery := `UPDATE tasks SET status = 'RUNNING', updated_at = CURRENT_TIMESTAMP WHERE id = ?`
-	_, err = tx.Exec(updateQuery, task.ID)
-	if err != nil {
-		return nil, err
-	}
-
-	if err := tx.Commit(); err != nil {
-		return nil, err
-	}
-
 	return &task, nil
 }
 
@@ -157,3 +143,49 @@ func (db *DB) GetQueueStatus() (map[string]int, error) {
 	}
 	return counts, nil
 }
+
+// ReprocessAllJobs wipes all job analysis results and re-queues PROCESS tasks for all existing raw jobs.
+func (db *DB) ReprocessAllJobs() error {
+	tx, err := db.Begin()
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	// 1. Delete all rows from job_analyses
+	if _, err := tx.Exec(`DELETE FROM job_analyses`); err != nil {
+		return fmt.Errorf("failed to clear job_analyses: %w", err)
+	}
+
+	// 2. Delete all existing PROCESS and ANALYZE tasks
+	if _, err := tx.Exec(`DELETE FROM tasks WHERE task_type IN ('PROCESS', 'ANALYZE')`); err != nil {
+		return fmt.Errorf("failed to clear old tasks: %w", err)
+	}
+
+	// 3. Queue up new PROCESS tasks for all raw jobs
+	rows, err := tx.Query(`SELECT id FROM raw_jobs`)
+	if err != nil {
+		return fmt.Errorf("failed to get raw jobs: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var rawJobIDs []int64
+	for rows.Next() {
+		var rawJobID int64
+		if err := rows.Scan(&rawJobID); err != nil {
+			return err
+		}
+		rawJobIDs = append(rawJobIDs, rawJobID)
+	}
+	rows.Close() // Explicitly close to release any read locks
+
+	// 4. Enqueue PROCESS task for each job
+	for _, rawJobID := range rawJobIDs {
+		payloadJSON := fmt.Sprintf(`{"raw_job_id":%d}`, rawJobID)
+		if _, err := tx.Exec(`INSERT INTO tasks (task_type, payload, status) VALUES ('PROCESS', ?, 'PENDING')`, payloadJSON); err != nil {
+			return fmt.Errorf("failed to enqueue process task for job %d: %w", rawJobID, err)
+		}
+	}
+
+	return tx.Commit()
+}