libnovel/backend/internal/runner/runner.go

// 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.
package runner

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"sync"
	"time"

	"github.com/libnovel/backend/internal/bookstore"
	"github.com/libnovel/backend/internal/domain"
	"github.com/libnovel/backend/internal/kokoro"
	"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
}

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

// New creates a Runner from cfg and deps.
// Any zero/nil field in deps will cause a panic at construction time to fail fast.
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 deps.Log == nil {
		deps.Log = slog.Default()
	}
	return &Runner{cfg: cfg, deps: deps}
}

// livenessFile is the path written on every successful poll so that the Docker
// healthcheck (CMD /healthcheck file /tmp/runner.alive <max_age>) can verify
// the runner is still making progress.
const livenessFile = "/tmp/runner.alive"

// touchAlive writes the current UTC time to livenessFile. Errors are logged but
// never fatal — liveness is best-effort and should not crash the runner.
func (r *Runner) touchAlive() {
	data := []byte(time.Now().UTC().Format(time.RFC3339))
	if err := os.WriteFile(livenessFile, data, 0o644); err != nil {
		r.deps.Log.Warn("runner: failed to write liveness file", "err", err)
	}
}

// Run starts the poll loop, blocking until ctx is cancelled.
// On each tick it claims and executes all available pending tasks.
// Scrape and audio tasks run in separate goroutine pools bounded by
// MaxConcurrentScrape and MaxConcurrentAudio respectively.
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,
	)

	scrapeSem := make(chan struct{}, r.cfg.MaxConcurrentScrape)
	audioSem := make(chan struct{}, r.cfg.MaxConcurrentAudio)
	var wg sync.WaitGroup

	// Write liveness file immediately so the first healthcheck passes before
	// the first poll completes.
	r.touchAlive()

	tick := time.NewTicker(r.cfg.PollInterval)
	defer tick.Stop()

	browseTick := time.NewTicker(r.cfg.BrowseRefreshInterval)
	defer browseTick.Stop()

	// Run one browse refresh and one poll immediately on startup.
	go r.runBrowseRefresh(ctx)

	// Run one poll immediately on startup, then on each tick.
	for {
		r.poll(ctx, scrapeSem, audioSem, &wg)
		r.touchAlive()

		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 <-tick.C:
		}
	}
}

// poll claims all available pending tasks and dispatches them to goroutines.
// It claims tasks in a tight loop until no more are available.
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 // queue empty
		}
		// Acquire semaphore (non-blocking when full — leave task running).
		select {
		case scrapeSem <- struct{}{}:
		default:
			// Too many concurrent scrapes — the task stays claimed but we can't
			// run it right now. Log and break; the next poll will pick it up if
			// still running (it won't be re-claimed while status=running).
			r.deps.Log.Warn("runner: scrape semaphore full, will retry next tick",
				"task_id", task.ID)
			break
		}
		wg.Add(1)
		go func(t domain.ScrapeTask) {
			defer wg.Done()
			defer func() { <-scrapeSem }()
			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 // queue empty
		}
		select {
		case audioSem <- struct{}{}:
		default:
			r.deps.Log.Warn("runner: audio semaphore full, will retry next tick",
				"task_id", task.ID)
			break
		}
		wg.Add(1)
		go func(t domain.AudioTask) {
			defer wg.Done()
			defer func() { <-audioSem }()
			r.runAudioTask(ctx, t)
		}(task)
	}
}

// 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")

	// Heartbeat goroutine: periodically PATCH heartbeat_at so the reaper knows
	// this task is still alive. Cancelled when the task finishes.
	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)
				}
			}
		}
	}()

	oCfg := orchestrator.Config{Workers: r.cfg.OrchestratorWorkers}
	o := orchestrator.New(oCfg, r.deps.Novel, r.deps.BookWriter, r.deps.Log)

	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)
	}
	log.Info("runner: scrape task finished",
		"scraped", result.ChaptersScraped,
		"skipped", result.ChaptersSkipped,
		"errors", result.Errors,
	)
}

// runCatalogueTask runs a full catalogue scrape by iterating catalogue entries
// and running a book task for each one.
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:
//  1. Read chapter text from MinIO.
//  2. Call Kokoro to generate audio.
//  3. Upload MP3 to MinIO under the standard audio object key.
//  4. Report result back to PocketBase.
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")

	// Heartbeat goroutine: periodically PATCH heartbeat_at so the reaper knows
	// this task is still alive. Cancelled when the task finishes.
	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)
		result := domain.AudioResult{ErrorMessage: msg}
		if err := r.deps.Consumer.FinishAudioTask(ctx, task.ID, result); err != nil {
			log.Error("runner: FinishAudioTask failed", "err", err)
		}
	}

	// Step 1: read chapter text.
	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
	}

	// Step 2: generate audio.
	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
	}

	// Step 3: upload to MinIO.
	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
	}

	// Step 4: report success.
	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)
}