libnovel/scraper/internal/orchestrator/orchestrator.go

// 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"
	"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.
	// Defaults to runtime.NumCPU() when 0.
	Workers int

	// StaticRoot is kept for backwards-compatibility but is no longer used
	// when a Store is provided.
	StaticRoot string

	// 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.
type Orchestrator struct {
	cfg     Config
	novel   scraper.NovelScraper
	store   storage.Store
	log     *slog.Logger
	workers int
}

// New returns a new Orchestrator backed by the provided Store.
func New(cfg Config, novel scraper.NovelScraper, log *slog.Logger, store storage.Store) *Orchestrator {
	workers := cfg.Workers
	if workers <= 0 {
		workers = runtime.NumCPU()
	}
	return &Orchestrator{
		cfg:     cfg,
		novel:   novel,
		store:   store,
		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,
	)

	// 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
		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 stored.
				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
				}

				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,
					)
					errors.Add(1)
					notify()
					continue
				}

				if err := o.store.WriteChapter(ctx, job.slug, chapter); err != nil {
					o.log.Error("chapter write failed",
						"book", job.slug,
						"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,
					"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)
			errors.Add(1)
			notify()
			return
		}

		// Persist / update metadata.
		if err := o.store.WriteMetadata(ctx, meta); err != nil {
			o.log.Error("metadata write failed", "slug", meta.Slug, "err", err)
			// 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
		}

		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)
				errors.Add(1)
				notify()
			}
		}()

		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()

	// Final progress notification.
	notify()

	if ctx.Err() != nil {
		return fmt.Errorf("orchestrator: context cancelled: %w", ctx.Err())
	}

	o.log.Info("orchestrator finished")
	return nil
}