package library import ( "context" "errors" "log/slog" "time" "github.com/google/uuid" "pipu/api/internal/platform/apperr" ) type ScheduleStore interface { ListSources(context.Context) ([]Source, error) SetSourceStatus(context.Context, uuid.UUID, string, *string) error } type sourceConnection interface { connect(context.Context, Source) (Source, error) check(context.Context, Source, bool) (string, *string) } type scheduledScan interface { startScan(Source, bool) error } // Scheduler owns the periodic lifetime, not source or template business rules. type Scheduler struct { store ScheduleStore sources sourceConnection scans scheduledScan log *slog.Logger } // Restore при старте api переподключает NFS-хранилища и проверяет local (после перезапуска агента). func (s *Scheduler) Restore(ctx context.Context) { sources, err := s.store.ListSources(ctx) if err != nil { s.log.Error("restore sources", "err", err) return } for _, src := range sources { if src.Status == "disabled" { continue } if _, err := s.sources.connect(ctx, src); err != nil { s.log.Error("restore source", "source_id", src.ID, "err", err) } } } // RunSchedule раз в минуту проверяет доступность хранилищ и запускает сканы по расписанию. func (s *Scheduler) RunSchedule(ctx context.Context) { t := time.NewTicker(time.Minute) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: } sources, err := s.store.ListSources(ctx) if err != nil { s.log.Warn("schedule: list sources", "err", err) continue } for _, src := range sources { if src.Status == "disabled" { continue } s.healthCheck(ctx, src) if src.ScanInterval == nil || src.Status != "mounted" { continue } if src.LastScanAt == nil || time.Since(*src.LastScanAt) >= *src.ScanInterval { if err := s.scans.startScan(src, false); err != nil && !errors.Is(err, apperr.Conflict(CodeScanRunning, "")) { s.log.Warn("scheduled scan", "source_id", src.ID, "err", err) } } } } } // healthCheck: недоступное хранилище — status=error (треки серые), вернулось — снова mounted. func (s *Scheduler) healthCheck(ctx context.Context, src Source) { ctx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() type result struct { status string message *string } done := make(chan result, 1) go func() { status, message := s.sources.check(ctx, src, false) done <- result{status, message} }() var status string var message *string select { case r := <-done: status, message = r.status, r.message case <-ctx.Done(): m := "storage does not respond" status, message = "error", &m } if status == "error" && src.Type == SourceNFS && src.Status == "error" { // Попробовать перемонтировать, как при старте. if _, err := s.sources.connect(ctx, src); err == nil { return } } if status != src.Status || deref(message) != deref(src.StatusMessage) { if err := s.store.SetSourceStatus(ctx, src.ID, status, message); err != nil { s.log.Warn("health check", "source_id", src.ID, "err", err) } } }