From a771405db88a50b9339590d8588558f130b2ffad Mon Sep 17 00:00:00 2001 From: Admin Date: Thu, 2 Apr 2026 16:19:14 +0500 Subject: [PATCH] feat(audio): WAV streaming, bulk audio generation admin endpoints, cancel/resume - Add StreamAudioWAV() to pocket-tts and Kokoro clients; pocket-tts streams raw WAV directly (no ffmpeg), Kokoro requests response_format:wav with stream:true - GET /api/audio-stream supports ?format=wav for lower-latency first-byte delivery; WAV cached separately in MinIO as {slug}/{n}/{voice}.wav - Add GET /api/admin/audio/jobs with optional ?slug filter - Add POST /api/admin/audio/bulk {slug, voice, from, to, skip_existing, force} where skip_existing=true (default) resumes interrupted bulk jobs - Add POST /api/admin/audio/cancel-bulk {slug} to cancel all pending/running tasks - Add CancelAudioTasksBySlug to taskqueue.Producer + asynqqueue implementation - Add AudioObjectKeyExt to bookstore.AudioStore for format-aware MinIO keys Co-Authored-By: Claude Sonnet 4.6 --- backend/cmd/backend/main.go | 4 + backend/cmd/runner/main.go | 4 + backend/internal/asynqqueue/producer.go | 6 + backend/internal/backend/handlers.go | 229 +++++++++++++++++-- backend/internal/backend/server.go | 5 + backend/internal/bookstore/bookstore.go | 9 +- backend/internal/bookstore/bookstore_test.go | 7 +- backend/internal/kokoro/client.go | 46 ++++ backend/internal/pockettts/client.go | 25 ++ backend/internal/runner/runner_test.go | 11 + backend/internal/storage/minio.go | 11 +- backend/internal/storage/store.go | 26 +++ backend/internal/taskqueue/taskqueue.go | 4 + backend/internal/taskqueue/taskqueue_test.go | 3 +- 14 files changed, 363 insertions(+), 27 deletions(-) diff --git a/backend/cmd/backend/main.go b/backend/cmd/backend/main.go index 9b67acf..032e9b4 100644 --- a/backend/cmd/backend/main.go +++ b/backend/cmd/backend/main.go @@ -200,6 +200,10 @@ func (n *noopKokoro) StreamAudioMP3(_ context.Context, _, _ string) (io.ReadClos return nil, fmt.Errorf("kokoro not configured (KOKORO_URL is empty)") } +func (n *noopKokoro) StreamAudioWAV(_ context.Context, _, _ string) (io.ReadCloser, error) { + return nil, fmt.Errorf("kokoro not configured (KOKORO_URL is empty)") +} + func (n *noopKokoro) ListVoices(_ context.Context) ([]string, error) { return nil, nil } diff --git a/backend/cmd/runner/main.go b/backend/cmd/runner/main.go index 8b74f0a..141cf07 100644 --- a/backend/cmd/runner/main.go +++ b/backend/cmd/runner/main.go @@ -227,6 +227,10 @@ func (n *noopKokoro) StreamAudioMP3(_ context.Context, _, _ string) (io.ReadClos return nil, fmt.Errorf("kokoro not configured (KOKORO_URL is empty)") } +func (n *noopKokoro) StreamAudioWAV(_ context.Context, _, _ string) (io.ReadCloser, error) { + return nil, fmt.Errorf("kokoro not configured (KOKORO_URL is empty)") +} + func (n *noopKokoro) ListVoices(_ context.Context) ([]string, error) { return nil, nil } diff --git a/backend/internal/asynqqueue/producer.go b/backend/internal/asynqqueue/producer.go index 8fe8688..5e6ba66 100644 --- a/backend/internal/asynqqueue/producer.go +++ b/backend/internal/asynqqueue/producer.go @@ -93,6 +93,12 @@ 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) diff --git a/backend/internal/backend/handlers.go b/backend/internal/backend/handlers.go index df892a6..2376bab 100644 --- a/backend/internal/backend/handlers.go +++ b/backend/internal/backend/handlers.go @@ -708,12 +708,17 @@ func (s *Server) handleAudioProxy(w http.ResponseWriter, r *http.Request) { // Fast path: if audio already exists in MinIO, redirects to the presigned URL // (same as handleAudioProxy) — the client plays from storage immediately. // -// Slow path (first request): streams MP3 audio directly to the client while -// simultaneously uploading it to MinIO. After the stream completes, any -// pending audio_jobs task for this key is marked done. Subsequent requests hit -// the fast path and skip TTS generation entirely. +// Slow path (first request): streams audio directly to the client while +// simultaneously uploading it to MinIO. After the stream completes, subsequent +// requests hit the fast path and skip TTS generation entirely. // -// Query params: voice (optional, defaults to DefaultVoice) +// Query params: +// +// voice (optional, defaults to DefaultVoice) +// format (optional, "mp3" or "wav"; defaults to "mp3") +// +// Using format=wav skips the ffmpeg transcode for pocket-tts voices, delivering +// raw WAV frames to the client with lower latency at the cost of larger files. func (s *Server) handleAudioStream(w http.ResponseWriter, r *http.Request) { slug := r.PathValue("slug") n, err := strconv.Atoi(r.PathValue("n")) @@ -727,7 +732,17 @@ func (s *Server) handleAudioStream(w http.ResponseWriter, r *http.Request) { voice = s.cfg.DefaultVoice } - audioKey := s.deps.AudioStore.AudioObjectKey(slug, n, voice) + format := r.URL.Query().Get("format") + if format != "wav" { + format = "mp3" + } + + contentType := "audio/mpeg" + if format == "wav" { + contentType = "audio/wav" + } + + audioKey := s.deps.AudioStore.AudioObjectKeyExt(slug, n, voice, format) // ── Fast path: already in MinIO ─────────────────────────────────────────── if s.deps.AudioStore.AudioExists(r.Context(), audioKey) { @@ -756,23 +771,39 @@ func (s *Server) handleAudioStream(w http.ResponseWriter, r *http.Request) { return } - // Open the TTS stream. + // Open the TTS stream (WAV or MP3 depending on format param). var audioStream io.ReadCloser - if pockettts.IsPocketTTSVoice(voice) { - if s.deps.PocketTTS == nil { - jsonError(w, http.StatusServiceUnavailable, "pocket-tts not configured") - return + if format == "wav" { + if pockettts.IsPocketTTSVoice(voice) { + if s.deps.PocketTTS == nil { + jsonError(w, http.StatusServiceUnavailable, "pocket-tts not configured") + return + } + audioStream, err = s.deps.PocketTTS.StreamAudioWAV(r.Context(), text, voice) + } else { + if s.deps.Kokoro == nil { + jsonError(w, http.StatusServiceUnavailable, "kokoro not configured") + return + } + audioStream, err = s.deps.Kokoro.StreamAudioWAV(r.Context(), text, voice) } - audioStream, err = s.deps.PocketTTS.StreamAudioMP3(r.Context(), text, voice) } else { - if s.deps.Kokoro == nil { - jsonError(w, http.StatusServiceUnavailable, "kokoro not configured") - return + if pockettts.IsPocketTTSVoice(voice) { + if s.deps.PocketTTS == nil { + jsonError(w, http.StatusServiceUnavailable, "pocket-tts not configured") + return + } + audioStream, err = s.deps.PocketTTS.StreamAudioMP3(r.Context(), text, voice) + } else { + if s.deps.Kokoro == nil { + jsonError(w, http.StatusServiceUnavailable, "kokoro not configured") + return + } + audioStream, err = s.deps.Kokoro.StreamAudioMP3(r.Context(), text, voice) } - audioStream, err = s.deps.Kokoro.StreamAudioMP3(r.Context(), text, voice) } if err != nil { - s.deps.Log.Error("handleAudioStream: TTS stream failed", "slug", slug, "n", n, "voice", voice, "err", err) + s.deps.Log.Error("handleAudioStream: TTS stream failed", "slug", slug, "n", n, "voice", voice, "format", format, "err", err) jsonError(w, http.StatusInternalServerError, "tts stream failed") return } @@ -787,11 +818,11 @@ func (s *Server) handleAudioStream(w http.ResponseWriter, r *http.Request) { go func() { uploadDone <- s.deps.AudioStore.PutAudioStream( context.Background(), // use background — request ctx may cancel after client disconnects - audioKey, pr, -1, "audio/mpeg", + audioKey, pr, -1, contentType, ) }() - w.Header().Set("Content-Type", "audio/mpeg") + w.Header().Set("Content-Type", contentType) w.Header().Set("Cache-Control", "no-store") w.Header().Set("X-Accel-Buffering", "no") // disable nginx/caddy buffering w.WriteHeader(http.StatusOK) @@ -1081,6 +1112,166 @@ func (s *Server) handleAdminTranslationBulk(w http.ResponseWriter, r *http.Reque }) } +// ── Admin Audio ──────────────────────────────────────────────────────────────── + +// handleAdminAudioJobs handles GET /api/admin/audio/jobs. +// Returns all audio jobs, optionally filtered by slug (?slug=...). +// Sorted by started descending. +func (s *Server) handleAdminAudioJobs(w http.ResponseWriter, r *http.Request) { + tasks, err := s.deps.TaskReader.ListAudioTasks(r.Context()) + if err != nil { + s.deps.Log.Error("handleAdminAudioJobs: ListAudioTasks failed", "err", err) + jsonError(w, http.StatusInternalServerError, "failed to list audio jobs") + return + } + + // Optional slug filter. + slugFilter := r.URL.Query().Get("slug") + + type jobRow struct { + ID string `json:"id"` + CacheKey string `json:"cache_key"` + Slug string `json:"slug"` + Chapter int `json:"chapter"` + Voice string `json:"voice"` + Status string `json:"status"` + WorkerID string `json:"worker_id"` + ErrorMessage string `json:"error_message"` + Started string `json:"started"` + Finished string `json:"finished"` + } + rows := make([]jobRow, 0, len(tasks)) + for _, t := range tasks { + if slugFilter != "" && t.Slug != slugFilter { + continue + } + rows = append(rows, jobRow{ + ID: t.ID, + CacheKey: t.CacheKey, + Slug: t.Slug, + Chapter: t.Chapter, + Voice: t.Voice, + Status: string(t.Status), + WorkerID: t.WorkerID, + ErrorMessage: t.ErrorMessage, + Started: t.Started.Format(time.RFC3339), + Finished: t.Finished.Format(time.RFC3339), + }) + } + writeJSON(w, 0, map[string]any{"jobs": rows, "total": len(rows)}) +} + +// handleAdminAudioBulk handles POST /api/admin/audio/bulk. +// Body: {"slug": "...", "voice": "af_bella", "from": 1, "to": 100, "skip_existing": true} +// +// Enqueues one audio task per chapter in [from, to]. +// skip_existing (default true): skip chapters already cached in MinIO — use this +// to resume a previously interrupted bulk job. +// force: if true, enqueue even when a pending/running task already exists. +// Max 1000 chapters per request. +func (s *Server) handleAdminAudioBulk(w http.ResponseWriter, r *http.Request) { + var body struct { + Slug string `json:"slug"` + Voice string `json:"voice"` + From int `json:"from"` + To int `json:"to"` + SkipExisting *bool `json:"skip_existing"` // pointer so we can detect omission + Force bool `json:"force"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + jsonError(w, http.StatusBadRequest, "invalid JSON body") + return + } + if body.Slug == "" { + jsonError(w, http.StatusBadRequest, "slug is required") + return + } + if body.Voice == "" { + body.Voice = s.cfg.DefaultVoice + } + if body.From < 1 || body.To < body.From { + jsonError(w, http.StatusBadRequest, "from must be >= 1 and to must be >= from") + return + } + if body.To-body.From > 999 { + jsonError(w, http.StatusBadRequest, "range too large; max 1000 chapters per request") + return + } + + // skip_existing defaults to true (resume-friendly). + skipExisting := true + if body.SkipExisting != nil { + skipExisting = *body.SkipExisting + } + + var taskIDs []string + skipped := 0 + + for n := body.From; n <= body.To; n++ { + // Skip chapters already cached in MinIO. + if skipExisting { + audioKey := s.deps.AudioStore.AudioObjectKey(body.Slug, n, body.Voice) + if s.deps.AudioStore.AudioExists(r.Context(), audioKey) { + skipped++ + continue + } + } + + // Skip chapters with an active (pending/running) task unless force=true. + if !body.Force { + cacheKey := fmt.Sprintf("%s/%d/%s", body.Slug, n, body.Voice) + existing, found, _ := s.deps.TaskReader.GetAudioTask(r.Context(), cacheKey) + if found && (existing.Status == domain.TaskStatusPending || existing.Status == domain.TaskStatusRunning) { + skipped++ + continue + } + } + + id, err := s.deps.Producer.CreateAudioTask(r.Context(), body.Slug, n, body.Voice) + if err != nil { + s.deps.Log.Error("handleAdminAudioBulk: CreateAudioTask failed", + "slug", body.Slug, "chapter", n, "voice", body.Voice, "err", err) + jsonError(w, http.StatusInternalServerError, + fmt.Sprintf("failed to create task for chapter %d: %s", n, err)) + return + } + taskIDs = append(taskIDs, id) + } + + writeJSON(w, http.StatusAccepted, map[string]any{ + "enqueued": len(taskIDs), + "skipped": skipped, + "task_ids": taskIDs, + }) +} + +// handleAdminAudioCancelBulk handles POST /api/admin/audio/cancel-bulk. +// Body: {"slug": "..."} +// Cancels all pending and running audio tasks for the given slug. +func (s *Server) handleAdminAudioCancelBulk(w http.ResponseWriter, r *http.Request) { + var body struct { + Slug string `json:"slug"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + jsonError(w, http.StatusBadRequest, "invalid JSON body") + return + } + if body.Slug == "" { + jsonError(w, http.StatusBadRequest, "slug is required") + return + } + + cancelled, err := s.deps.Producer.CancelAudioTasksBySlug(r.Context(), body.Slug) + if err != nil { + s.deps.Log.Error("handleAdminAudioCancelBulk: CancelAudioTasksBySlug failed", + "slug", body.Slug, "err", err) + jsonError(w, http.StatusInternalServerError, "failed to cancel tasks") + return + } + + writeJSON(w, 0, map[string]any{"cancelled": cancelled}) +} + // ── Voices ───────────────────────────────────────────────────────────────────── // Returns {"voices": [...]} — merged list from Kokoro and pocket-tts. func (s *Server) handleVoices(w http.ResponseWriter, r *http.Request) { diff --git a/backend/internal/backend/server.go b/backend/internal/backend/server.go index cf7c77a..696f79d 100644 --- a/backend/internal/backend/server.go +++ b/backend/internal/backend/server.go @@ -174,6 +174,11 @@ func (s *Server) ListenAndServe(ctx context.Context) error { mux.HandleFunc("GET /api/admin/translation/jobs", s.handleAdminTranslationJobs) mux.HandleFunc("POST /api/admin/translation/bulk", s.handleAdminTranslationBulk) + // Admin audio endpoints + mux.HandleFunc("GET /api/admin/audio/jobs", s.handleAdminAudioJobs) + mux.HandleFunc("POST /api/admin/audio/bulk", s.handleAdminAudioBulk) + mux.HandleFunc("POST /api/admin/audio/cancel-bulk", s.handleAdminAudioCancelBulk) + // Voices list mux.HandleFunc("GET /api/voices", s.handleVoices) diff --git a/backend/internal/bookstore/bookstore.go b/backend/internal/bookstore/bookstore.go index 1801127..2b212be 100644 --- a/backend/internal/bookstore/bookstore.go +++ b/backend/internal/bookstore/bookstore.go @@ -80,9 +80,14 @@ type RankingStore interface { // AudioStore covers audio object storage (runner writes; backend reads). type AudioStore interface { - // AudioObjectKey returns the MinIO object key for a cached audio file. + // AudioObjectKey returns the MinIO object key for a cached MP3 audio file. + // Format: {slug}/{n}/{voice}.mp3 AudioObjectKey(slug string, n int, voice string) string + // AudioObjectKeyExt returns the MinIO object key for a cached audio file + // with a custom extension (e.g. "mp3" or "wav"). + AudioObjectKeyExt(slug string, n int, voice, ext string) string + // AudioExists returns true when the audio object is present in MinIO. AudioExists(ctx context.Context, key string) bool @@ -91,7 +96,7 @@ type AudioStore interface { // PutAudioStream uploads audio from r to MinIO under key. // size must be the exact byte length of r, or -1 to use multipart upload. - // contentType should be "audio/mpeg". + // contentType should be "audio/mpeg" or "audio/wav". PutAudioStream(ctx context.Context, key string, r io.Reader, size int64, contentType string) error } diff --git a/backend/internal/bookstore/bookstore_test.go b/backend/internal/bookstore/bookstore_test.go index 9b2d7a1..d56909d 100644 --- a/backend/internal/bookstore/bookstore_test.go +++ b/backend/internal/bookstore/bookstore_test.go @@ -52,9 +52,10 @@ func (m *mockStore) RankingFreshEnough(_ context.Context, _ time.Duration) (bool } // AudioStore -func (m *mockStore) AudioObjectKey(_ string, _ int, _ string) string { return "" } -func (m *mockStore) AudioExists(_ context.Context, _ string) bool { return false } -func (m *mockStore) PutAudio(_ context.Context, _ string, _ []byte) error { return nil } +func (m *mockStore) AudioObjectKey(_ string, _ int, _ string) string { return "" } +func (m *mockStore) AudioObjectKeyExt(_ string, _ int, _, _ string) string { return "" } +func (m *mockStore) AudioExists(_ context.Context, _ string) bool { return false } +func (m *mockStore) PutAudio(_ context.Context, _ string, _ []byte) error { return nil } func (m *mockStore) PutAudioStream(_ context.Context, _ string, _ io.Reader, _ int64, _ string) error { return nil } diff --git a/backend/internal/kokoro/client.go b/backend/internal/kokoro/client.go index a309ac6..6895385 100644 --- a/backend/internal/kokoro/client.go +++ b/backend/internal/kokoro/client.go @@ -27,6 +27,11 @@ type Client interface { // waiting for the full output. The caller must always close the ReadCloser. StreamAudioMP3(ctx context.Context, text, voice string) (io.ReadCloser, error) + // StreamAudioWAV synthesises text and returns an io.ReadCloser that streams + // WAV-encoded audio incrementally using kokoro-fastapi's streaming mode with + // response_format:"wav". The caller must always close the ReadCloser. + StreamAudioWAV(ctx context.Context, text, voice string) (io.ReadCloser, error) + // ListVoices returns the available voice IDs. Falls back to an empty slice // on error — callers should treat an empty list as "service unavailable". ListVoices(ctx context.Context) ([]string, error) @@ -167,6 +172,47 @@ func (c *httpClient) StreamAudioMP3(ctx context.Context, text, voice string) (io return resp.Body, nil } +// StreamAudioWAV calls POST /v1/audio/speech with stream:true and response_format:wav, +// returning an io.ReadCloser that delivers WAV bytes as kokoro generates them. +func (c *httpClient) StreamAudioWAV(ctx context.Context, text, voice string) (io.ReadCloser, error) { + if text == "" { + return nil, fmt.Errorf("kokoro: empty text") + } + if voice == "" { + voice = "af_bella" + } + + reqBody, err := json.Marshal(map[string]any{ + "model": "kokoro", + "input": text, + "voice": voice, + "response_format": "wav", + "speed": 1.0, + "stream": true, + }) + if err != nil { + return nil, fmt.Errorf("kokoro: marshal wav stream request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + c.baseURL+"/v1/audio/speech", bytes.NewReader(reqBody)) + if err != nil { + return nil, fmt.Errorf("kokoro: build wav stream request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.http.Do(req) + if err != nil { + return nil, fmt.Errorf("kokoro: wav stream request: %w", err) + } + if resp.StatusCode != http.StatusOK { + _, _ = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + return nil, fmt.Errorf("kokoro: wav stream returned %d", resp.StatusCode) + } + return resp.Body, nil +} + // ListVoices calls GET /v1/audio/voices and returns the list of voice IDs. func (c *httpClient) ListVoices(ctx context.Context) ([]string, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, diff --git a/backend/internal/pockettts/client.go b/backend/internal/pockettts/client.go index 4c3a542..52bab1f 100644 --- a/backend/internal/pockettts/client.go +++ b/backend/internal/pockettts/client.go @@ -59,6 +59,12 @@ type Client interface { // The caller must always close the returned ReadCloser. StreamAudioMP3(ctx context.Context, text, voice string) (io.ReadCloser, error) + // StreamAudioWAV synthesises text and returns an io.ReadCloser that streams + // raw WAV audio directly from pocket-tts without any transcoding. + // The stream begins with a WAV header followed by 16-bit PCM frames at 16 kHz. + // The caller must always close the returned ReadCloser. + StreamAudioWAV(ctx context.Context, text, voice string) (io.ReadCloser, error) + // ListVoices returns the available predefined voice names. ListVoices(ctx context.Context) ([]string, error) } @@ -160,6 +166,25 @@ func (c *httpClient) StreamAudioMP3(ctx context.Context, text, voice string) (io return pr, nil } +// StreamAudioWAV posts to POST /tts and returns an io.ReadCloser that delivers +// raw WAV bytes directly from pocket-tts — no ffmpeg transcoding required. +// The first bytes will be a WAV header (RIFF/fmt chunk) followed by PCM frames. +// The caller must always close the returned ReadCloser. +func (c *httpClient) StreamAudioWAV(ctx context.Context, text, voice string) (io.ReadCloser, error) { + if text == "" { + return nil, fmt.Errorf("pockettts: empty text") + } + if voice == "" { + voice = "alba" + } + + resp, err := c.postTTS(ctx, text, voice) + if err != nil { + return nil, err + } + return resp.Body, nil +} + // ListVoices returns the statically known predefined voice names. // pocket-tts has no REST endpoint for listing voices. func (c *httpClient) ListVoices(_ context.Context) ([]string, error) { diff --git a/backend/internal/runner/runner_test.go b/backend/internal/runner/runner_test.go index cc3d590..848625a 100644 --- a/backend/internal/runner/runner_test.go +++ b/backend/internal/runner/runner_test.go @@ -126,6 +126,9 @@ type stubAudioStore struct { func (s *stubAudioStore) AudioObjectKey(slug string, n int, voice string) string { return slug + "/" + string(rune('0'+n)) + "/" + voice + ".mp3" } +func (s *stubAudioStore) AudioObjectKeyExt(slug string, n int, voice, ext string) string { + return slug + "/" + string(rune('0'+n)) + "/" + voice + "." + ext +} func (s *stubAudioStore) AudioExists(_ context.Context, _ string) bool { return false } func (s *stubAudioStore) PutAudio(_ context.Context, _ string, _ []byte) error { s.putCalled.Add(1) @@ -199,6 +202,14 @@ func (s *stubKokoro) StreamAudioMP3(_ context.Context, _, _ string) (io.ReadClos return io.NopCloser(bytes.NewReader(s.data)), nil } +func (s *stubKokoro) StreamAudioWAV(_ context.Context, _, _ string) (io.ReadCloser, error) { + s.called.Add(1) + if s.genErr != nil { + return nil, s.genErr + } + return io.NopCloser(bytes.NewReader(s.data)), nil +} + func (s *stubKokoro) ListVoices(_ context.Context) ([]string, error) { return []string{"af_bella"}, nil } diff --git a/backend/internal/storage/minio.go b/backend/internal/storage/minio.go index 994be25..0898c9e 100644 --- a/backend/internal/storage/minio.go +++ b/backend/internal/storage/minio.go @@ -109,10 +109,17 @@ func ChapterObjectKey(slug string, n int) string { return fmt.Sprintf("%s/chapter-%06d.md", slug, n) } -// AudioObjectKey returns the MinIO object key for a cached audio file. +// AudioObjectKeyExt returns the MinIO object key for a cached audio file +// with a custom extension (e.g. "mp3" or "wav"). +// Format: {slug}/{n}/{voice}.{ext} +func AudioObjectKeyExt(slug string, n int, voice, ext string) string { + return fmt.Sprintf("%s/%d/%s.%s", slug, n, voice, ext) +} + +// AudioObjectKey returns the MinIO object key for a cached MP3 audio file. // Format: {slug}/{n}/{voice}.mp3 func AudioObjectKey(slug string, n int, voice string) string { - return fmt.Sprintf("%s/%d/%s.mp3", slug, n, voice) + return AudioObjectKeyExt(slug, n, voice, "mp3") } // AvatarObjectKey returns the MinIO object key for a user avatar image. diff --git a/backend/internal/storage/store.go b/backend/internal/storage/store.go index 9ed2d00..ad789c1 100644 --- a/backend/internal/storage/store.go +++ b/backend/internal/storage/store.go @@ -376,6 +376,10 @@ func (s *Store) AudioObjectKey(slug string, n int, voice string) string { return AudioObjectKey(slug, n, voice) } +func (s *Store) AudioObjectKeyExt(slug string, n int, voice, ext string) string { + return AudioObjectKeyExt(slug, n, voice, ext) +} + func (s *Store) AudioExists(ctx context.Context, key string) bool { return s.mc.objectExists(ctx, s.mc.bucketAudio, key) } @@ -574,6 +578,28 @@ func (s *Store) CancelTask(ctx context.Context, id string) error { map[string]string{"status": string(domain.TaskStatusCancelled)}) } +func (s *Store) CancelAudioTasksBySlug(ctx context.Context, slug string) (int, error) { + filter := fmt.Sprintf(`slug='%s'&&(status='pending'||status='running')`, slug) + items, err := s.pb.listAll(ctx, "audio_jobs", filter, "") + if err != nil { + return 0, fmt.Errorf("CancelAudioTasksBySlug list: %w", err) + } + cancelled := 0 + for _, raw := range items { + var rec struct { + ID string `json:"id"` + } + if json.Unmarshal(raw, &rec) == nil && rec.ID != "" { + if patchErr := s.pb.patch(ctx, + fmt.Sprintf("/api/collections/audio_jobs/records/%s", rec.ID), + map[string]string{"status": string(domain.TaskStatusCancelled)}); patchErr == nil { + cancelled++ + } + } + } + return cancelled, nil +} + // ── taskqueue.Consumer ──────────────────────────────────────────────────────── func (s *Store) ClaimNextScrapeTask(ctx context.Context, workerID string) (domain.ScrapeTask, bool, error) { diff --git a/backend/internal/taskqueue/taskqueue.go b/backend/internal/taskqueue/taskqueue.go index 05dca6e..b92feb3 100644 --- a/backend/internal/taskqueue/taskqueue.go +++ b/backend/internal/taskqueue/taskqueue.go @@ -36,6 +36,10 @@ type Producer interface { // CancelTask transitions a pending task to status=cancelled. // Returns ErrNotFound if the task does not exist. CancelTask(ctx context.Context, id string) error + + // CancelAudioTasksBySlug cancels all pending or running audio tasks for slug. + // Returns the number of tasks cancelled. + CancelAudioTasksBySlug(ctx context.Context, slug string) (int, error) } // Consumer is the read/claim side of the task queue used by the runner. diff --git a/backend/internal/taskqueue/taskqueue_test.go b/backend/internal/taskqueue/taskqueue_test.go index e2372d3..4c014c2 100644 --- a/backend/internal/taskqueue/taskqueue_test.go +++ b/backend/internal/taskqueue/taskqueue_test.go @@ -26,7 +26,8 @@ func (s *stubStore) CreateAudioTask(_ context.Context, _ string, _ int, _ string func (s *stubStore) CreateTranslationTask(_ context.Context, _ string, _ int, _ string) (string, error) { return "translation-1", nil } -func (s *stubStore) CancelTask(_ context.Context, _ string) error { return nil } +func (s *stubStore) CancelTask(_ context.Context, _ string) error { return nil } +func (s *stubStore) CancelAudioTasksBySlug(_ context.Context, _ string) (int, error) { return 0, nil } func (s *stubStore) ClaimNextScrapeTask(_ context.Context, _ string) (domain.ScrapeTask, bool, error) { return domain.ScrapeTask{ID: "task-1", Status: domain.TaskStatusRunning}, true, nil