thread-master/apps/backend/internal/module/radar/usecase/sweep_schedule.go

package usecase

import (
	"context"
	"fmt"
	"strings"
	"time"

	"apps/backend/internal/module/radar/domain"
)

// DailySweepHourUTC is the fixed daily schedule (spec §4.2：UTC 22:00 ＝台北 06:00)。
const DailySweepHourUTC = 22

// SweepJobScheduler schedules one radar_sweep job per watch per UTC day.
// Implemented by a thin adapter over job usecase.Service (see worker / service_context).
type SweepJobScheduler interface {
	ScheduleRadarSweep(ctx context.Context, ownerUID int64, watchID string, runAt int64) (jobID string, err error)
}

// ManualSweepJobScheduler is optional. TriggerSweep uses it so 「立即巡邏」
// is not swallowed by a finished same-day daily job.
type ManualSweepJobScheduler interface {
	ScheduleManualRadarSweep(ctx context.Context, ownerUID int64, watchID string, runAt int64) (jobID string, err error)
}

// SweepJobSchedulerFunc adapts a function to SweepJobScheduler.
type SweepJobSchedulerFunc func(ctx context.Context, ownerUID int64, watchID string, runAt int64) (jobID string, err error)

func (f SweepJobSchedulerFunc) ScheduleRadarSweep(ctx context.Context, ownerUID int64, watchID string, runAt int64) (string, error) {
	return f(ctx, ownerUID, watchID, runAt)
}

// ScheduleDailySweeps 為每個 active watch 補齊「今天（台北）已到期」的時段 Job。
//
// 預設時段是台北 06:00（等同舊的 UTC 22:00）。會員可在 /radar/schedule 多選時段。
// 呼叫端必須已持有 worker maintenance Redis lock。回傳建立（或已存在）的 slot 數。
func (s *Service) ScheduleDailySweeps(ctx context.Context, now time.Time) (int, error) {
	if s.SweepJobs == nil {
		return 0, fmt.Errorf("%w: sweep job scheduler not configured", domain.ErrNotReady)
	}
	if now.IsZero() {
		now = time.Now()
	}

	watches, err := s.Repo.ListAllActiveWatches(ctx)
	if err != nil {
		return 0, err
	}
	hoursByOwner := map[int64][]int{}
	n := 0
	for _, w := range watches {
		if w == nil || w.Status != domain.WatchActive {
			continue
		}
		hours, ok := hoursByOwner[w.OwnerUID]
		if !ok {
			sch, gerr := s.GetRadarSchedule(ctx, w.OwnerUID)
			if gerr != nil {
				return n, fmt.Errorf("schedule owner %d: %w", w.OwnerUID, gerr)
			}
			hours = sch.Hours
			hoursByOwner[w.OwnerUID] = hours
		}
		slots := domain.DueSweepSlots(now, hours)
		for _, slot := range slots {
			if _, err := s.SweepJobs.ScheduleRadarSweep(ctx, w.OwnerUID, w.ID, slot.RunAt); err != nil {
				return n, fmt.Errorf("schedule watch %s owner %d hour %d: %w", w.ID, w.OwnerUID, slot.Hour, err)
			}
			n++
		}
	}
	return n, nil
}

// PastDailySweepSlot reports whether the default Taipei 06:00 slot has started.
func PastDailySweepSlot(now time.Time) bool {
	return len(domain.DueSweepSlots(now, domain.DefaultSweepHours())) > 0
}

// DailySweepRunAt returns unix-ns for today's default Taipei 06:00 slot (or next if not yet).
func DailySweepRunAt(now time.Time) int64 {
	slots := domain.DueSweepSlots(now, domain.DefaultSweepHours())
	if len(slots) > 0 {
		return slots[0].RunAt
	}
	loc := domain.TaipeiLocation()
	local := now.In(loc)
	slot := time.Date(local.Year(), local.Month(), local.Day(), domain.DefaultSweepHour, 0, 0, 0, loc)
	return slot.UnixNano()
}

// BeginSweepRecord creates the RadarSweep shell for a claimed job (T527).
// Fetch / judge (T528–T530) attach progress onto the same record via UpdateSweep.
func (s *Service) BeginSweepRecord(ctx context.Context, ownerUID int64, watchID, jobID, path string) (*domain.RadarSweep, error) {
	if ownerUID <= 0 {
		return nil, fmt.Errorf("%w: owner_uid required", domain.ErrValidation)
	}
	// watchID may be empty for on-demand explore (no subscription).
	if path == "" {
		path = domain.SweepPathAPI
	}
	sw := &domain.RadarSweep{
		ID:        domain.NewID(),
		OwnerUID:  ownerUID,
		WatchID:   strings.TrimSpace(watchID),
		JobID:     jobID,
		Path:      path,
		StartedAt: domain.NowNano(),
	}
	if err := sw.Normalize(); err != nil {
		return nil, err
	}
	if err := s.Repo.CreateSweep(ctx, sw); err != nil {
		return nil, err
	}
	return sw, nil
}