diff --git a/scraper/internal/orchestrator/orchestrator.go b/scraper/internal/orchestrator/orchestrator.go index 619c1b7..3440e3f 100644 --- a/scraper/internal/orchestrator/orchestrator.go +++ b/scraper/internal/orchestrator/orchestrator.go @@ -17,11 +17,20 @@ import ( "log/slog" "runtime" "sync" + "sync/atomic" "github.com/libnovel/scraper/internal/scraper" "github.com/libnovel/scraper/internal/storage" ) +// Progress is a snapshot of counters at a point in time. +type Progress struct { + BooksFound int + ChaptersScraped int + ChaptersSkipped int + Errors int +} + // Config holds tunable parameters for the orchestrator. type Config struct { // Workers is the number of goroutines used to scrape chapters in parallel. @@ -35,6 +44,10 @@ type Config struct { // SingleBookURL when non-empty causes the orchestrator to scrape only // that one book instead of walking the full catalogue. SingleBookURL string + + // OnProgress is called periodically with the current progress counters. + // It is always called on completion (success or failure). May be nil. + OnProgress func(p Progress) } // Orchestrator coordinates the full scrape pipeline. @@ -69,6 +82,29 @@ func (o *Orchestrator) Run(ctx context.Context) error { "workers", o.workers, ) + // Atomic counters updated by concurrent goroutines. + var ( + booksFound atomic.Int64 + chaptersScraped atomic.Int64 + chaptersSkipped atomic.Int64 + errors atomic.Int64 + ) + + snapshot := func() Progress { + return Progress{ + BooksFound: int(booksFound.Load()), + ChaptersScraped: int(chaptersScraped.Load()), + ChaptersSkipped: int(chaptersSkipped.Load()), + Errors: int(errors.Load()), + } + } + + notify := func() { + if o.cfg.OnProgress != nil { + o.cfg.OnProgress(snapshot()) + } + } + // chapterWork is the shared queue consumed by chapter worker goroutines. type chapterJob struct { slug string @@ -93,6 +129,8 @@ func (o *Orchestrator) Run(ctx context.Context) error { if o.store.ChapterExists(ctx, job.slug, job.ref) { o.log.Debug("chapter already exists, skipping", "book", job.slug, "chapter", job.ref.Number) + chaptersSkipped.Add(1) + notify() continue } @@ -104,6 +142,8 @@ func (o *Orchestrator) Run(ctx context.Context) error { "url", job.ref.URL, "err", err, ) + errors.Add(1) + notify() continue } @@ -113,9 +153,13 @@ func (o *Orchestrator) Run(ctx context.Context) error { "chapter", job.ref.Number, "err", err, ) + errors.Add(1) + notify() continue } + chaptersScraped.Add(1) + notify() o.log.Info("chapter saved", "book", job.slug, "chapter", job.ref.Number, @@ -132,6 +176,8 @@ func (o *Orchestrator) Run(ctx context.Context) error { meta, err := o.novel.ScrapeMetadata(ctx, bookURL) if err != nil { o.log.Error("metadata scrape failed", "url", bookURL, "err", err) + errors.Add(1) + notify() return } @@ -141,12 +187,16 @@ func (o *Orchestrator) Run(ctx context.Context) error { // Continue — chapters can still be scraped. } + booksFound.Add(1) + notify() o.log.Info("metadata saved", "slug", meta.Slug, "title", meta.Title) // Fetch chapter list. refs, err := o.novel.ScrapeChapterList(ctx, bookURL) if err != nil { o.log.Error("chapter list scrape failed", "slug", meta.Slug, "err", err) + errors.Add(1) + notify() return } @@ -174,6 +224,8 @@ func (o *Orchestrator) Run(ctx context.Context) error { go func() { for err := range catErrs { o.log.Error("catalogue error", "err", err) + errors.Add(1) + notify() } }() @@ -204,6 +256,9 @@ func (o *Orchestrator) Run(ctx context.Context) error { // Wait for all in-flight chapter scrapes to finish. chapterWG.Wait() + // Final progress notification. + notify() + if ctx.Err() != nil { return fmt.Errorf("orchestrator: context cancelled: %w", ctx.Err()) }