feat(orchestrator): add Progress type and OnProgress callback

Adds atomic counters for books_found, chapters_scraped, chapters_skipped,
and errors. The new OnProgress callback fires after each counter update
and once more at the end of Run, giving callers a live progress feed.
This commit is contained in:
Admin
2026-03-04 00:40:24 +05:00
parent 041099598b
commit 1b234754e8

View File

@@ -17,11 +17,20 @@ import (
"log/slog" "log/slog"
"runtime" "runtime"
"sync" "sync"
"sync/atomic"
"github.com/libnovel/scraper/internal/scraper" "github.com/libnovel/scraper/internal/scraper"
"github.com/libnovel/scraper/internal/storage" "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. // Config holds tunable parameters for the orchestrator.
type Config struct { type Config struct {
// Workers is the number of goroutines used to scrape chapters in parallel. // 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 // SingleBookURL when non-empty causes the orchestrator to scrape only
// that one book instead of walking the full catalogue. // that one book instead of walking the full catalogue.
SingleBookURL string 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. // Orchestrator coordinates the full scrape pipeline.
@@ -69,6 +82,29 @@ func (o *Orchestrator) Run(ctx context.Context) error {
"workers", o.workers, "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. // chapterWork is the shared queue consumed by chapter worker goroutines.
type chapterJob struct { type chapterJob struct {
slug string slug string
@@ -93,6 +129,8 @@ func (o *Orchestrator) Run(ctx context.Context) error {
if o.store.ChapterExists(ctx, job.slug, job.ref) { if o.store.ChapterExists(ctx, job.slug, job.ref) {
o.log.Debug("chapter already exists, skipping", o.log.Debug("chapter already exists, skipping",
"book", job.slug, "chapter", job.ref.Number) "book", job.slug, "chapter", job.ref.Number)
chaptersSkipped.Add(1)
notify()
continue continue
} }
@@ -104,6 +142,8 @@ func (o *Orchestrator) Run(ctx context.Context) error {
"url", job.ref.URL, "url", job.ref.URL,
"err", err, "err", err,
) )
errors.Add(1)
notify()
continue continue
} }
@@ -113,9 +153,13 @@ func (o *Orchestrator) Run(ctx context.Context) error {
"chapter", job.ref.Number, "chapter", job.ref.Number,
"err", err, "err", err,
) )
errors.Add(1)
notify()
continue continue
} }
chaptersScraped.Add(1)
notify()
o.log.Info("chapter saved", o.log.Info("chapter saved",
"book", job.slug, "book", job.slug,
"chapter", job.ref.Number, "chapter", job.ref.Number,
@@ -132,6 +176,8 @@ func (o *Orchestrator) Run(ctx context.Context) error {
meta, err := o.novel.ScrapeMetadata(ctx, bookURL) meta, err := o.novel.ScrapeMetadata(ctx, bookURL)
if err != nil { if err != nil {
o.log.Error("metadata scrape failed", "url", bookURL, "err", err) o.log.Error("metadata scrape failed", "url", bookURL, "err", err)
errors.Add(1)
notify()
return return
} }
@@ -141,12 +187,16 @@ func (o *Orchestrator) Run(ctx context.Context) error {
// Continue — chapters can still be scraped. // Continue — chapters can still be scraped.
} }
booksFound.Add(1)
notify()
o.log.Info("metadata saved", "slug", meta.Slug, "title", meta.Title) o.log.Info("metadata saved", "slug", meta.Slug, "title", meta.Title)
// Fetch chapter list. // Fetch chapter list.
refs, err := o.novel.ScrapeChapterList(ctx, bookURL) refs, err := o.novel.ScrapeChapterList(ctx, bookURL)
if err != nil { if err != nil {
o.log.Error("chapter list scrape failed", "slug", meta.Slug, "err", err) o.log.Error("chapter list scrape failed", "slug", meta.Slug, "err", err)
errors.Add(1)
notify()
return return
} }
@@ -174,6 +224,8 @@ func (o *Orchestrator) Run(ctx context.Context) error {
go func() { go func() {
for err := range catErrs { for err := range catErrs {
o.log.Error("catalogue error", "err", err) 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. // Wait for all in-flight chapter scrapes to finish.
chapterWG.Wait() chapterWG.Wait()
// Final progress notification.
notify()
if ctx.Err() != nil { if ctx.Err() != nil {
return fmt.Errorf("orchestrator: context cancelled: %w", ctx.Err()) return fmt.Errorf("orchestrator: context cancelled: %w", ctx.Err())
} }