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