| @@ -164,32 +164,33 @@ func NewScheduler(store FeedStore, logger *slog.Logger, tickInterval, staleInter |
| 164 | } | 164 | } |
| 165 | | 165 | |
| 166 | func (s *Scheduler) Run(ctx context.Context) error { | 166 | func (s *Scheduler) Run(ctx context.Context) error { |
| | 167 | + s.logger.Info("starting initial feed refresh") |
| | 168 | + s.fetchAll(ctx, 0) |
| | 169 | + |
| 167 | ticker := time.NewTicker(s.tickInterval) | 170 | ticker := time.NewTicker(s.tickInterval) |
| 168 | defer ticker.Stop() | 171 | defer ticker.Stop() |
| 169 | | 172 | |
| 170 | - s.fetchAll(ctx) | | |
| 171 | - | | |
| 172 | for { | 173 | for { |
| 173 | select { | 174 | select { |
| 174 | case <-ctx.Done(): | 175 | case <-ctx.Done(): |
| 175 | return ctx.Err() | 176 | return ctx.Err() |
| 176 | case <-ticker.C: | 177 | case <-ticker.C: |
| 177 | - s.fetchAll(ctx) | 178 | + s.fetchAll(ctx, s.staleInterval) |
| 178 | } | 179 | } |
| 179 | } | 180 | } |
| 180 | } | 181 | } |
| 181 | | 182 | |
| 182 | -func (s *Scheduler) fetchAll(ctx context.Context) { | 183 | +func (s *Scheduler) fetchAll(ctx context.Context, olderThan time.Duration) { |
| 183 | - feeds, err := s.store.GetFeedsToFetch(ctx, s.staleInterval, 1_000) | 184 | + feeds, err := s.store.GetFeedsToFetch(ctx, olderThan, 1_000) |
| 184 | if err != nil { | 185 | if err != nil { |
| 185 | s.logger.Error("failed to get feeds", "error", err) | 186 | s.logger.Error("failed to get feeds", "error", err) |
| 186 | return | 187 | return |
| 187 | } | 188 | } |
| 188 | | 189 | |
| 189 | - s.logger.Info("feeds fetched", "count", len(feeds)) | 190 | + s.logger.Info("fetching feeds", "count", len(feeds), "older_than", olderThan) |
| 190 | | 191 | |
| 191 | g, gCtx := errgroup.WithContext(ctx) | 192 | g, gCtx := errgroup.WithContext(ctx) |
| 192 | - g.SetLimit(20) | 193 | + g.SetLimit(50) |
| 193 | for _, f := range feeds { | 194 | for _, f := range feeds { |
| 194 | g.Go(func() error { | 195 | g.Go(func() error { |
| 195 | s.FetchFeed(gCtx, f) | 196 | s.FetchFeed(gCtx, f) |