// Package orchestrator coordinates metadata extraction, chapter-list fetching, // and parallel chapter scraping for a single book. // // Design: // - RunBook scrapes one book (metadata + chapter list + chapter texts) end-to-end. // - N worker goroutines pull chapter refs from a shared queue and call ScrapeChapterText. // - The caller (runner poll loop) owns the outer task-claim / finish cycle. // - An optional PostMetadata hook (set in Config) is called after WriteMetadata // succeeds. The runner uses this to upsert books into Meilisearch. package orchestrator import ( "context" "fmt" "log/slog" "runtime" "sync" "sync/atomic" "github.com/libnovel/backend/internal/bookstore" "github.com/libnovel/backend/internal/domain" "github.com/libnovel/backend/internal/scraper" ) // 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 // PostMetadata is an optional hook called with the scraped BookMeta after // WriteMetadata succeeds. Errors from the hook are logged but not fatal. // Used by the runner to index books in Meilisearch. PostMetadata func(ctx context.Context, meta domain.BookMeta) } // Orchestrator runs a single-book scrape pipeline. type Orchestrator struct { novel scraper.NovelScraper store bookstore.BookWriter log *slog.Logger workers int postMetadata func(ctx context.Context, meta domain.BookMeta) } // New returns a new Orchestrator. func New(cfg Config, novel scraper.NovelScraper, store bookstore.BookWriter, log *slog.Logger) *Orchestrator { if log == nil { log = slog.Default() } workers := cfg.Workers if workers <= 0 { workers = runtime.NumCPU() } return &Orchestrator{ novel: novel, store: store, log: log, workers: workers, postMetadata: cfg.PostMetadata, } } // RunBook scrapes a single book described by task. It handles: // 1. Metadata scrape + write // 2. Chapter list scrape + write // 3. Parallel chapter text scrape + write (worker pool) // // Returns a ScrapeResult with counters. The result's ErrorMessage is non-empty // if the run failed at the metadata or chapter-list level. func (o *Orchestrator) RunBook(ctx context.Context, task domain.ScrapeTask) domain.ScrapeResult { o.log.Debug("orchestrator: RunBook starting", "task_id", task.ID, "kind", task.Kind, "url", task.TargetURL, "workers", o.workers, ) var result domain.ScrapeResult if task.TargetURL == "" { result.ErrorMessage = "task has no target URL" return result } // ── Step 1: Metadata ────────────────────────────────────────────────────── meta, err := o.novel.ScrapeMetadata(ctx, task.TargetURL) if err != nil { o.log.Error("metadata scrape failed", "url", task.TargetURL, "err", err) result.ErrorMessage = fmt.Sprintf("metadata: %v", err) result.Errors++ return result } result.Slug = meta.Slug if err := o.store.WriteMetadata(ctx, meta); err != nil { o.log.Error("metadata write failed", "slug", meta.Slug, "err", err) // non-fatal: continue to chapters result.Errors++ } else { result.BooksFound = 1 result.Slug = meta.Slug // Fire optional post-metadata hook (e.g. Meilisearch indexing). if o.postMetadata != nil { o.postMetadata(ctx, meta) } } o.log.Debug("metadata saved", "slug", meta.Slug, "title", meta.Title) // ── Step 2: Chapter list ────────────────────────────────────────────────── refs, err := o.novel.ScrapeChapterList(ctx, task.TargetURL, task.ToChapter) if err != nil { o.log.Error("chapter list scrape failed", "slug", meta.Slug, "err", err) result.ErrorMessage = fmt.Sprintf("chapter list: %v", err) result.Errors++ return result } o.log.Debug("chapter list fetched", "slug", meta.Slug, "chapters", len(refs)) // Persist chapter refs (without text) so the index exists early. if wErr := o.store.WriteChapterRefs(ctx, meta.Slug, refs); wErr != nil { o.log.Warn("chapter refs write failed", "slug", meta.Slug, "err", wErr) } // ── Step 3: Chapter texts (worker pool) ─────────────────────────────────── type chapterJob struct { slug string ref domain.ChapterRef total int // total chapters to scrape (for progress logging) } work := make(chan chapterJob, o.workers*4) var scraped, skipped, errors atomic.Int64 var wg sync.WaitGroup for i := 0; i < o.workers; i++ { wg.Add(1) go func(workerID int) { defer wg.Done() for job := range work { select { case <-ctx.Done(): return default: } if o.store.ChapterExists(ctx, job.slug, job.ref) { o.log.Debug("chapter already exists, skipping", "slug", job.slug, "chapter", job.ref.Number) skipped.Add(1) continue } ch, err := o.novel.ScrapeChapterText(ctx, job.ref) if err != nil { o.log.Error("chapter scrape failed", "slug", job.slug, "chapter", job.ref.Number, "err", err) errors.Add(1) continue } if err := o.store.WriteChapter(ctx, job.slug, ch); err != nil { o.log.Error("chapter write failed", "slug", job.slug, "chapter", job.ref.Number, "err", err) errors.Add(1) continue } n := scraped.Add(1) // Log a progress summary every 25 chapters scraped. if n%25 == 0 { o.log.Info("scraping chapters", "slug", job.slug, "scraped", n, "total", job.total) } } }(i) } // Count how many chapters will actually be enqueued (for progress logging). toScrape := 0 for _, ref := range refs { if task.FromChapter > 0 && ref.Number < task.FromChapter { continue } if task.ToChapter > 0 && ref.Number > task.ToChapter { continue } toScrape++ } // Enqueue chapter jobs respecting the optional range filter from the task. for _, ref := range refs { if task.FromChapter > 0 && ref.Number < task.FromChapter { skipped.Add(1) continue } if task.ToChapter > 0 && ref.Number > task.ToChapter { skipped.Add(1) continue } select { case <-ctx.Done(): goto drain case work <- chapterJob{slug: meta.Slug, ref: ref, total: toScrape}: } } drain: close(work) wg.Wait() result.ChaptersScraped = int(scraped.Load()) result.ChaptersSkipped = int(skipped.Load()) result.Errors += int(errors.Load()) o.log.Info("book scrape finished", "slug", meta.Slug, "scraped", result.ChaptersScraped, "skipped", result.ChaptersSkipped, "errors", result.Errors, ) return result }