Implement WhisperX HTTP adapter

2026-05-03 15:29:34 -05:00
parent c2a715d3f6
commit 24833194db
4 changed files with 603 additions and 1 deletions
--- 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
 }
--- 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}
 	}
--- a/internal/adapters/whisperx/http.go
+++ 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
+}
--- a/internal/adapters/whisperx/http_test.go
+++ 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)
+	}
+}