From 24833194db14bc7d828ee96454596928d8df620f Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 3 May 2026 15:29:34 -0500 Subject: [PATCH] Implement WhisperX HTTP adapter --- internal/adapters/whisperx/client.go | 8 +- internal/adapters/whisperx/fake.go | 5 + internal/adapters/whisperx/http.go | 320 ++++++++++++++++++++++++ internal/adapters/whisperx/http_test.go | 271 ++++++++++++++++++++ 4 files changed, 603 insertions(+), 1 deletion(-) create mode 100644 internal/adapters/whisperx/http.go create mode 100644 internal/adapters/whisperx/http_test.go diff --git a/internal/adapters/whisperx/client.go b/internal/adapters/whisperx/client.go index 40925f8..9413f49 100644 --- a/internal/adapters/whisperx/client.go +++ b/internal/adapters/whisperx/client.go @@ -1,7 +1,10 @@ // Package whisperx declares the adapter contract for WhisperX transcription. package whisperx -import "context" +import ( + "context" + "time" +) // TODO: implement a real WhisperX HTTP client adapter. @@ -20,5 +23,8 @@ type TranscribeRequest struct { // TranscribeResult describes the transcript output and adapter metadata. type TranscribeResult struct { OutputRawTranscriptPath string + Attempts int + HTTPStatus int + Duration time.Duration Metadata map[string]any } diff --git a/internal/adapters/whisperx/fake.go b/internal/adapters/whisperx/fake.go index 885b9f9..bd8c2bf 100644 --- a/internal/adapters/whisperx/fake.go +++ b/internal/adapters/whisperx/fake.go @@ -12,6 +12,8 @@ func (n *NoopClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra } return TranscribeResult{ OutputRawTranscriptPath: req.OutputRawTranscriptPath, + Attempts: 1, + HTTPStatus: 200, Metadata: map[string]any{ "placeholder": true, }, @@ -38,6 +40,9 @@ func (f *FakeClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra if res.OutputRawTranscriptPath == "" { res.OutputRawTranscriptPath = req.OutputRawTranscriptPath } + if res.Attempts == 0 { + res.Attempts = 1 + } if res.Metadata == nil { res.Metadata = map[string]any{"fake": true} } diff --git a/internal/adapters/whisperx/http.go b/internal/adapters/whisperx/http.go new file mode 100644 index 0000000..72e0b33 --- /dev/null +++ b/internal/adapters/whisperx/http.go @@ -0,0 +1,320 @@ +package whisperx + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "mime/multipart" + "net" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "time" +) + +const defaultMaxResponseBytes int64 = 10 * 1024 * 1024 + +// HTTPClientConfig contains parsed, deterministic WhisperX HTTP client settings. +type HTTPClientConfig struct { + TranscribeURL string + Language string + Timeout time.Duration + Retries int + RetryDelay time.Duration + HTTPClient *http.Client + MaxResponseBytes int64 +} + +// HTTPClient is a real WhisperX HTTP adapter implementation. +type HTTPClient struct { + url *url.URL + language string + timeout time.Duration + retries int + retryDelay time.Duration + httpClient *http.Client + maxResponseBytes int64 +} + +// NewHTTPClientFromConfigValues builds a client from config values and parses durations once. +func NewHTTPClientFromConfigValues(transcribeURL, language, timeout, retryDelay string, retries int) (*HTTPClient, error) { + if strings.TrimSpace(timeout) == "" { + return nil, fmt.Errorf("whisperx timeout is required") + } + parsedTimeout, err := time.ParseDuration(timeout) + if err != nil { + return nil, fmt.Errorf("parse whisperx timeout %q: %w", timeout, err) + } + if strings.TrimSpace(retryDelay) == "" { + return nil, fmt.Errorf("whisperx retry_delay is required") + } + parsedRetryDelay, err := time.ParseDuration(retryDelay) + if err != nil { + return nil, fmt.Errorf("parse whisperx retry_delay %q: %w", retryDelay, err) + } + return NewHTTPClient(HTTPClientConfig{ + TranscribeURL: transcribeURL, + Language: language, + Timeout: parsedTimeout, + Retries: retries, + RetryDelay: parsedRetryDelay, + }) +} + +// NewHTTPClient constructs an HTTP WhisperX client with validated settings. +func NewHTTPClient(cfg HTTPClientConfig) (*HTTPClient, error) { + if strings.TrimSpace(cfg.TranscribeURL) == "" { + return nil, fmt.Errorf("whisperx transcribe_url is required") + } + u, err := url.Parse(cfg.TranscribeURL) + if err != nil || u.Scheme == "" || u.Host == "" { + if err != nil { + return nil, fmt.Errorf("invalid whisperx transcribe_url %q: %w", cfg.TranscribeURL, err) + } + return nil, fmt.Errorf("invalid whisperx transcribe_url %q", cfg.TranscribeURL) + } + if cfg.Timeout <= 0 { + return nil, fmt.Errorf("whisperx timeout must be > 0") + } + if cfg.Retries < 0 { + return nil, fmt.Errorf("whisperx retries must be >= 0") + } + if cfg.RetryDelay < 0 { + return nil, fmt.Errorf("whisperx retry_delay must be >= 0") + } + if strings.TrimSpace(cfg.Language) == "" { + return nil, fmt.Errorf("whisperx language is required") + } + + client := cfg.HTTPClient + if client == nil { + client = &http.Client{} + } + + maxBytes := cfg.MaxResponseBytes + if maxBytes <= 0 { + maxBytes = defaultMaxResponseBytes + } + + return &HTTPClient{ + url: u, + language: cfg.Language, + timeout: cfg.Timeout, + retries: cfg.Retries, + retryDelay: cfg.RetryDelay, + httpClient: client, + maxResponseBytes: maxBytes, + }, nil +} + +// Transcribe submits one audio file to WhisperX and writes validated JSON output atomically. +func (c *HTTPClient) Transcribe(ctx context.Context, req TranscribeRequest) (TranscribeResult, error) { + if c == nil { + return TranscribeResult{}, fmt.Errorf("whisperx http client is nil") + } + if strings.TrimSpace(req.AudioPath) == "" { + return TranscribeResult{}, fmt.Errorf("whisperx transcribe request audio path is required") + } + if strings.TrimSpace(req.OutputRawTranscriptPath) == "" { + return TranscribeResult{}, fmt.Errorf("whisperx transcribe request output path is required") + } + + start := time.Now() + result := TranscribeResult{ + OutputRawTranscriptPath: req.OutputRawTranscriptPath, + } + attempts := c.retries + 1 + var lastErr error + + for attempt := 1; attempt <= attempts; attempt++ { + result.Attempts = attempt + + attemptCtx, cancel := context.WithTimeout(ctx, c.timeout) + status, body, err := c.doTranscribeAttempt(attemptCtx, req.AudioPath) + cancel() + + if status > 0 { + result.HTTPStatus = status + } + + if err == nil { + if err := c.validateJSON(body); err != nil { + result.Duration = time.Since(start) + return result, fmt.Errorf("whisperx attempt %d returned invalid json: %w", attempt, err) + } + if err := writeFileAtomic(req.OutputRawTranscriptPath, body, 0o644); err != nil { + result.Duration = time.Since(start) + return result, fmt.Errorf("whisperx write transcript output %q: %w", req.OutputRawTranscriptPath, err) + } + result.Duration = time.Since(start) + result.Metadata = map[string]any{ + "adapter": "whisperx_http", + } + return result, nil + } + + lastErr = err + if !c.shouldRetry(ctx, err, status) || attempt == attempts { + break + } + + if c.retryDelay > 0 { + timer := time.NewTimer(c.retryDelay) + select { + case <-ctx.Done(): + timer.Stop() + result.Duration = time.Since(start) + return result, fmt.Errorf("whisperx attempt %d aborted during retry delay: %w", attempt, ctx.Err()) + case <-timer.C: + } + } + } + + result.Duration = time.Since(start) + if lastErr == nil { + lastErr = fmt.Errorf("whisperx transcription failed without explicit error") + } + return result, fmt.Errorf("whisperx transcription failed after %d attempt(s): %w", result.Attempts, lastErr) +} + +func (c *HTTPClient) doTranscribeAttempt(ctx context.Context, audioPath string) (int, []byte, error) { + bodyBuf := &bytes.Buffer{} + writer := multipart.NewWriter(bodyBuf) + + fileWriter, err := writer.CreateFormFile("file", filepath.Base(audioPath)) + if err != nil { + return 0, nil, fmt.Errorf("create multipart file field: %w", err) + } + + audioFile, err := os.Open(audioPath) + if err != nil { + return 0, nil, fmt.Errorf("open audio file %q: %w", audioPath, err) + } + if _, err := io.Copy(fileWriter, audioFile); err != nil { + _ = audioFile.Close() + return 0, nil, fmt.Errorf("copy audio file %q: %w", audioPath, err) + } + if err := audioFile.Close(); err != nil { + return 0, nil, fmt.Errorf("close audio file %q: %w", audioPath, err) + } + + if err := writer.WriteField("language", c.language); err != nil { + return 0, nil, fmt.Errorf("write language form field: %w", err) + } + if err := writer.Close(); err != nil { + return 0, nil, fmt.Errorf("close multipart writer: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), bodyBuf) + if err != nil { + return 0, nil, fmt.Errorf("build whisperx request: %w", err) + } + req.Header.Set("Content-Type", writer.FormDataContentType()) + + resp, err := c.httpClient.Do(req) + if err != nil { + return 0, nil, fmt.Errorf("perform whisperx request: %w", err) + } + defer resp.Body.Close() + + data, err := readBounded(resp.Body, c.maxResponseBytes) + if err != nil { + return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err) + } + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return resp.StatusCode, nil, fmt.Errorf("whisperx returned status %d", resp.StatusCode) + } + return resp.StatusCode, data, nil +} + +func (c *HTTPClient) validateJSON(data []byte) error { + var v any + if err := json.Unmarshal(data, &v); err != nil { + return err + } + return nil +} + +func (c *HTTPClient) shouldRetry(parent context.Context, err error, status int) bool { + if parent.Err() != nil { + return false + } + if status == http.StatusTooManyRequests || status >= 500 { + return true + } + if status >= 400 && status < 500 { + return false + } + if errors.Is(err, context.Canceled) { + return false + } + if errors.Is(err, context.DeadlineExceeded) { + return true + } + var nerr net.Error + if errors.As(err, &nerr) { + return true + } + return false +} + +func readBounded(r io.Reader, maxBytes int64) ([]byte, error) { + limited := io.LimitReader(r, maxBytes+1) + data, err := io.ReadAll(limited) + if err != nil { + return nil, err + } + if int64(len(data)) > maxBytes { + return nil, fmt.Errorf("response exceeds max size %d bytes", maxBytes) + } + return data, nil +} + +func writeFileAtomic(path string, data []byte, perm os.FileMode) error { + if strings.TrimSpace(path) == "" { + return fmt.Errorf("path is required") + } + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("create parent dir %q: %w", dir, err) + } + + base := filepath.Base(path) + tmp, err := os.CreateTemp(dir, "."+base+".tmp-*") + if err != nil { + return fmt.Errorf("create temp file: %w", err) + } + tmpPath := tmp.Name() + removeTmp := true + defer func() { + if removeTmp { + _ = os.Remove(tmpPath) + } + }() + + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + return fmt.Errorf("write temp file: %w", err) + } + if err := tmp.Sync(); err != nil { + _ = tmp.Close() + return fmt.Errorf("sync temp file: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("close temp file: %w", err) + } + if err := os.Chmod(tmpPath, perm); err != nil { + return fmt.Errorf("chmod temp file: %w", err) + } + if err := os.Rename(tmpPath, path); err != nil { + return fmt.Errorf("rename temp file: %w", err) + } + removeTmp = false + return nil +} diff --git a/internal/adapters/whisperx/http_test.go b/internal/adapters/whisperx/http_test.go new file mode 100644 index 0000000..a7a7e95 --- /dev/null +++ b/internal/adapters/whisperx/http_test.go @@ -0,0 +1,271 @@ +package whisperx + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestHTTPClientTranscribeSuccess(t *testing.T) { + var gotLanguage string + var gotFileField string + var gotFileSize int + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Fatalf("method = %s, want POST", r.Method) + } + if err := r.ParseMultipartForm(4 << 20); err != nil { + t.Fatalf("ParseMultipartForm() error = %v", err) + } + + gotLanguage = r.FormValue("language") + file, fh, err := r.FormFile("file") + if err != nil { + t.Fatalf("FormFile(file) error = %v", err) + } + defer file.Close() + gotFileField = fh.Filename + data, err := io.ReadAll(file) + if err != nil { + t.Fatalf("ReadAll(file) error = %v", err) + } + gotFileSize = len(data) + + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"schema":"speaker_transcript.v1","segments":[]}`)) + })) + defer srv.Close() + + client, err := NewHTTPClientFromConfigValues(srv.URL, "en", "2s", "10ms", 0) + if err != nil { + t.Fatalf("NewHTTPClientFromConfigValues() error = %v", err) + } + + audioPath := writeWhisperXTestFile(t, "audio.flac", "audio-data") + outPath := filepath.Join(t.TempDir(), "raw.json") + + res, err := client.Transcribe(context.Background(), TranscribeRequest{ + SpeakerID: "alice", + AudioPath: audioPath, + OutputRawTranscriptPath: outPath, + }) + if err != nil { + t.Fatalf("Transcribe() error = %v", err) + } + if res.Attempts != 1 { + t.Fatalf("Attempts = %d, want 1", res.Attempts) + } + if res.HTTPStatus != http.StatusOK { + t.Fatalf("HTTPStatus = %d, want %d", res.HTTPStatus, http.StatusOK) + } + if gotLanguage != "en" { + t.Fatalf("language = %q, want en", gotLanguage) + } + if gotFileField != "audio.flac" { + t.Fatalf("file field filename = %q, want audio.flac", gotFileField) + } + if gotFileSize == 0 { + t.Fatal("file size = 0, want >0") + } + verifyJSONFile(t, outPath) +} + +func TestHTTPClientRetriesOnTransientAndSucceeds(t *testing.T) { + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := calls.Add(1) + if n == 1 { + http.Error(w, "temporary", http.StatusInternalServerError) + return + } + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer srv.Close() + + client, err := NewHTTPClientFromConfigValues(srv.URL, "en", "2s", "1ms", 2) + if err != nil { + t.Fatalf("NewHTTPClientFromConfigValues() error = %v", err) + } + + audioPath := writeWhisperXTestFile(t, "audio.flac", "audio-data") + outPath := filepath.Join(t.TempDir(), "raw.json") + res, err := client.Transcribe(context.Background(), TranscribeRequest{ + AudioPath: audioPath, + OutputRawTranscriptPath: outPath, + }) + if err != nil { + t.Fatalf("Transcribe() error = %v", err) + } + if res.Attempts != 2 { + t.Fatalf("Attempts = %d, want 2", res.Attempts) + } + if calls.Load() != 2 { + t.Fatalf("calls = %d, want 2", calls.Load()) + } + verifyJSONFile(t, outPath) +} + +func TestHTTPClientDoesNotRetryOnNonRetryableStatus(t *testing.T) { + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + http.Error(w, "bad request", http.StatusBadRequest) + })) + defer srv.Close() + + client, err := NewHTTPClientFromConfigValues(srv.URL, "en", "2s", "1ms", 3) + if err != nil { + t.Fatalf("NewHTTPClientFromConfigValues() error = %v", err) + } + + audioPath := writeWhisperXTestFile(t, "audio.flac", "audio-data") + outPath := filepath.Join(t.TempDir(), "raw.json") + _, err = client.Transcribe(context.Background(), TranscribeRequest{ + AudioPath: audioPath, + OutputRawTranscriptPath: outPath, + }) + if err == nil { + t.Fatal("Transcribe() error = nil, want non-nil") + } + if calls.Load() != 1 { + t.Fatalf("calls = %d, want 1", calls.Load()) + } + if _, statErr := os.Stat(outPath); !errors.Is(statErr, os.ErrNotExist) { + t.Fatalf("output should not exist, stat err = %v", statErr) + } +} + +func TestHTTPClientInvalidJSONFailsAndDoesNotPromote(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`not-json`)) + })) + defer srv.Close() + + client, err := NewHTTPClientFromConfigValues(srv.URL, "en", "2s", "1ms", 0) + if err != nil { + t.Fatalf("NewHTTPClientFromConfigValues() error = %v", err) + } + + audioPath := writeWhisperXTestFile(t, "audio.flac", "audio-data") + outPath := filepath.Join(t.TempDir(), "raw.json") + _, err = client.Transcribe(context.Background(), TranscribeRequest{ + AudioPath: audioPath, + OutputRawTranscriptPath: outPath, + }) + if err == nil { + t.Fatal("Transcribe() error = nil, want non-nil") + } + if !strings.Contains(strings.ToLower(err.Error()), "json") { + t.Fatalf("error = %q, want json context", err.Error()) + } + if _, statErr := os.Stat(outPath); !errors.Is(statErr, os.ErrNotExist) { + t.Fatalf("output should not exist, stat err = %v", statErr) + } +} + +func TestHTTPClientTimeoutRetryDeterministic(t *testing.T) { + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + time.Sleep(150 * time.Millisecond) + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer srv.Close() + + client, err := NewHTTPClientFromConfigValues(srv.URL, "en", "50ms", "1ms", 1) + if err != nil { + t.Fatalf("NewHTTPClientFromConfigValues() error = %v", err) + } + + audioPath := writeWhisperXTestFile(t, "audio.flac", "audio-data") + outPath := filepath.Join(t.TempDir(), "raw.json") + res, err := client.Transcribe(context.Background(), TranscribeRequest{ + AudioPath: audioPath, + OutputRawTranscriptPath: outPath, + }) + if err == nil { + t.Fatal("Transcribe() error = nil, want timeout error") + } + if calls.Load() != 2 { + t.Fatalf("calls = %d, want 2 attempts", calls.Load()) + } + if res.Attempts != 2 { + t.Fatalf("Attempts = %d, want 2", res.Attempts) + } +} + +func TestHTTPClientCancelDoesNotRetry(t *testing.T) { + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + time.Sleep(100 * time.Millisecond) + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer srv.Close() + + client, err := NewHTTPClientFromConfigValues(srv.URL, "en", "2s", "1ms", 3) + if err != nil { + t.Fatalf("NewHTTPClientFromConfigValues() error = %v", err) + } + + audioPath := writeWhisperXTestFile(t, "audio.flac", "audio-data") + outPath := filepath.Join(t.TempDir(), "raw.json") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err = client.Transcribe(ctx, TranscribeRequest{ + AudioPath: audioPath, + OutputRawTranscriptPath: outPath, + }) + if err == nil { + t.Fatal("Transcribe() error = nil, want cancellation error") + } + if calls.Load() > 1 { + t.Fatalf("calls = %d, want <= 1 when already canceled", calls.Load()) + } +} + +func TestHTTPClientConstructorValidation(t *testing.T) { + _, err := NewHTTPClientFromConfigValues("", "en", "30m", "2s", 1) + if err == nil { + t.Fatal("expected missing URL error") + } + _, err = NewHTTPClientFromConfigValues("https://example.com", "en", "bad", "2s", 1) + if err == nil { + t.Fatal("expected bad timeout error") + } + _, err = NewHTTPClientFromConfigValues("https://example.com", "en", "30m", "bad", 1) + if err == nil { + t.Fatal("expected bad retry_delay error") + } +} + +func writeWhisperXTestFile(t *testing.T, name, contents string) string { + t.Helper() + path := filepath.Join(t.TempDir(), name) + if err := os.WriteFile(path, []byte(contents), 0o644); err != nil { + t.Fatalf("WriteFile(%q) error = %v", path, err) + } + return path +} + +func verifyJSONFile(t *testing.T, path string) { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("ReadFile(%q) error = %v", path, err) + } + var v any + if err := json.Unmarshal(data, &v); err != nil { + t.Fatalf("json unmarshal output %q: %v", path, err) + } +}