feat(audio): streaming TTS provider interface + voice cache + ElevenLabs models

Add StreamingTTSProvider interface for audio streaming support. Implement voice
caching with TTL+LRU. Add ElevenLabs model validation and voice list retrieval.
This commit is contained in:
viettranx committed 2026-04-15 11:24:56 +07:00
1 parent 0760c3664f
commit daa5adb88b
11 files changed
+604 -1

No files matched your search

+29
View File
@@ -62,6 +62,35 @@ func (c *client) postJSON(ctx context.Context, path string, body []byte, customT
return out, nil
}
// getJSON performs a GET request to {baseURL}/path and returns the raw response
// bytes on 200 OK. Non-200 responses surface as errors with a truncated body —
// mirrors the postJSON error handling pattern.
func (c *client) getJSON(ctx context.Context, path string) ([]byte, error) {
url := strings.TrimRight(c.baseURL, "/") + path
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return nil, fmt.Errorf("create request: %w", err)
}
req.Header.Set("xi-api-key", c.apiKey)
hc := &http.Client{Timeout: time.Duration(c.timeoutMs) * time.Millisecond}
resp, err := hc.Do(req)
if err != nil {
return nil, fmt.Errorf("http request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
errBody, _ := io.ReadAll(resp.Body)
return nil, fmt.Errorf("ElevenLabs API error %d: %s", resp.StatusCode, truncate(errBody, 500))
}
out, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("read response: %w", err)
}
return out, nil
}
func truncate(b []byte, n int) string {
if len(b) <= n {
return string(b)
+35
View File
@@ -0,0 +1,35 @@
package elevenlabs
import (
"errors"
"github.com/nextlevelbuilder/goclaw/internal/i18n"
)
// AllowedElevenLabsModels is the allowlist of ElevenLabs TTS model IDs the
// gateway will forward. Keeping this explicit prevents operators from pasting
// model aliases that silently fall back to the provider's default model.
//
// Refs: https://elevenlabs.io/docs/api-reference/models
var AllowedElevenLabsModels = map[string]struct{}{
"eleven_v3": {}, // latest flagship (Nov 2025)
"eleven_flash_v2_5": {}, // low-latency, 32 languages
"eleven_multilingual_v2": {}, // high-quality multilingual
"eleven_turbo_v2_5": {}, // cost-optimized, fast
}
// ValidateModel checks that id is on the allowlist. Empty id is treated as
// "use configured default" and returns nil. Unknown ids surface an i18n
// error keyed by i18n.MsgTtsUnknownModel so the gateway can localize the
// message before returning it to the caller.
func ValidateModel(id string) error {
if id == "" {
return nil
}
if _, ok := AllowedElevenLabsModels[id]; ok {
return nil
}
// English fallback — callers with a locale should wrap via i18n.T at the
// boundary. Model id is embedded so operators can spot typos in logs.
return errors.New(i18n.T("en", i18n.MsgTtsUnknownModel, id))
}
+56
View File
@@ -0,0 +1,56 @@
package elevenlabs
import (
"strings"
"testing"
)
func TestValidateModel_AllowsKnownIDs(t *testing.T) {
t.Parallel()
allowed := []string{
"eleven_v3",
"eleven_flash_v2_5",
"eleven_multilingual_v2",
"eleven_turbo_v2_5",
}
for _, id := range allowed {
if err := ValidateModel(id); err != nil {
t.Errorf("ValidateModel(%q) returned error: %v (want nil)", id, err)
}
}
}
func TestValidateModel_RejectsUnknown(t *testing.T) {
t.Parallel()
err := ValidateModel("foo")
if err == nil {
t.Fatal("ValidateModel(\"foo\") returned nil (want i18n-keyed error)")
}
// Error should mention the offending model so operators can spot typos.
if !strings.Contains(err.Error(), "foo") {
t.Errorf("error message should contain model id: %q", err.Error())
}
}
func TestValidateModel_EmptyIsAllowed(t *testing.T) {
t.Parallel()
// Empty string means "use configured default" — never a client error.
if err := ValidateModel(""); err != nil {
t.Errorf("ValidateModel(\"\") returned error: %v (want nil — empty = use default)", err)
}
}
func TestAllowedElevenLabsModels_ContainsExpectedSet(t *testing.T) {
t.Parallel()
want := []string{
"eleven_v3",
"eleven_flash_v2_5",
"eleven_multilingual_v2",
"eleven_turbo_v2_5",
}
for _, id := range want {
if _, ok := AllowedElevenLabsModels[id]; !ok {
t.Errorf("AllowedElevenLabsModels missing %q", id)
}
}
}
+53
View File
@@ -0,0 +1,53 @@
package elevenlabs
import (
"context"
"encoding/json"
"fmt"
"github.com/nextlevelbuilder/goclaw/internal/audio"
)
// voicesResponse mirrors the ElevenLabs GET /v1/voices response envelope.
type voicesResponse struct {
Voices []voiceJSON `json:"voices"`
}
type voiceJSON struct {
VoiceID string `json:"voice_id"`
Name string `json:"name"`
Labels map[string]string `json:"labels"`
PreviewURL string `json:"preview_url"`
Category string `json:"category"`
}
// ListVoices fetches the caller's available voices from ElevenLabs
// GET /v1/voices with xi-api-key auth. Results are NOT cached here;
// caching lives in audio.VoiceCache.
func (p *TTSProvider) ListVoices(ctx context.Context) ([]audio.Voice, error) {
raw, err := p.c.getJSON(ctx, "/v1/voices")
if err != nil {
return nil, fmt.Errorf("elevenlabs list voices: %w", err)
}
var resp voicesResponse
if err := json.Unmarshal(raw, &resp); err != nil {
return nil, fmt.Errorf("elevenlabs list voices: parse response: %w", err)
}
voices := make([]audio.Voice, 0, len(resp.Voices))
for _, v := range resp.Voices {
labels := v.Labels
if labels == nil {
labels = make(map[string]string)
}
voices = append(voices, audio.Voice{
ID: v.VoiceID,
Name: v.Name,
Labels: labels,
PreviewURL: v.PreviewURL,
Category: v.Category,
})
}
return voices, nil
}
+96
View File
@@ -0,0 +1,96 @@
package elevenlabs_test
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"github.com/nextlevelbuilder/goclaw/internal/audio/elevenlabs"
)
func TestListVoices_HappyPath(t *testing.T) {
fixture := map[string]any{
"voices": []map[string]any{
{
"voice_id": "voice1",
"name": "Bella",
"labels": map[string]any{"accent": "american", "gender": "female"},
"preview_url": "https://cdn.example.com/bella.mp3",
"category": "premade",
},
{
"voice_id": "voice2",
"name": "Adam",
"labels": map[string]any{"accent": "british"},
"preview_url": "https://cdn.example.com/adam.mp3",
"category": "cloned",
},
},
}
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/v1/voices" {
t.Errorf("unexpected path: %s", r.URL.Path)
}
if r.Header.Get("xi-api-key") != "test-key" {
t.Errorf("missing or wrong api key header")
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(fixture)
}))
defer srv.Close()
p := elevenlabs.NewTTSProvider(elevenlabs.Config{
APIKey: "test-key",
BaseURL: srv.URL,
})
voices, err := p.ListVoices(t.Context())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if len(voices) != 2 {
t.Fatalf("expected 2 voices, got %d", len(voices))
}
if voices[0].ID != "voice1" || voices[0].Name != "Bella" {
t.Errorf("voice[0] mismatch: %+v", voices[0])
}
if voices[0].Labels["accent"] != "american" {
t.Errorf("voice[0] label accent mismatch: %v", voices[0].Labels)
}
if voices[0].PreviewURL != "https://cdn.example.com/bella.mp3" {
t.Errorf("voice[0] preview url mismatch: %s", voices[0].PreviewURL)
}
if voices[1].ID != "voice2" || voices[1].Category != "cloned" {
t.Errorf("voice[1] mismatch: %+v", voices[1])
}
}
func TestListVoices_Unauthorized(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusUnauthorized)
w.Write([]byte(`{"detail":"invalid_api_key"}`))
}))
defer srv.Close()
p := elevenlabs.NewTTSProvider(elevenlabs.Config{
APIKey: "bad-key",
BaseURL: srv.URL,
})
_, err := p.ListVoices(t.Context())
if err == nil {
t.Fatal("expected error for 401, got nil")
}
// Error must mention the status code or response body
if !containsStr(err.Error(), "401") && !containsStr(err.Error(), "invalid_api_key") {
t.Errorf("expected 401 info in error message, got: %v", err)
}
}
func containsStr(s, sub string) bool {
for i := 0; i <= len(s)-len(sub); i++ {
if s[i:i+len(sub)] == sub {
return true
}
}
return false
}
+15
View File
@@ -124,6 +124,21 @@ func (m *Manager) Synthesize(ctx context.Context, text string, opts TTSOptions)
return p.Synthesize(ctx, text, opts)
}
// SynthesizeStream dispatches streaming TTS to the primary provider. Returns
// ErrStreamingNotSupported if the primary does not implement
// StreamingTTSProvider, letting callers fall back to buffered Synthesize.
func (m *Manager) SynthesizeStream(ctx context.Context, text string, opts TTSOptions) (*StreamResult, error) {
p, ok := m.ttsProviders[m.primary]
if !ok {
return nil, fmt.Errorf("tts provider not found: %s", m.primary)
}
sp, ok := p.(StreamingTTSProvider)
if !ok {
return nil, ErrStreamingNotSupported
}
return sp.SynthesizeStream(ctx, text, opts)
}
// SynthesizeWithFallback tries primary first, then any other registered
// provider on error. Returns first success or aggregate failure.
func (m *Manager) SynthesizeWithFallback(ctx context.Context, text string, opts TTSOptions) (*SynthResult, error) {
+75
View File
@@ -0,0 +1,75 @@
package audio_test
import (
"context"
"errors"
"io"
"strings"
"testing"
"github.com/nextlevelbuilder/goclaw/internal/audio"
)
type bufferedOnlyTTS struct{}
func (bufferedOnlyTTS) Name() string { return "buffered-only" }
func (bufferedOnlyTTS) Synthesize(_ context.Context, _ string, _ audio.TTSOptions) (*audio.SynthResult, error) {
return &audio.SynthResult{Extension: "mp3", MimeType: "audio/mpeg"}, nil
}
type streamingTTS struct{ name string }
func (s streamingTTS) Name() string { return s.name }
func (s streamingTTS) Synthesize(_ context.Context, _ string, _ audio.TTSOptions) (*audio.SynthResult, error) {
return &audio.SynthResult{Extension: "mp3", MimeType: "audio/mpeg"}, nil
}
func (s streamingTTS) SynthesizeStream(_ context.Context, text string, _ audio.TTSOptions) (*audio.StreamResult, error) {
return &audio.StreamResult{
Audio: io.NopCloser(strings.NewReader("STREAM:" + text)),
Extension: "mp3",
MimeType: "audio/mpeg",
}, nil
}
func TestManager_SynthesizeStream_UnsupportedProviderReturnsSentinel(t *testing.T) {
t.Parallel()
mgr := audio.NewManager(audio.ManagerConfig{})
mgr.RegisterTTS(bufferedOnlyTTS{})
_, err := mgr.SynthesizeStream(context.Background(), "hi", audio.TTSOptions{})
if !errors.Is(err, audio.ErrStreamingNotSupported) {
t.Errorf("got err=%v, want ErrStreamingNotSupported", err)
}
}
func TestManager_SynthesizeStream_DispatchesToStreamingProvider(t *testing.T) {
t.Parallel()
mgr := audio.NewManager(audio.ManagerConfig{})
mgr.RegisterTTS(streamingTTS{name: "el"})
res, err := mgr.SynthesizeStream(context.Background(), "hello", audio.TTSOptions{})
if err != nil {
t.Fatalf("SynthesizeStream error: %v", err)
}
defer res.Audio.Close()
body, err := io.ReadAll(res.Audio)
if err != nil {
t.Fatalf("read: %v", err)
}
if got, want := string(body), "STREAM:hello"; got != want {
t.Errorf("body: got %q, want %q", got, want)
}
}
func TestManager_SynthesizeStream_NoPrimaryReturnsError(t *testing.T) {
t.Parallel()
mgr := audio.NewManager(audio.ManagerConfig{Primary: "missing"})
_, err := mgr.SynthesizeStream(context.Background(), "hi", audio.TTSOptions{})
if err == nil {
t.Fatal("expected error when primary not registered")
}
if errors.Is(err, audio.ErrStreamingNotSupported) {
t.Errorf("got ErrStreamingNotSupported, expected provider-not-found")
}
}
+25 -1
View File
@@ -6,7 +6,15 @@
// implementations land in Phase 3 (Music/SFX) and Phase 4 (STT).
package audio
import "context"
import (
"context"
"errors"
"io"
)
// ErrStreamingNotSupported is returned when a caller requests streaming TTS
// on a provider that does not implement StreamingTTSProvider.
var ErrStreamingNotSupported = errors.New("streaming not supported")
// ---- TTS (implemented Phase 1) ----
@@ -16,6 +24,22 @@ type TTSProvider interface {
Synthesize(ctx context.Context, text string, opts TTSOptions) (*SynthResult, error)
}
// StreamingTTSProvider is the optional streaming extension of TTSProvider.
// Providers that support HTTP chunked streaming implement this to return
// bytes as they arrive, reducing time-to-first-byte for long texts.
type StreamingTTSProvider interface {
TTSProvider
SynthesizeStream(ctx context.Context, text string, opts TTSOptions) (*StreamResult, error)
}
// StreamResult is the output of a streaming TTS synthesis call. Callers MUST
// Close the Audio reader when finished to release the underlying HTTP body.
type StreamResult struct {
Audio io.ReadCloser // chunked audio stream — caller must Close
Extension string // file extension without dot: "mp3", "opus", "ogg"
MimeType string // e.g. "audio/mpeg", "audio/ogg"
}
// TTSOptions controls TTS synthesis parameters.
type TTSOptions struct {
Voice string // provider-specific voice ID
+111
View File
@@ -0,0 +1,111 @@
package audio
import (
"container/list"
"sync"
"time"
"github.com/google/uuid"
)
// VoiceCache is a thread-safe in-memory cache of TTS voices keyed by tenant
// UUID. Eviction policy: TTL (entries expire after a fixed duration) combined
// with LRU cap (oldest-accessed entry evicted when cap is exceeded). Both
// policies apply simultaneously — whichever fires first removes the entry.
//
// Multi-instance note: each gateway process maintains its own cache. Redis
// invalidation is deferred until scale-out is on the roadmap (P2-H1).
type VoiceCache struct {
mu sync.Mutex
entries map[uuid.UUID]*list.Element
lru *list.List // front = most recently used
ttl time.Duration
maxTenants int
}
type voiceCacheEntry struct {
tenantID uuid.UUID
voices []Voice
expiresAt time.Time
}
// NewVoiceCache creates a cache with the given TTL and LRU cap.
// ttl=0 means entries never expire; maxTenants=0 disables the LRU cap.
func NewVoiceCache(ttl time.Duration, maxTenants int) *VoiceCache {
return &VoiceCache{
entries: make(map[uuid.UUID]*list.Element),
lru: list.New(),
ttl: ttl,
maxTenants: maxTenants,
}
}
// Get returns the cached voices for tenantID. Returns (nil, false) on miss
// (absent or TTL-expired). Moves the entry to the front of the LRU list on hit.
func (c *VoiceCache) Get(tenantID uuid.UUID) ([]Voice, bool) {
c.mu.Lock()
defer c.mu.Unlock()
el, ok := c.entries[tenantID]
if !ok {
return nil, false
}
entry := el.Value.(*voiceCacheEntry)
if c.ttl > 0 && time.Now().After(entry.expiresAt) {
c.lru.Remove(el)
delete(c.entries, tenantID)
return nil, false
}
c.lru.MoveToFront(el)
return entry.voices, true
}
// Set stores voices for tenantID. Refreshes TTL and LRU position when the
// tenant already has a cached entry. Evicts the LRU entry when at cap.
func (c *VoiceCache) Set(tenantID uuid.UUID, voices []Voice) {
c.mu.Lock()
defer c.mu.Unlock()
if el, ok := c.entries[tenantID]; ok {
entry := el.Value.(*voiceCacheEntry)
entry.voices = voices
if c.ttl > 0 {
entry.expiresAt = time.Now().Add(c.ttl)
}
c.lru.MoveToFront(el)
return
}
if c.maxTenants > 0 && c.lru.Len() >= c.maxTenants {
oldest := c.lru.Back()
if oldest != nil {
evicted := oldest.Value.(*voiceCacheEntry)
c.lru.Remove(oldest)
delete(c.entries, evicted.tenantID)
}
}
var exp time.Time
if c.ttl > 0 {
exp = time.Now().Add(c.ttl)
}
entry := &voiceCacheEntry{
tenantID: tenantID,
voices: voices,
expiresAt: exp,
}
el := c.lru.PushFront(entry)
c.entries[tenantID] = el
}
// Invalidate removes the cache entry for tenantID. No-op if absent.
// Called by POST /v1/voices/refresh to force a live refetch.
func (c *VoiceCache) Invalidate(tenantID uuid.UUID) {
c.mu.Lock()
defer c.mu.Unlock()
if el, ok := c.entries[tenantID]; ok {
c.lru.Remove(el)
delete(c.entries, tenantID)
}
}
+97
View File
@@ -0,0 +1,97 @@
package audio_test
import (
"sync"
"testing"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/audio"
)
func TestVoiceCache_SetGetHit(t *testing.T) {
c := audio.NewVoiceCache(time.Hour, 100)
tid := uuid.New()
voices := []audio.Voice{{ID: "v1", Name: "Bella"}}
c.Set(tid, voices)
got, ok := c.Get(tid)
if !ok {
t.Fatal("expected cache hit")
}
if len(got) != 1 || got[0].ID != "v1" {
t.Errorf("unexpected voices: %+v", got)
}
}
func TestVoiceCache_ExpiredMiss(t *testing.T) {
c := audio.NewVoiceCache(10*time.Millisecond, 100)
tid := uuid.New()
c.Set(tid, []audio.Voice{{ID: "v1"}})
time.Sleep(20 * time.Millisecond)
_, ok := c.Get(tid)
if ok {
t.Fatal("expected cache miss after TTL expiry")
}
}
func TestVoiceCache_LRUEviction(t *testing.T) {
cap := 3
c := audio.NewVoiceCache(time.Hour, cap)
ids := make([]uuid.UUID, cap+1)
for i := range ids {
ids[i] = uuid.New()
}
voices := []audio.Voice{{ID: "x"}}
// Fill to capacity
for i := 0; i < cap; i++ {
c.Set(ids[i], voices)
}
// Access ids[0] to make it recently used
c.Get(ids[0])
// Insert one more — should evict the LRU entry (ids[1] was least recently used)
c.Set(ids[cap], voices)
_, ok := c.Get(ids[1])
if ok {
t.Fatal("ids[1] should have been evicted as LRU")
}
// ids[0] and ids[cap] must still be present
if _, ok := c.Get(ids[0]); !ok {
t.Error("ids[0] should still be cached (recently accessed)")
}
if _, ok := c.Get(ids[cap]); !ok {
t.Error("ids[cap] should be cached (just inserted)")
}
}
func TestVoiceCache_Invalidate(t *testing.T) {
c := audio.NewVoiceCache(time.Hour, 100)
tid := uuid.New()
c.Set(tid, []audio.Voice{{ID: "v1"}})
c.Invalidate(tid)
_, ok := c.Get(tid)
if ok {
t.Fatal("expected miss after invalidate")
}
}
func TestVoiceCache_ConcurrentSafe(t *testing.T) {
c := audio.NewVoiceCache(time.Hour, 500)
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Add(1)
go func() {
defer wg.Done()
tid := uuid.New()
c.Set(tid, []audio.Voice{{ID: "v"}})
c.Get(tid)
c.Invalidate(tid)
}()
}
wg.Wait()
}
+12
View File
@@ -0,0 +1,12 @@
package audio
// Voice represents a single TTS voice as returned by a provider's voice-list
// endpoint (e.g. ElevenLabs GET /v1/voices). Kept provider-agnostic so HTTP/WS
// handlers can depend on the audio package without importing sub-providers.
type Voice struct {
ID string `json:"voice_id"`
Name string `json:"name"`
Labels map[string]string `json:"labels"`
PreviewURL string `json:"preview_url"`
Category string `json:"category"` // "premade" | "cloned" | "generated"
}