thread-master/backend/internal/model/job/usecase/maintenance.go

package usecase

import (
	"context"

	"haixun-backend/internal/library/clock"
	"haixun-backend/internal/model/job/domain/enum"
	domusecase "haixun-backend/internal/model/job/domain/usecase"
)

func (u *jobUseCase) RunMaintenance(ctx context.Context) (domusecase.MaintenanceResult, error) {
	result := domusecase.MaintenanceResult{}

	pending, err := u.runs.FindPendingDue(ctx, clock.NowUnixNano(), 50)
	if err != nil {
		return result, err
	}
	for _, run := range pending {
		if _, err := u.enqueueRun(ctx, run); err == nil {
			result.EnqueuedPending++
		}
	}

	now := clock.NowUnixNano()
	cancelRuns, err := u.runs.FindCancelRequestedBefore(ctx, now, 50)
	if err != nil {
		return result, err
	}
	for _, run := range cancelRuns {
		template, err := u.templates.FindByType(ctx, run.TemplateType)
		if err != nil {
			continue
		}
		grace := clock.SecondsToNanos(template.CancelPolicy.GraceSeconds)
		if run.CancelRequestedAt == nil || now-*run.CancelRequestedAt < grace {
			continue
		}
		jobID := run.ID.Hex()
		fromStatus := string(run.Status)
		run.Status = enum.RunStatusCancelled
		run.CompletedAt = &now
		run.Progress.Summary = "cancelled after grace timeout"
		updated, err := u.runs.Update(ctx, run)
		if err != nil {
			continue
		}
		_ = u.queue.ReleaseLock(ctx, jobID, run.LockedBy)
		_ = u.queue.ClearCancelSignal(ctx, jobID)
		u.releaseDedupe(ctx, updated)
		_ = u.appendEvent(ctx, jobID, "status_changed", fromStatus, string(enum.RunStatusCancelled), "cancel grace timeout", nil)
		result.ReapedCancelGrace++
	}

	stuckRuns, err := u.runs.FindQueuedWithLock(ctx, 50)
	if err != nil {
		return result, err
	}
	for _, run := range stuckRuns {
		jobID := run.ID.Hex()
		if run.StartedAt != nil && run.LockedUntil != nil && *run.LockedUntil > now {
			fromStatus := string(run.Status)
			patch := *run
			patch.Status = enum.RunStatusRunning
			if patch.Progress.Summary == "" || patch.Progress.Summary == "waiting for worker" {
				patch.Progress.Summary = "worker claimed job (status repaired)"
			}
			updated, err := u.runs.UpdateIfStatus(ctx, &patch, []enum.RunStatus{
				enum.RunStatusPending,
				enum.RunStatusQueued,
			})
			if err != nil {
				continue
			}
			_ = u.appendEvent(ctx, jobID, "status_changed", fromStatus, string(enum.RunStatusRunning), "repaired queued job with active worker lock", nil)
			result.RepairedStuckClaims++
			_ = updated
			continue
		}
		fromStatus := string(run.Status)
		workerID := run.LockedBy
		patch := *run
		patch.LockedBy = ""
		patch.LockedUntil = nil
		patch.StartedAt = nil
		patch.Progress.Summary = "waiting for worker"
		patch.Progress.Percentage = 0
		if _, err := u.runs.UpdateIfStatus(ctx, &patch, []enum.RunStatus{
			enum.RunStatusPending,
			enum.RunStatusQueued,
		}); err != nil {
			continue
		}
		_ = u.queue.ReleaseLock(ctx, jobID, workerID)
		if err := u.queue.Enqueue(ctx, patch.WorkerType, jobID); err != nil {
			continue
		}
		_ = u.appendEvent(ctx, jobID, "status_changed", fromStatus, string(enum.RunStatusQueued), "requeued orphaned lock", nil)
		result.ReapedOrphanedLocks++
	}

	expiredRuns, err := u.runs.FindRunningTimedOut(ctx, now, 50)
	if err != nil {
		return result, err
	}
	for _, run := range expiredRuns {
		jobID := run.ID.Hex()
		fromStatus := string(run.Status)
		run.Status = enum.RunStatusExpired
		run.CompletedAt = &now
		run.Progress.Summary = "job lock expired"
		updated, err := u.runs.Update(ctx, run)
		if err != nil {
			continue
		}
		_ = u.queue.ReleaseLock(ctx, jobID, run.LockedBy)
		_ = u.queue.ClearCancelSignal(ctx, jobID)
		u.releaseDedupe(ctx, updated)
		_ = u.appendEvent(ctx, jobID, "status_changed", fromStatus, string(enum.RunStatusExpired), "job lock expired", nil)
		result.ReapedExpiredLocks++
	}

	return result, nil
}