kandev/apps/backend/internal/github/poller_schedule.go

package github

import (
	"context"
	"sort"
	"strconv"
	"time"

	"github.com/kandev/kandev/internal/common/logger"
	"go.uber.org/zap"
)

const (
	searchFastPollInterval     = 1 * time.Minute
	searchIdlePollInterval     = 15 * time.Minute
	searchVeryIdlePollInterval = 30 * time.Minute
	searchIdleThreshold        = 2 * time.Hour
	searchVeryIdleThreshold    = 24 * time.Hour
)

// PRWatchTaskActivity is the narrow task-domain projection used to decide
// whether a searching PR watch should be checked. It intentionally excludes
// generic task/watch UpdatedAt values because provider synchronization can
// write those fields.
type PRWatchTaskActivity struct {
	LastActivityAt time.Time
	Running        bool
}

// TaskActivityProvider supplies persisted activity for a bulk set of tasks.
// Implementations must use task-local evidence only; the poller never asks a
// provider or remote Git host to classify idleness.
type TaskActivityProvider interface {
	LoadPRWatchTaskActivity(ctx context.Context, taskIDs []string) (map[string]PRWatchTaskActivity, error)
}

func (p *Poller) selectDuePRWatches(ctx context.Context, watches []*PRWatch) []*PRWatch {
	return selectDuePRWatches(ctx, watches, p.taskActivityProvider, p.now(), p.logger)
}

func selectDuePRWatches(
	ctx context.Context,
	watches []*PRWatch,
	provider TaskActivityProvider,
	now time.Time,
	log *logger.Logger,
) []*PRWatch {
	if len(watches) == 0 {
		return nil
	}
	activityByTask := loadPRWatchTaskActivities(ctx, watches, provider, log)
	groupDue := duePRWatchGroups(now, watches, activityByTask)

	due := make([]*PRWatch, 0, len(watches))
	for _, watch := range watches {
		if watch == nil {
			continue
		}
		if groupDue[prWatchDiscoveryTargetKey(watch)] {
			due = append(due, watch)
		}
	}
	return due
}

func loadPRWatchTaskActivities(
	ctx context.Context,
	watches []*PRWatch,
	provider TaskActivityProvider,
	log *logger.Logger,
) map[string]PRWatchTaskActivity {
	if provider == nil {
		return nil
	}
	taskIDs := prWatchActivityTaskIDs(watches)
	if len(taskIDs) == 0 {
		return nil
	}
	loaded, err := provider.LoadPRWatchTaskActivity(ctx, taskIDs)
	if err != nil {
		if log != nil {
			log.Debug("failed to load PR watch task activity", zap.Error(err))
		}
		return nil
	}
	return loaded
}

func prWatchActivityTaskIDs(watches []*PRWatch) []string {
	seen := make(map[string]struct{}, len(watches))
	taskIDs := make([]string, 0, len(watches))
	for _, watch := range watches {
		if watch == nil || watch.PRNumber != 0 || watch.TaskID == "" {
			continue
		}
		if _, ok := seen[watch.TaskID]; ok {
			continue
		}
		seen[watch.TaskID] = struct{}{}
		taskIDs = append(taskIDs, watch.TaskID)
	}
	sort.Strings(taskIDs)
	return taskIDs
}

func duePRWatchGroups(
	now time.Time,
	watches []*PRWatch,
	activityByTask map[string]PRWatchTaskActivity,
) map[string]bool {
	groupDue := make(map[string]bool, len(watches))
	for _, watch := range watches {
		markDuePRWatchGroup(groupDue, now, watch, activityByTask)
	}
	return groupDue
}

func markDuePRWatchGroup(
	groupDue map[string]bool,
	now time.Time,
	watch *PRWatch,
	activityByTask map[string]PRWatchTaskActivity,
) {
	if watch == nil {
		return
	}
	key := prWatchDiscoveryTargetKey(watch)
	if watch.PRNumber != 0 {
		groupDue[key] = true
		return
	}
	activity := activityByTask[watch.TaskID]
	if prWatchIsDue(now, watch, activity) {
		groupDue[key] = true
	}
}

func prWatchSearchInterval(now time.Time, watch *PRWatch, activity PRWatchTaskActivity) time.Duration {
	if watch == nil || watch.PRNumber != 0 {
		return defaultPRPollInterval
	}
	if activity.Running || activity.LastActivityAt.IsZero() {
		return searchFastPollInterval
	}
	activityAt := activity.LastActivityAt
	if watch != nil && watch.CreatedAt.After(activityAt) {
		activityAt = watch.CreatedAt
	}
	idleFor := now.Sub(activityAt)
	if idleFor < searchIdleThreshold {
		return searchFastPollInterval
	}
	if idleFor < searchVeryIdleThreshold {
		return searchIdlePollInterval
	}
	return searchVeryIdlePollInterval
}

func prWatchIsDue(now time.Time, watch *PRWatch, activity PRWatchTaskActivity) bool {
	if watch == nil {
		return false
	}
	if watch.LastCheckedAt == nil {
		return true
	}
	return now.Sub(*watch.LastCheckedAt) >= prWatchSearchInterval(now, watch, activity)
}

func prWatchDiscoveryTargetKey(watch *PRWatch) string {
	if watch == nil {
		return ""
	}
	mode := "search"
	if watch.PRNumber != 0 {
		mode = "pr:" + strconv.Itoa(watch.PRNumber)
	}
	return watch.WorkspaceID + "\x00" + mode + "\x00" + watch.Owner + "\x00" + watch.Repo + "\x00" + watch.Branch
}

func (p *Poller) now() time.Time {
	if p != nil && p.clock != nil {
		return p.clock().UTC()
	}
	return time.Now().UTC()
}

func (s *Service) now() time.Time {
	if s == nil {
		return time.Now().UTC()
	}
	s.clockMu.RLock()
	clock := s.clock
	s.clockMu.RUnlock()
	if clock != nil {
		return clock().UTC()
	}
	return time.Now().UTC()
}