libnovel/backend/internal/orchestrator/orchestrator.go

// 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.Info("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
	}

	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
		// Fire optional post-metadata hook (e.g. Meilisearch indexing).
		if o.postMetadata != nil {
			o.postMetadata(ctx, meta)
		}
	}

	o.log.Info("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.Info("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
}