feat: add v2 stack (backend, runner, ui-v2) with release workflow
All checks were successful
Release / Scraper / Test (push) Successful in 10s
Release / UI / Build (push) Successful in 26s
Release / v2 / Build ui-v2 (push) Successful in 17s
Release / Scraper / Docker (push) Successful in 47s
Release / UI / Docker (push) Successful in 56s
CI / Scraper / Lint (pull_request) Successful in 7s
CI / Scraper / Test (pull_request) Successful in 8s
CI / UI / Build (pull_request) Successful in 16s
CI / Scraper / Docker Push (pull_request) Has been skipped
CI / UI / Docker Push (pull_request) Has been skipped
Release / v2 / Docker / ui-v2 (push) Successful in 56s
Release / v2 / Test backend (push) Successful in 4m35s
iOS CI / Build (pull_request) Successful in 4m28s
Release / v2 / Docker / backend (push) Successful in 1m29s
Release / v2 / Docker / runner (push) Successful in 1m39s
iOS CI / Test (pull_request) Successful in 9m51s
All checks were successful
Release / Scraper / Test (push) Successful in 10s
Release / UI / Build (push) Successful in 26s
Release / v2 / Build ui-v2 (push) Successful in 17s
Release / Scraper / Docker (push) Successful in 47s
Release / UI / Docker (push) Successful in 56s
CI / Scraper / Lint (pull_request) Successful in 7s
CI / Scraper / Test (pull_request) Successful in 8s
CI / UI / Build (pull_request) Successful in 16s
CI / Scraper / Docker Push (pull_request) Has been skipped
CI / UI / Docker Push (pull_request) Has been skipped
Release / v2 / Docker / ui-v2 (push) Successful in 56s
Release / v2 / Test backend (push) Successful in 4m35s
iOS CI / Build (pull_request) Successful in 4m28s
Release / v2 / Docker / backend (push) Successful in 1m29s
Release / v2 / Docker / runner (push) Successful in 1m39s
iOS CI / Test (pull_request) Successful in 9m51s
- backend/: Go API server and runner binaries with PocketBase + MinIO storage - ui-v2/: SvelteKit frontend rewrite - docker-compose-new.yml: compose file for the v2 stack - .gitea/workflows/release-v2.yaml: CI/CD for backend, runner, and ui-v2 Docker Hub images - scripts/pb-init.sh: migrate from wget to curl, add superuser bootstrap for fresh installs - .env.example: document DOCKER_BUILDKIT=1 for Colima users
This commit is contained in:
21
backend/internal/runner/helpers.go
Normal file
21
backend/internal/runner/helpers.go
Normal file
@@ -0,0 +1,21 @@
|
||||
package runner
|
||||
|
||||
import (
|
||||
"regexp"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// stripMarkdown removes common markdown syntax from src, returning plain text
|
||||
// suitable for TTS. Mirrors the helper in the scraper's server package.
|
||||
func stripMarkdown(src string) string {
|
||||
src = regexp.MustCompile(`(?m)^#{1,6}\s+`).ReplaceAllString(src, "")
|
||||
src = regexp.MustCompile(`\*{1,3}|_{1,3}`).ReplaceAllString(src, "")
|
||||
src = regexp.MustCompile("(?s)```.*?```").ReplaceAllString(src, "")
|
||||
src = regexp.MustCompile("`[^`]*`").ReplaceAllString(src, "")
|
||||
src = regexp.MustCompile(`\[([^\]]+)\]\([^)]+\)`).ReplaceAllString(src, "$1")
|
||||
src = regexp.MustCompile(`!\[[^\]]*\]\([^)]+\)`).ReplaceAllString(src, "")
|
||||
src = regexp.MustCompile(`(?m)^>\s?`).ReplaceAllString(src, "")
|
||||
src = regexp.MustCompile(`(?m)^[-*_]{3,}\s*$`).ReplaceAllString(src, "")
|
||||
src = regexp.MustCompile(`\n{3,}`).ReplaceAllString(src, "\n\n")
|
||||
return strings.TrimSpace(src)
|
||||
}
|
||||
386
backend/internal/runner/runner.go
Normal file
386
backend/internal/runner/runner.go
Normal file
@@ -0,0 +1,386 @@
|
||||
// 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
|
||||
}
|
||||
|
||||
// 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
|
||||
// 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 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,
|
||||
)
|
||||
|
||||
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()
|
||||
|
||||
// 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 <-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)
|
||||
}
|
||||
365
backend/internal/runner/runner_test.go
Normal file
365
backend/internal/runner/runner_test.go
Normal file
@@ -0,0 +1,365 @@
|
||||
package runner_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/libnovel/backend/internal/domain"
|
||||
"github.com/libnovel/backend/internal/runner"
|
||||
)
|
||||
|
||||
// ── Stub types ────────────────────────────────────────────────────────────────
|
||||
|
||||
// stubConsumer is a test double for taskqueue.Consumer.
|
||||
type stubConsumer struct {
|
||||
scrapeQueue []domain.ScrapeTask
|
||||
audioQueue []domain.AudioTask
|
||||
scrapeIdx int
|
||||
audioIdx int
|
||||
finished []string
|
||||
failCalled []string
|
||||
claimErr error
|
||||
}
|
||||
|
||||
func (s *stubConsumer) ClaimNextScrapeTask(_ context.Context, _ string) (domain.ScrapeTask, bool, error) {
|
||||
if s.claimErr != nil {
|
||||
return domain.ScrapeTask{}, false, s.claimErr
|
||||
}
|
||||
if s.scrapeIdx >= len(s.scrapeQueue) {
|
||||
return domain.ScrapeTask{}, false, nil
|
||||
}
|
||||
t := s.scrapeQueue[s.scrapeIdx]
|
||||
s.scrapeIdx++
|
||||
return t, true, nil
|
||||
}
|
||||
|
||||
func (s *stubConsumer) ClaimNextAudioTask(_ context.Context, _ string) (domain.AudioTask, bool, error) {
|
||||
if s.claimErr != nil {
|
||||
return domain.AudioTask{}, false, s.claimErr
|
||||
}
|
||||
if s.audioIdx >= len(s.audioQueue) {
|
||||
return domain.AudioTask{}, false, nil
|
||||
}
|
||||
t := s.audioQueue[s.audioIdx]
|
||||
s.audioIdx++
|
||||
return t, true, nil
|
||||
}
|
||||
|
||||
func (s *stubConsumer) FinishScrapeTask(_ context.Context, id string, _ domain.ScrapeResult) error {
|
||||
s.finished = append(s.finished, id)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *stubConsumer) FinishAudioTask(_ context.Context, id string, _ domain.AudioResult) error {
|
||||
s.finished = append(s.finished, id)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *stubConsumer) FailTask(_ context.Context, id, _ string) error {
|
||||
s.failCalled = append(s.failCalled, id)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *stubConsumer) HeartbeatTask(_ context.Context, _ string) error { return nil }
|
||||
|
||||
func (s *stubConsumer) ReapStaleTasks(_ context.Context, _ time.Duration) (int, error) {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// stubBookWriter satisfies bookstore.BookWriter (no-op).
|
||||
type stubBookWriter struct{}
|
||||
|
||||
func (s *stubBookWriter) WriteMetadata(_ context.Context, _ domain.BookMeta) error { return nil }
|
||||
func (s *stubBookWriter) WriteChapter(_ context.Context, _ string, _ domain.Chapter) error {
|
||||
return nil
|
||||
}
|
||||
func (s *stubBookWriter) WriteChapterRefs(_ context.Context, _ string, _ []domain.ChapterRef) error {
|
||||
return nil
|
||||
}
|
||||
func (s *stubBookWriter) ChapterExists(_ context.Context, _ string, _ domain.ChapterRef) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// stubBookReader satisfies bookstore.BookReader — returns a single chapter.
|
||||
type stubBookReader struct {
|
||||
text string
|
||||
readErr error
|
||||
}
|
||||
|
||||
func (s *stubBookReader) ReadChapter(_ context.Context, _ string, _ int) (string, error) {
|
||||
return s.text, s.readErr
|
||||
}
|
||||
func (s *stubBookReader) ReadMetadata(_ context.Context, _ string) (domain.BookMeta, bool, error) {
|
||||
return domain.BookMeta{}, false, nil
|
||||
}
|
||||
func (s *stubBookReader) ListBooks(_ context.Context) ([]domain.BookMeta, error) { return nil, nil }
|
||||
func (s *stubBookReader) LocalSlugs(_ context.Context) (map[string]bool, error) { return nil, nil }
|
||||
func (s *stubBookReader) MetadataMtime(_ context.Context, _ string) int64 { return 0 }
|
||||
func (s *stubBookReader) ListChapters(_ context.Context, _ string) ([]domain.ChapterInfo, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (s *stubBookReader) CountChapters(_ context.Context, _ string) int { return 0 }
|
||||
func (s *stubBookReader) ReindexChapters(_ context.Context, _ string) (int, error) {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// stubAudioStore satisfies bookstore.AudioStore.
|
||||
type stubAudioStore struct {
|
||||
putCalled atomic.Int32
|
||||
putErr error
|
||||
}
|
||||
|
||||
func (s *stubAudioStore) AudioObjectKey(slug string, n int, voice string) string {
|
||||
return slug + "/" + string(rune('0'+n)) + "/" + voice + ".mp3"
|
||||
}
|
||||
func (s *stubAudioStore) AudioExists(_ context.Context, _ string) bool { return false }
|
||||
func (s *stubAudioStore) PutAudio(_ context.Context, _ string, _ []byte) error {
|
||||
s.putCalled.Add(1)
|
||||
return s.putErr
|
||||
}
|
||||
|
||||
// stubNovelScraper satisfies scraper.NovelScraper minimally.
|
||||
type stubNovelScraper struct {
|
||||
entries []domain.CatalogueEntry
|
||||
metaErr error
|
||||
chapters []domain.ChapterRef
|
||||
}
|
||||
|
||||
func (s *stubNovelScraper) ScrapeCatalogue(_ context.Context) (<-chan domain.CatalogueEntry, <-chan error) {
|
||||
ch := make(chan domain.CatalogueEntry, len(s.entries))
|
||||
errCh := make(chan error, 1)
|
||||
for _, e := range s.entries {
|
||||
ch <- e
|
||||
}
|
||||
close(ch)
|
||||
close(errCh)
|
||||
return ch, errCh
|
||||
}
|
||||
|
||||
func (s *stubNovelScraper) ScrapeMetadata(_ context.Context, _ string) (domain.BookMeta, error) {
|
||||
if s.metaErr != nil {
|
||||
return domain.BookMeta{}, s.metaErr
|
||||
}
|
||||
return domain.BookMeta{Slug: "test-book", Title: "Test Book", SourceURL: "https://example.com/book/test-book"}, nil
|
||||
}
|
||||
|
||||
func (s *stubNovelScraper) ScrapeChapterList(_ context.Context, _ string) ([]domain.ChapterRef, error) {
|
||||
return s.chapters, nil
|
||||
}
|
||||
|
||||
func (s *stubNovelScraper) ScrapeChapterText(_ context.Context, ref domain.ChapterRef) (domain.Chapter, error) {
|
||||
return domain.Chapter{Ref: ref, Text: "# Chapter\n\nSome text."}, nil
|
||||
}
|
||||
|
||||
func (s *stubNovelScraper) ScrapeRanking(_ context.Context, _ int) (<-chan domain.BookMeta, <-chan error) {
|
||||
ch := make(chan domain.BookMeta)
|
||||
errCh := make(chan error, 1)
|
||||
close(ch)
|
||||
close(errCh)
|
||||
return ch, errCh
|
||||
}
|
||||
|
||||
func (s *stubNovelScraper) SourceName() string { return "stub" }
|
||||
|
||||
// stubKokoro satisfies kokoro.Client.
|
||||
type stubKokoro struct {
|
||||
data []byte
|
||||
genErr error
|
||||
called atomic.Int32
|
||||
}
|
||||
|
||||
func (s *stubKokoro) GenerateAudio(_ context.Context, _, _ string) ([]byte, error) {
|
||||
s.called.Add(1)
|
||||
return s.data, s.genErr
|
||||
}
|
||||
|
||||
func (s *stubKokoro) ListVoices(_ context.Context) ([]string, error) {
|
||||
return []string{"af_bella"}, nil
|
||||
}
|
||||
|
||||
// ── stripMarkdown helper ──────────────────────────────────────────────────────
|
||||
|
||||
func TestStripMarkdownViaAudioTask(t *testing.T) {
|
||||
// Verify markdown is stripped before sending to Kokoro.
|
||||
// We inject chapter text with markdown; the kokoro stub verifies data flows.
|
||||
consumer := &stubConsumer{
|
||||
audioQueue: []domain.AudioTask{
|
||||
{ID: "a1", Slug: "book", Chapter: 1, Voice: "af_bella", Status: domain.TaskStatusRunning},
|
||||
},
|
||||
}
|
||||
bookReader := &stubBookReader{text: "## Chapter 1\n\nPlain **text** here."}
|
||||
audioStore := &stubAudioStore{}
|
||||
kokoroStub := &stubKokoro{data: []byte("mp3")}
|
||||
|
||||
cfg := runner.Config{
|
||||
WorkerID: "test",
|
||||
PollInterval: time.Hour, // long poll — we'll cancel manually
|
||||
}
|
||||
deps := runner.Dependencies{
|
||||
Consumer: consumer,
|
||||
BookWriter: &stubBookWriter{},
|
||||
BookReader: bookReader,
|
||||
AudioStore: audioStore,
|
||||
Novel: &stubNovelScraper{},
|
||||
Kokoro: kokoroStub,
|
||||
}
|
||||
|
||||
r := runner.New(cfg, deps)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
_ = r.Run(ctx)
|
||||
|
||||
if kokoroStub.called.Load() != 1 {
|
||||
t.Errorf("expected Kokoro.GenerateAudio called once, got %d", kokoroStub.called.Load())
|
||||
}
|
||||
if audioStore.putCalled.Load() != 1 {
|
||||
t.Errorf("expected PutAudio called once, got %d", audioStore.putCalled.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAudioTask_ReadChapterError(t *testing.T) {
|
||||
consumer := &stubConsumer{
|
||||
audioQueue: []domain.AudioTask{
|
||||
{ID: "a2", Slug: "book", Chapter: 2, Voice: "af_bella", Status: domain.TaskStatusRunning},
|
||||
},
|
||||
}
|
||||
bookReader := &stubBookReader{readErr: errors.New("chapter not found")}
|
||||
audioStore := &stubAudioStore{}
|
||||
kokoroStub := &stubKokoro{data: []byte("mp3")}
|
||||
|
||||
cfg := runner.Config{WorkerID: "test", PollInterval: time.Hour}
|
||||
deps := runner.Dependencies{
|
||||
Consumer: consumer,
|
||||
BookWriter: &stubBookWriter{},
|
||||
BookReader: bookReader,
|
||||
AudioStore: audioStore,
|
||||
Novel: &stubNovelScraper{},
|
||||
Kokoro: kokoroStub,
|
||||
}
|
||||
|
||||
r := runner.New(cfg, deps)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
_ = r.Run(ctx)
|
||||
|
||||
// Kokoro should not be called; FinishAudioTask should be called with error.
|
||||
if kokoroStub.called.Load() != 0 {
|
||||
t.Errorf("expected Kokoro not called, got %d", kokoroStub.called.Load())
|
||||
}
|
||||
if len(consumer.finished) != 1 {
|
||||
t.Errorf("expected FinishAudioTask called once, got %d", len(consumer.finished))
|
||||
}
|
||||
}
|
||||
|
||||
func TestAudioTask_KokoroError(t *testing.T) {
|
||||
consumer := &stubConsumer{
|
||||
audioQueue: []domain.AudioTask{
|
||||
{ID: "a3", Slug: "book", Chapter: 3, Voice: "af_bella", Status: domain.TaskStatusRunning},
|
||||
},
|
||||
}
|
||||
bookReader := &stubBookReader{text: "Chapter text."}
|
||||
audioStore := &stubAudioStore{}
|
||||
kokoroStub := &stubKokoro{genErr: errors.New("tts failed")}
|
||||
|
||||
cfg := runner.Config{WorkerID: "test", PollInterval: time.Hour}
|
||||
deps := runner.Dependencies{
|
||||
Consumer: consumer,
|
||||
BookWriter: &stubBookWriter{},
|
||||
BookReader: bookReader,
|
||||
AudioStore: audioStore,
|
||||
Novel: &stubNovelScraper{},
|
||||
Kokoro: kokoroStub,
|
||||
}
|
||||
|
||||
r := runner.New(cfg, deps)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
_ = r.Run(ctx)
|
||||
|
||||
if audioStore.putCalled.Load() != 0 {
|
||||
t.Errorf("expected PutAudio not called, got %d", audioStore.putCalled.Load())
|
||||
}
|
||||
if len(consumer.finished) != 1 {
|
||||
t.Errorf("expected FinishAudioTask called once, got %d", len(consumer.finished))
|
||||
}
|
||||
}
|
||||
|
||||
func TestScrapeTask_BookKind(t *testing.T) {
|
||||
consumer := &stubConsumer{
|
||||
scrapeQueue: []domain.ScrapeTask{
|
||||
{ID: "s1", Kind: "book", TargetURL: "https://example.com/book/test-book", Status: domain.TaskStatusRunning},
|
||||
},
|
||||
}
|
||||
|
||||
cfg := runner.Config{WorkerID: "test", PollInterval: time.Hour}
|
||||
deps := runner.Dependencies{
|
||||
Consumer: consumer,
|
||||
BookWriter: &stubBookWriter{},
|
||||
BookReader: &stubBookReader{},
|
||||
AudioStore: &stubAudioStore{},
|
||||
Novel: &stubNovelScraper{},
|
||||
Kokoro: &stubKokoro{},
|
||||
}
|
||||
|
||||
r := runner.New(cfg, deps)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
_ = r.Run(ctx)
|
||||
|
||||
if len(consumer.finished) != 1 || consumer.finished[0] != "s1" {
|
||||
t.Errorf("expected task s1 finished, got %v", consumer.finished)
|
||||
}
|
||||
}
|
||||
|
||||
func TestScrapeTask_UnknownKind(t *testing.T) {
|
||||
consumer := &stubConsumer{
|
||||
scrapeQueue: []domain.ScrapeTask{
|
||||
{ID: "s2", Kind: "unknown_kind", Status: domain.TaskStatusRunning},
|
||||
},
|
||||
}
|
||||
|
||||
cfg := runner.Config{WorkerID: "test", PollInterval: time.Hour}
|
||||
deps := runner.Dependencies{
|
||||
Consumer: consumer,
|
||||
BookWriter: &stubBookWriter{},
|
||||
BookReader: &stubBookReader{},
|
||||
AudioStore: &stubAudioStore{},
|
||||
Novel: &stubNovelScraper{},
|
||||
Kokoro: &stubKokoro{},
|
||||
}
|
||||
|
||||
r := runner.New(cfg, deps)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
_ = r.Run(ctx)
|
||||
|
||||
// Unknown kind still finishes the task (with error message in result).
|
||||
if len(consumer.finished) != 1 || consumer.finished[0] != "s2" {
|
||||
t.Errorf("expected task s2 finished, got %v", consumer.finished)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_CancelImmediately(t *testing.T) {
|
||||
consumer := &stubConsumer{}
|
||||
cfg := runner.Config{WorkerID: "test", PollInterval: 10 * time.Millisecond}
|
||||
deps := runner.Dependencies{
|
||||
Consumer: consumer,
|
||||
BookWriter: &stubBookWriter{},
|
||||
BookReader: &stubBookReader{},
|
||||
AudioStore: &stubAudioStore{},
|
||||
Novel: &stubNovelScraper{},
|
||||
Kokoro: &stubKokoro{},
|
||||
}
|
||||
|
||||
r := runner.New(cfg, deps)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel() // cancel before Run
|
||||
|
||||
err := r.Run(ctx)
|
||||
if err != nil {
|
||||
t.Errorf("expected nil on graceful shutdown, got %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user