core/internal/history/worker.go

package history

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"time"
)

const (
	workerProcessRetryAttempts  = 3
	workerCompleteRetryAttempts = 5
	workerStaleRunningAfter     = 30 * time.Minute
)

func (s *Service) StartWorker(ctx context.Context) {
	if s == nil || s.db == nil {
		return
	}
	if os.Getenv("HISTORY_WORKER_DISABLED") == "1" {
		log.Printf("history worker disabled via HISTORY_WORKER_DISABLED=1")
		return
	}
	go s.workerLoop(ctx)
}

func (s *Service) workerLoop(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}
		job, ok, err := s.claimNextJob(ctx)
		if err != nil {
			log.Printf("history worker claim failed err=%v", err)
			select {
			case <-ctx.Done():
				return
			case <-time.After(300 * time.Millisecond):
			}
			continue
		}
		if !ok {
			select {
			case <-ctx.Done():
				return
			case <-time.After(150 * time.Millisecond):
			}
			continue
		}
		log.Printf("history worker claimed job_id=%s type=%s entity=%s:%s", job.ID, job.JobType, job.EntityType, job.EntityID)
		started := time.Now()
		result, procErr := s.processJobWithRetry(ctx, job)
		duration := time.Since(started)
		if procErr != nil {
			log.Printf("history worker job failed job_id=%s type=%s entity=%s:%s duration_ms=%d err=%v", job.ID, job.JobType, job.EntityType, job.EntityID, duration.Milliseconds(), procErr)
			if isTransientDBErr(procErr) {
				if rqErr := s.requeueJobTransient(ctx, job.ID, procErr); rqErr != nil {
					log.Printf("history worker transient requeue failed job_id=%s err=%v", job.ID, rqErr)
				} else {
					log.Printf("history worker transient requeued job_id=%s", job.ID)
				}
				select {
				case <-ctx.Done():
					return
				case <-time.After(750 * time.Millisecond):
				}
				continue
			}
		} else {
			log.Printf("history worker job completed job_id=%s type=%s entity=%s:%s duration_ms=%d", job.ID, job.JobType, job.EntityType, job.EntityID, duration.Milliseconds())
		}
		if completeErr := s.completeJobWithRetry(ctx, job.ID, result, procErr); completeErr != nil {
			log.Printf("history worker complete failed job_id=%s err=%v", job.ID, completeErr)
		}
	}
}

func (s *Service) requeueJobTransient(ctx context.Context, jobID string, procErr error) error {
	now := time.Now().UTC()
	_, err := s.db.ExecContext(ctx, `
		UPDATE history_recompute_jobs
		SET status = 'queued',
		    started_at = NULL,
		    locked_at = NULL,
		    worker_id = NULL,
		    finished_at = NULL,
		    error = ?
		WHERE id = ?`,
		fmt.Sprintf("transient db error at %s: %v", now.Format(time.RFC3339), procErr), jobID,
	)
	return err
}

func (s *Service) claimNextJob(ctx context.Context) (JobRecord, bool, error) {
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
	if err != nil {
		return JobRecord{}, false, err
	}
	defer func() { _ = tx.Rollback() }()
	now := time.Now().UTC()

	staleBefore := now.Add(-workerStaleRunningAfter)
	if _, err := tx.ExecContext(ctx, `
		UPDATE history_recompute_jobs
		SET status = 'queued',
		    started_at = NULL,
		    locked_at = NULL,
		    worker_id = NULL,
		    error = CONCAT('auto-requeued stale running job at ', ?, '; previous error: ', COALESCE(error, 'none'))
		WHERE status = 'running'
		  AND (
		    (locked_at IS NOT NULL AND locked_at < ?)
		    OR (locked_at IS NULL AND started_at IS NOT NULL AND started_at < ?)
		  )`,
		now.Format(time.RFC3339), staleBefore, staleBefore,
	); err != nil {
		return JobRecord{}, false, err
	}

	var jobID string
	err = tx.QueryRowContext(ctx, `
		SELECT id
		FROM history_recompute_jobs
		WHERE status = 'queued'
		ORDER BY created_at, id
		LIMIT 1
		FOR UPDATE`).Scan(&jobID)
	if err == sql.ErrNoRows {
		return JobRecord{}, false, nil
	}
	if err != nil {
		return JobRecord{}, false, err
	}
	if _, err := tx.ExecContext(ctx, `
		UPDATE history_recompute_jobs
		SET status = 'running', started_at = ?, locked_at = ?, attempt_count = attempt_count + 1, worker_id = ?
		WHERE id = ?`,
		now, now, "local-worker", jobID,
	); err != nil {
		return JobRecord{}, false, err
	}
	if err := tx.Commit(); err != nil {
		return JobRecord{}, false, err
	}
	job, err := s.GetJob(ctx, jobID)
	if err != nil {
		return JobRecord{}, false, err
	}
	return job, true, nil
}

func (s *Service) completeJob(ctx context.Context, jobID string, result map[string]any, procErr error) error {
	var resultJSON any
	if result != nil {
		b, err := json.Marshal(result)
		if err != nil {
			return err
		}
		resultJSON = b
	}
	now := time.Now().UTC()
	if procErr != nil {
		_, err := s.db.ExecContext(ctx, `
			UPDATE history_recompute_jobs
			SET status = 'failed', finished_at = ?, error = ?, result = ?
			WHERE id = ?`,
			now, procErr.Error(), resultJSON, jobID,
		)
		return err
	}
	_, err := s.db.ExecContext(ctx, `
		UPDATE history_recompute_jobs
		SET status = 'completed', finished_at = ?, error = NULL, result = ?
		WHERE id = ?`,
		now, resultJSON, jobID,
	)
	return err
}

func (s *Service) processJob(ctx context.Context, job JobRecord) (map[string]any, error) {
	switch job.JobType {
	case "recompute":
		return s.processRecomputeJob(ctx, job)
	case "delete_event_recompute":
		return s.processDeleteEventRecompute(ctx, job)
	case "rollback":
		return s.processRollbackJob(ctx, job)
	case "hard_restore":
		return s.processHardRestoreJob(ctx, job)
	default:
		return nil, fmt.Errorf("unsupported job type %q", job.JobType)
	}
}

func (s *Service) processJobWithRetry(ctx context.Context, job JobRecord) (map[string]any, error) {
	var (
		result map[string]any
		err    error
	)
	for attempt := 0; attempt < workerProcessRetryAttempts; attempt++ {
		result, err = s.processJob(ctx, job)
		if err == nil {
			return result, nil
		}
		if !isTransientDBErr(err) || attempt == workerProcessRetryAttempts-1 {
			return nil, err
		}
		time.Sleep(time.Duration(120*(attempt+1)) * time.Millisecond)
	}
	return result, err
}

func (s *Service) completeJobWithRetry(ctx context.Context, jobID string, result map[string]any, procErr error) error {
	var err error
	for attempt := 0; attempt < workerCompleteRetryAttempts; attempt++ {
		err = s.completeJob(ctx, jobID, result, procErr)
		if err == nil {
			return nil
		}
		if !isTransientDBErr(err) || attempt == workerCompleteRetryAttempts-1 {
			return err
		}
		time.Sleep(time.Duration(120*(attempt+1)) * time.Millisecond)
	}
	return err
}