// Package orchestrator coordinates the catalogue walk, metadata extraction, // chapter-list fetching, and parallel chapter scraping. // // Concurrency model // - One goroutine runs ScrapeCatalogue and feeds book URLs into a channel. // - For each book, a dedicated goroutine calls ScrapeMetadata (metadata goroutine). // - ScrapeChapterList is called in the metadata goroutine once metadata is done. // - N worker goroutines (default: runtime.NumCPU()) each pull ChapterRef values // from a shared work queue and call ScrapeChapterText. // - A sync.WaitGroup ensures all chapter workers finish before the orchestrator // signals completion. package orchestrator import ( "context" "fmt" "log/slog" "runtime" "sync" "github.com/libnovel/scraper/internal/scraper" "github.com/libnovel/scraper/internal/writer" ) // Config holds tunable parameters for the orchestrator. type Config struct { // Workers is the number of goroutines used to scrape chapters in parallel. // Defaults to runtime.NumCPU() when 0. Workers int // StaticRoot is the path to the static/books output directory. StaticRoot string // SingleBookURL when non-empty causes the orchestrator to scrape only // that one book instead of walking the full catalogue. SingleBookURL string } // Orchestrator coordinates the full scrape pipeline. type Orchestrator struct { cfg Config novel scraper.NovelScraper writer *writer.Writer log *slog.Logger workers int } // New returns a new Orchestrator. func New(cfg Config, novel scraper.NovelScraper, log *slog.Logger) *Orchestrator { workers := cfg.Workers if workers <= 0 { workers = runtime.NumCPU() } return &Orchestrator{ cfg: cfg, novel: novel, writer: writer.New(cfg.StaticRoot), log: log, workers: workers, } } // Run executes the full scrape pipeline and blocks until it is complete or ctx // is cancelled. func (o *Orchestrator) Run(ctx context.Context) error { o.log.Info("orchestrator starting", "source", o.novel.SourceName(), "workers", o.workers, "static_root", o.cfg.StaticRoot, ) // chapterWork is the shared queue consumed by chapter worker goroutines. type chapterJob struct { slug string ref scraper.ChapterRef } chapterWork := make(chan chapterJob, o.workers*4) // Start chapter worker pool. var chapterWG sync.WaitGroup for i := 0; i < o.workers; i++ { chapterWG.Add(1) go func(workerID int) { defer chapterWG.Done() for job := range chapterWork { select { case <-ctx.Done(): return default: } // Skip if already on disk. if o.writer.ChapterExists(job.slug, job.ref) { o.log.Debug("chapter already exists, skipping", "book", job.slug, "chapter", job.ref.Number) continue } chapter, err := o.novel.ScrapeChapterText(ctx, job.ref) if err != nil { o.log.Error("chapter scrape failed", "book", job.slug, "chapter", job.ref.Number, "url", job.ref.URL, "err", err, ) continue } if err := o.writer.WriteChapter(job.slug, chapter); err != nil { o.log.Error("chapter write failed", "book", job.slug, "chapter", job.ref.Number, "err", err, ) continue } o.log.Info("chapter saved", "book", job.slug, "chapter", job.ref.Number, "worker", workerID, ) } }(i) } // processBook scrapes metadata + chapter list for one book, then enqueues // chapter jobs. It is called inside a goroutine per book. processBook := func(bookURL string) { // Metadata goroutine. meta, err := o.novel.ScrapeMetadata(ctx, bookURL) if err != nil { o.log.Error("metadata scrape failed", "url", bookURL, "err", err) return } // Persist / update metadata.yaml. if err := o.writer.WriteMetadata(meta); err != nil { o.log.Error("metadata write failed", "slug", meta.Slug, "err", err) // Continue — chapters can still be scraped. } 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) return } o.log.Info("chapter list fetched", "slug", meta.Slug, "chapters", len(refs)) // Enqueue chapter jobs. for _, ref := range refs { select { case <-ctx.Done(): return case chapterWork <- chapterJob{slug: meta.Slug, ref: ref}: } } } if o.cfg.SingleBookURL != "" { // Single-book mode: skip catalogue entirely. o.log.Info("single-book mode", "url", o.cfg.SingleBookURL) processBook(o.cfg.SingleBookURL) } else { // Catalogue mode: stream every book. entries, catErrs := o.novel.ScrapeCatalogue(ctx) // Drain catalogue errors in a separate goroutine. go func() { for err := range catErrs { o.log.Error("catalogue error", "err", err) } }() var bookWG sync.WaitGroup for entry := range entries { select { case <-ctx.Done(): break default: } bookWG.Add(1) bookURL := entry.URL go func() { defer bookWG.Done() processBook(bookURL) }() } // Wait for all book goroutines to enqueue their chapters before // closing the chapter work queue. bookWG.Wait() } // Signal chapter workers there is no more work. close(chapterWork) // Wait for all in-flight chapter scrapes to finish. chapterWG.Wait() if ctx.Err() != nil { return fmt.Errorf("orchestrator: context cancelled: %w", ctx.Err()) } o.log.Info("orchestrator finished") return nil }