4e70a2981d
- homelab/docker-compose.yml: add redis:7-alpine service (port 6379 bound to host so Caddy TLS proxy on prod can reach it), add libretranslate service, add redis_data and libretranslate_data volumes - asynqqueue/producer.go: Asynq enqueue failures are now logged as warnings instead of returned as errors — PB record already exists so runner picks it up via poll - backend/main.go: pass logger to NewProducer Root cause: Redis was not reachable at 192.168.0.109:6379 because the redis container had no host port binding. Caddy TLS proxy terminates TLS but could not TCP-connect to the backend Redis.
105 lines
3.2 KiB
Go
105 lines
3.2 KiB
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)
|
|
}
|
|
|
|
// 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
|
|
}
|