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 <noreply@anthropic.com>

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
 }

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
 }

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)

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) {

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)
 

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
 }
 

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
 }

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,

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) {

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
 }

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.

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) {

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.

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

ui/src/routes/api/webhooks/polar/+server.ts

@@ -42,11 +42,11 @@ export const POST: RequestHandler = async ({ request }) => {
 			case 'subscription.updated':
 			case 'subscription.canceled':
 			case 'subscription.revoked':
-				await handleSubscriptionEvent(type, data as Parameters<typeof handleSubscriptionEvent>[1]);
+				await handleSubscriptionEvent(type, data as unknown as Parameters<typeof handleSubscriptionEvent>[1]);
 				break;
 
 			case 'order.created':
-				await handleOrderCreated(data as Parameters<typeof handleOrderCreated>[0]);
+				await handleOrderCreated(data as unknown as Parameters<typeof handleOrderCreated>[0]);
 				break;
 
 			default: