feat(v3): add v3 stack — backend rewrite, renamed env vars, docs

- New Go backend binary (backend + runner) replacing old scraper/ - Rename SCRAPER_API_URL → BACKEND_API_URL in UI env and docker-compose - Rename scraperFetch → backendFetch across all 19 UI server files - Remove SCRAPER_PROXY env var and proxy transport from browser.Config - Add Meilisearch, Valkey, Caddy to docker-compose - Add docs/: api-endpoints.md, request-flow.mermaid.md, data-flow.mermaid.md
2026-03-22 17:27:32 +05:00
parent 29d0eeb7e8
commit a85636d5db
178 changed files with 25951 additions and 0 deletions
--- a/v3/backend/internal/runner/runner.go
+++ b/v3/backend/internal/runner/runner.go
@@ -0,0 +1,440 @@
+// Package runner implements the worker loop that polls PocketBase for pending
+// scrape and audio tasks, executes them, and reports results back.
+//
+// Design:
+//   - Run(ctx) loops on a ticker; each tick claims and dispatches pending tasks.
+//   - Scrape tasks are dispatched to the Orchestrator (one goroutine per task,
+//     up to MaxConcurrentScrape).
+//   - Audio tasks fetch chapter text, call Kokoro, upload to MinIO, and report
+//     the result back (up to MaxConcurrentAudio goroutines).
+//   - The runner is stateless between ticks; all state lives in PocketBase.
+//   - Atomic task counters are exposed via /metrics (see metrics.go).
+//   - Books are indexed in Meilisearch via an orchestrator.Config.PostMetadata
+//     hook injected at construction time.
+package runner
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/libnovel/backend/internal/bookstore"
+	"github.com/libnovel/backend/internal/domain"
+	"github.com/libnovel/backend/internal/kokoro"
+	"github.com/libnovel/backend/internal/meili"
+	"github.com/libnovel/backend/internal/orchestrator"
+	"github.com/libnovel/backend/internal/scraper"
+	"github.com/libnovel/backend/internal/taskqueue"
+)
+
+// Config tunes the runner behaviour.
+type Config struct {
+	// WorkerID uniquely identifies this runner instance in PocketBase records.
+	WorkerID string
+	// PollInterval is how often the runner checks for new tasks.
+	PollInterval time.Duration
+	// MaxConcurrentScrape limits simultaneous book-scrape goroutines.
+	MaxConcurrentScrape int
+	// MaxConcurrentAudio limits simultaneous audio-generation goroutines.
+	MaxConcurrentAudio int
+	// OrchestratorWorkers is the chapter-scraping parallelism inside each book run.
+	OrchestratorWorkers int
+	// HeartbeatInterval is how often active tasks PATCH their heartbeat_at
+	// timestamp to signal they are still alive. Defaults to 30s when 0.
+	HeartbeatInterval time.Duration
+	// StaleTaskThreshold is how old a heartbeat must be (or absent) before the
+	// task is considered orphaned and reset to pending. Defaults to 2m when 0.
+	StaleTaskThreshold time.Duration
+	// BrowseRefreshInterval is how often the runner pre-fetches browse page
+	// snapshots from novelfire.net and stores them in MinIO. Defaults to 6h.
+	BrowseRefreshInterval time.Duration
+	// CatalogueRefreshInterval is how often the runner walks the full catalogue,
+	// scrapes per-book metadata, downloads covers, and re-indexes everything in
+	// Meilisearch. Defaults to 24h (expensive — full catalogue walk).
+	CatalogueRefreshInterval time.Duration
+	// MetricsAddr is the HTTP listen address for the /metrics endpoint.
+	// Defaults to ":9091". Set to "" to disable.
+	MetricsAddr string
+}
+
+// Dependencies are the external services the runner depends on.
+type Dependencies struct {
+	// Consumer claims tasks from PocketBase.
+	Consumer taskqueue.Consumer
+	// BookWriter persists scraped data (used by orchestrator).
+	BookWriter bookstore.BookWriter
+	// BookReader reads chapter text for audio generation.
+	BookReader bookstore.BookReader
+	// AudioStore persists generated audio and checks key existence.
+	AudioStore bookstore.AudioStore
+	// BrowseStore stores browse page snapshots in MinIO.
+	BrowseStore bookstore.BrowseStore
+	// CoverStore stores book cover images in MinIO.
+	// If nil, cover downloads are skipped during catalogue refresh.
+	CoverStore bookstore.CoverStore
+	// SearchIndex indexes books in Meilisearch after scraping.
+	// If nil a no-op is used.
+	SearchIndex meili.Client
+	// Novel is the scraper implementation.
+	Novel scraper.NovelScraper
+	// Kokoro is the TTS client.
+	Kokoro kokoro.Client
+	// Log is the structured logger.
+	Log *slog.Logger
+}
+
+// Runner is the main worker process.
+type Runner struct {
+	cfg  Config
+	deps Dependencies
+
+	// Atomic task counters — read by /metrics without locking.
+	tasksRunning   atomic.Int64
+	tasksCompleted atomic.Int64
+	tasksFailed    atomic.Int64
+
+	startedAt time.Time
+}
+
+// New creates a Runner from cfg and deps.
+func New(cfg Config, deps Dependencies) *Runner {
+	if cfg.PollInterval <= 0 {
+		cfg.PollInterval = 30 * time.Second
+	}
+	if cfg.MaxConcurrentScrape <= 0 {
+		cfg.MaxConcurrentScrape = 2
+	}
+	if cfg.MaxConcurrentAudio <= 0 {
+		cfg.MaxConcurrentAudio = 1
+	}
+	if cfg.WorkerID == "" {
+		cfg.WorkerID = "runner"
+	}
+	if cfg.HeartbeatInterval <= 0 {
+		cfg.HeartbeatInterval = 30 * time.Second
+	}
+	if cfg.StaleTaskThreshold <= 0 {
+		cfg.StaleTaskThreshold = 2 * time.Minute
+	}
+	if cfg.BrowseRefreshInterval <= 0 {
+		cfg.BrowseRefreshInterval = 6 * time.Hour
+	}
+	if cfg.CatalogueRefreshInterval <= 0 {
+		cfg.CatalogueRefreshInterval = 24 * time.Hour
+	}
+	if cfg.MetricsAddr == "" {
+		cfg.MetricsAddr = ":9091"
+	}
+	if deps.Log == nil {
+		deps.Log = slog.Default()
+	}
+	if deps.SearchIndex == nil {
+		deps.SearchIndex = meili.NoopClient{}
+	}
+	return &Runner{cfg: cfg, deps: deps, startedAt: time.Now()}
+}
+
+// Run starts the poll loop and the metrics HTTP server, blocking until ctx is
+// cancelled.
+func (r *Runner) Run(ctx context.Context) error {
+	r.deps.Log.Info("runner: starting",
+		"worker_id", r.cfg.WorkerID,
+		"poll_interval", r.cfg.PollInterval,
+		"max_scrape", r.cfg.MaxConcurrentScrape,
+		"max_audio", r.cfg.MaxConcurrentAudio,
+		"browse_refresh_interval", r.cfg.BrowseRefreshInterval,
+		"catalogue_refresh_interval", r.cfg.CatalogueRefreshInterval,
+		"metrics_addr", r.cfg.MetricsAddr,
+	)
+
+	// Start metrics HTTP server in background if configured.
+	if r.cfg.MetricsAddr != "" {
+		ms := newMetricsServer(r.cfg.MetricsAddr, r, r.deps.Log)
+		go func() {
+			if err := ms.ListenAndServe(ctx); err != nil {
+				r.deps.Log.Error("runner: metrics server error", "err", err)
+			}
+		}()
+	}
+
+	scrapeSem := make(chan struct{}, r.cfg.MaxConcurrentScrape)
+	audioSem := make(chan struct{}, r.cfg.MaxConcurrentAudio)
+	var wg sync.WaitGroup
+
+	tick := time.NewTicker(r.cfg.PollInterval)
+	defer tick.Stop()
+
+	browseTick := time.NewTicker(r.cfg.BrowseRefreshInterval)
+	defer browseTick.Stop()
+
+	catalogueTick := time.NewTicker(r.cfg.CatalogueRefreshInterval)
+	defer catalogueTick.Stop()
+
+	// Run one browse refresh immediately on startup.
+	go r.runBrowseRefresh(ctx)
+	// Run one catalogue refresh immediately on startup.
+	go r.runCatalogueRefresh(ctx)
+
+	// Run one poll immediately on startup, then on each tick.
+	for {
+		r.poll(ctx, scrapeSem, audioSem, &wg)
+
+		select {
+		case <-ctx.Done():
+			r.deps.Log.Info("runner: context cancelled, draining active tasks")
+			done := make(chan struct{})
+			go func() {
+				wg.Wait()
+				close(done)
+			}()
+			select {
+			case <-done:
+				r.deps.Log.Info("runner: all tasks drained, exiting")
+			case <-time.After(2 * time.Minute):
+				r.deps.Log.Warn("runner: drain timeout exceeded, forcing exit")
+			}
+			return nil
+		case <-browseTick.C:
+			go r.runBrowseRefresh(ctx)
+		case <-catalogueTick.C:
+			go r.runCatalogueRefresh(ctx)
+		case <-tick.C:
+		}
+	}
+}
+
+// poll claims all available pending tasks and dispatches them to goroutines.
+func (r *Runner) poll(ctx context.Context, scrapeSem, audioSem chan struct{}, wg *sync.WaitGroup) {
+	// ── Reap orphaned tasks ───────────────────────────────────────────────
+	if n, err := r.deps.Consumer.ReapStaleTasks(ctx, r.cfg.StaleTaskThreshold); err != nil {
+		r.deps.Log.Warn("runner: reap stale tasks failed", "err", err)
+	} else if n > 0 {
+		r.deps.Log.Info("runner: reaped stale tasks", "count", n)
+	}
+
+	// ── Scrape tasks ──────────────────────────────────────────────────────
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		task, ok, err := r.deps.Consumer.ClaimNextScrapeTask(ctx, r.cfg.WorkerID)
+		if err != nil {
+			r.deps.Log.Error("runner: ClaimNextScrapeTask failed", "err", err)
+			break
+		}
+		if !ok {
+			break
+		}
+		select {
+		case scrapeSem <- struct{}{}:
+		default:
+			r.deps.Log.Warn("runner: scrape semaphore full, will retry next tick",
+				"task_id", task.ID)
+			break
+		}
+		r.tasksRunning.Add(1)
+		wg.Add(1)
+		go func(t domain.ScrapeTask) {
+			defer wg.Done()
+			defer func() { <-scrapeSem }()
+			defer r.tasksRunning.Add(-1)
+			r.runScrapeTask(ctx, t)
+		}(task)
+	}
+
+	// ── Audio tasks ───────────────────────────────────────────────────────
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		task, ok, err := r.deps.Consumer.ClaimNextAudioTask(ctx, r.cfg.WorkerID)
+		if err != nil {
+			r.deps.Log.Error("runner: ClaimNextAudioTask failed", "err", err)
+			break
+		}
+		if !ok {
+			break
+		}
+		select {
+		case audioSem <- struct{}{}:
+		default:
+			r.deps.Log.Warn("runner: audio semaphore full, will retry next tick",
+				"task_id", task.ID)
+			break
+		}
+		r.tasksRunning.Add(1)
+		wg.Add(1)
+		go func(t domain.AudioTask) {
+			defer wg.Done()
+			defer func() { <-audioSem }()
+			defer r.tasksRunning.Add(-1)
+			r.runAudioTask(ctx, t)
+		}(task)
+	}
+}
+
+// newOrchestrator builds an orchestrator with the Meilisearch post-hook wired in.
+func (r *Runner) newOrchestrator() *orchestrator.Orchestrator {
+	oCfg := orchestrator.Config{
+		Workers: r.cfg.OrchestratorWorkers,
+		PostMetadata: func(ctx context.Context, meta domain.BookMeta) {
+			if err := r.deps.SearchIndex.UpsertBook(ctx, meta); err != nil {
+				r.deps.Log.Warn("runner: meilisearch upsert failed",
+					"slug", meta.Slug, "err", err)
+			}
+		},
+	}
+	return orchestrator.New(oCfg, r.deps.Novel, r.deps.BookWriter, r.deps.Log)
+}
+
+// runScrapeTask executes one scrape task end-to-end and reports the result.
+func (r *Runner) runScrapeTask(ctx context.Context, task domain.ScrapeTask) {
+	log := r.deps.Log.With("task_id", task.ID, "kind", task.Kind, "url", task.TargetURL)
+	log.Info("runner: scrape task starting")
+
+	hbCtx, hbCancel := context.WithCancel(ctx)
+	defer hbCancel()
+	go func() {
+		tick := time.NewTicker(r.cfg.HeartbeatInterval)
+		defer tick.Stop()
+		for {
+			select {
+			case <-hbCtx.Done():
+				return
+			case <-tick.C:
+				if err := r.deps.Consumer.HeartbeatTask(ctx, task.ID); err != nil {
+					log.Warn("runner: heartbeat failed", "err", err)
+				}
+			}
+		}
+	}()
+
+	o := r.newOrchestrator()
+	var result domain.ScrapeResult
+
+	switch task.Kind {
+	case "catalogue":
+		result = r.runCatalogueTask(ctx, task, o, log)
+	case "book", "book_range":
+		result = o.RunBook(ctx, task)
+	default:
+		result.ErrorMessage = fmt.Sprintf("unknown task kind: %q", task.Kind)
+		log.Warn("runner: unknown task kind")
+	}
+
+	if err := r.deps.Consumer.FinishScrapeTask(ctx, task.ID, result); err != nil {
+		log.Error("runner: FinishScrapeTask failed", "err", err)
+	}
+
+	if result.ErrorMessage != "" {
+		r.tasksFailed.Add(1)
+	} else {
+		r.tasksCompleted.Add(1)
+	}
+
+	log.Info("runner: scrape task finished",
+		"scraped", result.ChaptersScraped,
+		"skipped", result.ChaptersSkipped,
+		"errors", result.Errors,
+	)
+}
+
+// runCatalogueTask runs a full catalogue scrape.
+func (r *Runner) runCatalogueTask(ctx context.Context, task domain.ScrapeTask, o *orchestrator.Orchestrator, log *slog.Logger) domain.ScrapeResult {
+	entries, errCh := r.deps.Novel.ScrapeCatalogue(ctx)
+	var result domain.ScrapeResult
+
+	for entry := range entries {
+		if ctx.Err() != nil {
+			break
+		}
+		bookTask := domain.ScrapeTask{
+			ID:        task.ID,
+			Kind:      "book",
+			TargetURL: entry.URL,
+		}
+		bookResult := o.RunBook(ctx, bookTask)
+		result.BooksFound += bookResult.BooksFound + 1
+		result.ChaptersScraped += bookResult.ChaptersScraped
+		result.ChaptersSkipped += bookResult.ChaptersSkipped
+		result.Errors += bookResult.Errors
+	}
+
+	if err := <-errCh; err != nil {
+		log.Warn("runner: catalogue scrape finished with error", "err", err)
+		result.Errors++
+		if result.ErrorMessage == "" {
+			result.ErrorMessage = err.Error()
+		}
+	}
+	return result
+}
+
+// runAudioTask executes one audio-generation task.
+func (r *Runner) runAudioTask(ctx context.Context, task domain.AudioTask) {
+	log := r.deps.Log.With("task_id", task.ID, "slug", task.Slug, "chapter", task.Chapter, "voice", task.Voice)
+	log.Info("runner: audio task starting")
+
+	hbCtx, hbCancel := context.WithCancel(ctx)
+	defer hbCancel()
+	go func() {
+		tick := time.NewTicker(r.cfg.HeartbeatInterval)
+		defer tick.Stop()
+		for {
+			select {
+			case <-hbCtx.Done():
+				return
+			case <-tick.C:
+				if err := r.deps.Consumer.HeartbeatTask(ctx, task.ID); err != nil {
+					log.Warn("runner: heartbeat failed", "err", err)
+				}
+			}
+		}
+	}()
+
+	fail := func(msg string) {
+		log.Error("runner: audio task failed", "reason", msg)
+		r.tasksFailed.Add(1)
+		result := domain.AudioResult{ErrorMessage: msg}
+		if err := r.deps.Consumer.FinishAudioTask(ctx, task.ID, result); err != nil {
+			log.Error("runner: FinishAudioTask failed", "err", err)
+		}
+	}
+
+	raw, err := r.deps.BookReader.ReadChapter(ctx, task.Slug, task.Chapter)
+	if err != nil {
+		fail(fmt.Sprintf("read chapter: %v", err))
+		return
+	}
+	text := stripMarkdown(raw)
+	if text == "" {
+		fail("chapter text is empty after stripping markdown")
+		return
+	}
+
+	if r.deps.Kokoro == nil {
+		fail("kokoro client not configured")
+		return
+	}
+	audioData, err := r.deps.Kokoro.GenerateAudio(ctx, text, task.Voice)
+	if err != nil {
+		fail(fmt.Sprintf("kokoro generate: %v", err))
+		return
+	}
+
+	key := r.deps.AudioStore.AudioObjectKey(task.Slug, task.Chapter, task.Voice)
+	if err := r.deps.AudioStore.PutAudio(ctx, key, audioData); err != nil {
+		fail(fmt.Sprintf("put audio: %v", err))
+		return
+	}
+
+	r.tasksCompleted.Add(1)
+	result := domain.AudioResult{ObjectKey: key}
+	if err := r.deps.Consumer.FinishAudioTask(ctx, task.ID, result); err != nil {
+		log.Error("runner: FinishAudioTask failed", "err", err)
+	}
+	log.Info("runner: audio task finished", "key", key)
+}