libnovel/backend/internal/taskqueue/taskqueue.go

// Package taskqueue defines the interfaces for creating and consuming
// scrape/audio tasks stored in PocketBase.
//
// Interface segregation:
//   - Producer is used only by the backend (creates tasks, cancels tasks).
//   - Consumer is used only by the runner (claims tasks, reports results).
//   - Reader is used by the backend for status/history endpoints.
//
// Concrete implementations live in internal/storage.
package taskqueue

import (
	"context"
	"time"

	"github.com/libnovel/backend/internal/domain"
)

// Producer is the write side of the task queue used by the backend service.
// It creates new tasks in PocketBase for the runner to pick up.
type Producer interface {
	// CreateScrapeTask inserts a new scrape task with status=pending and
	// returns the assigned PocketBase record ID.
	// kind is one of "catalogue", "book", or "book_range".
	// targetURL is the book URL (empty for catalogue-wide tasks).
	CreateScrapeTask(ctx context.Context, kind, targetURL string, fromChapter, toChapter int) (string, error)

	// CreateAudioTask inserts a new audio task with status=pending and
	// returns the assigned PocketBase record ID.
	CreateAudioTask(ctx context.Context, slug string, chapter int, voice string) (string, error)

	// CancelTask transitions a pending task to status=cancelled.
	// Returns ErrNotFound if the task does not exist.
	CancelTask(ctx context.Context, id string) error
}

// Consumer is the read/claim side of the task queue used by the runner.
type Consumer interface {
	// ClaimNextScrapeTask atomically finds the oldest pending scrape task,
	// sets its status=running and worker_id=workerID, and returns it.
	// Returns (zero, false, nil) when the queue is empty.
	ClaimNextScrapeTask(ctx context.Context, workerID string) (domain.ScrapeTask, bool, error)

	// ClaimNextAudioTask atomically finds the oldest pending audio task,
	// sets its status=running and worker_id=workerID, and returns it.
	// Returns (zero, false, nil) when the queue is empty.
	ClaimNextAudioTask(ctx context.Context, workerID string) (domain.AudioTask, bool, error)

	// FinishScrapeTask marks a running scrape task as done and records the result.
	FinishScrapeTask(ctx context.Context, id string, result domain.ScrapeResult) error

	// FinishAudioTask marks a running audio task as done and records the result.
	FinishAudioTask(ctx context.Context, id string, result domain.AudioResult) error

	// FailTask marks a task (scrape or audio) as failed with an error message.
	FailTask(ctx context.Context, id, errMsg string) error

	// HeartbeatTask updates the heartbeat_at timestamp on a running task.
	// Should be called periodically by the runner while the task is active so
	// the reaper knows the task is still alive.
	HeartbeatTask(ctx context.Context, id string) error

	// ReapStaleTasks finds all running tasks whose heartbeat_at is older than
	// staleAfter (or was never set) and resets them to pending so they can be
	// re-claimed by a healthy runner. Returns the number of tasks reaped.
	ReapStaleTasks(ctx context.Context, staleAfter time.Duration) (int, error)
}

// Reader is the read-only side used by the backend for status pages.
type Reader interface {
	// ListScrapeTasks returns all scrape tasks sorted by started descending.
	ListScrapeTasks(ctx context.Context) ([]domain.ScrapeTask, error)

	// GetScrapeTask returns a single scrape task by ID.
	// Returns (zero, false, nil) if not found.
	GetScrapeTask(ctx context.Context, id string) (domain.ScrapeTask, bool, error)

	// ListAudioTasks returns all audio tasks sorted by started descending.
	ListAudioTasks(ctx context.Context) ([]domain.AudioTask, error)

	// GetAudioTask returns the most recent audio task for cacheKey.
	// Returns (zero, false, nil) if not found.
	GetAudioTask(ctx context.Context, cacheKey string) (domain.AudioTask, bool, error)
}