libnovel/backend/internal/asynqqueue/producer.go

package asynqqueue

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"

	"github.com/hibiken/asynq"
	"github.com/libnovel/backend/internal/taskqueue"
)

// Producer dual-writes every task: first to PocketBase (via pb, for audit /
// UI status), then to Redis via Asynq so the runner picks it up immediately.
type Producer struct {
	pb     taskqueue.Producer // underlying PocketBase producer
	client *asynq.Client
	log    *slog.Logger
}

// NewProducer wraps an existing PocketBase Producer with Asynq dispatch.
func NewProducer(pb taskqueue.Producer, redisOpt asynq.RedisConnOpt, log *slog.Logger) *Producer {
	return &Producer{
		pb:     pb,
		client: asynq.NewClient(redisOpt),
		log:    log,
	}
}

// Close shuts down the underlying Asynq client connection.
func (p *Producer) Close() error {
	return p.client.Close()
}

// CreateScrapeTask creates a PocketBase record then enqueues an Asynq job.
func (p *Producer) CreateScrapeTask(ctx context.Context, kind, targetURL string, fromChapter, toChapter int) (string, error) {
	id, err := p.pb.CreateScrapeTask(ctx, kind, targetURL, fromChapter, toChapter)
	if err != nil {
		return "", err
	}

	payload := ScrapePayload{
		PBTaskID:    id,
		Kind:        kind,
		TargetURL:   targetURL,
		FromChapter: fromChapter,
		ToChapter:   toChapter,
	}
	taskType := TypeScrapeBook
	if kind == "catalogue" {
		taskType = TypeScrapeCatalogue
	}
	if err := p.enqueue(ctx, taskType, payload); err != nil {
		// Non-fatal: PB record exists; runner will pick it up on next poll.
		p.log.Warn("asynq enqueue scrape failed (task still in PB, runner will poll)",
			"task_id", id, "err", err)
		return id, nil
	}
	return id, nil
}

// CreateAudioTask creates a PocketBase record then enqueues an Asynq job.
func (p *Producer) CreateAudioTask(ctx context.Context, slug string, chapter int, voice string) (string, error) {
	id, err := p.pb.CreateAudioTask(ctx, slug, chapter, voice)
	if err != nil {
		return "", err
	}

	payload := AudioPayload{
		PBTaskID: id,
		Slug:     slug,
		Chapter:  chapter,
		Voice:    voice,
	}
	if err := p.enqueue(ctx, TypeAudioGenerate, payload); err != nil {
		// Non-fatal: PB record exists; runner will pick it up on next poll.
		p.log.Warn("asynq enqueue audio failed (task still in PB, runner will poll)",
			"task_id", id, "err", err)
		return id, nil
	}
	return id, nil
}

// CreateTranslationTask creates a PocketBase record. Translation tasks are
// not currently dispatched via Asynq — the runner picks them up via polling.
func (p *Producer) CreateTranslationTask(ctx context.Context, slug string, chapter int, lang string) (string, error) {
	return p.pb.CreateTranslationTask(ctx, slug, chapter, lang)
}

// CancelTask delegates to PocketBase; Asynq jobs may already be running and
// cannot be reliably cancelled, so we only update the audit record.
func (p *Producer) CancelTask(ctx context.Context, id string) error {
	return p.pb.CancelTask(ctx, id)
}

// CancelAudioTasksBySlug delegates to PocketBase to cancel all pending/running
// audio tasks for slug.
func (p *Producer) CancelAudioTasksBySlug(ctx context.Context, slug string) (int, error) {
	return p.pb.CancelAudioTasksBySlug(ctx, slug)
}

// enqueue serialises payload and dispatches it to Asynq.
func (p *Producer) enqueue(_ context.Context, taskType string, payload any) error {
	b, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("marshal payload: %w", err)
	}
	_, err = p.client.Enqueue(asynq.NewTask(taskType, b))
	return err
}