package worker

import (
	"sync/atomic"
	"time"

	"git.gozeros.com/wsk/eightbf-analysis-go/internal/domain"
	"git.gozeros.com/wsk/eightbf-analysis-go/internal/logging"
	"git.gozeros.com/wsk/eightbf-analysis-go/internal/repo"
)

// LiveWorker 定时扫 queued，最多并发 MaxInflight 场交给 LiveChainRunner。
type LiveWorker struct {
	Repo        *repo.JobRepo
	Runner      *LiveChainRunner
	PollEvery   time.Duration
	MaxInflight int32
	inflight    int32
}

func (w *LiveWorker) Start() {
	if w.PollEvery <= 0 {
		w.PollEvery = time.Second
	}
	go func() {
		t := time.NewTicker(w.PollEvery)
		defer t.Stop()
		for range t.C {
			w.poll()
		}
	}()
}

func (w *LiveWorker) poll() {
	max := w.MaxInflight
	if max < 1 {
		max = 2
	}
	if atomic.LoadInt32(&w.inflight) >= max {
		return
	}
	queued, err := w.Repo.FindQueuedJobs()
	if err != nil {
		logging.L.Warnw("LiveWorker 扫队列失败", "err", err)
		return
	}
	for _, job := range queued {
		if atomic.LoadInt32(&w.inflight) >= max {
			break
		}
		job.Status = domain.StatusRunning
		job.Phase = "claimed"
		if err := w.Repo.UpdateProgress(job); err != nil {
			logging.L.Warnw("LiveWorker 占坑失败", logging.KeyJobID, job.ID, "err", err)
			continue
		}
		atomic.AddInt32(&w.inflight, 1)
		jobID := job.ID
		go func() {
			defer atomic.AddInt32(&w.inflight, -1)
			fresh, err := w.Repo.FindByID(jobID)
			if err != nil || fresh == nil {
				return
			}
			logging.L.Infow("开始真模型会商",
				logging.KeyJobID, jobID,
				logging.KeyRequestID, fresh.RequestID,
				logging.KeyTaskID, fresh.TaskID,
				logging.KeyModel, fresh.Model,
				logging.KeyStatus, fresh.Status,
			)
			w.Runner.Run(fresh)
		}()
	}
}