narratio/internal/adapters/whisperx/http.go

package whisperx

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"mime/multipart"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"gitea.maximumdirect.net/eric/narratio/internal/fileops"
)

const (
	defaultMaxWhisperXResponseBytes int64 = 10 * 1024 * 1024
	whisperXUploadBufferSize              = 32 * 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
	openAudio        func(string) (io.ReadCloser, error)
}

// 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.IsAbs() || u.Host == "" || !isHTTPURLScheme(u.Scheme) {
		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: must be an absolute http or https URL", 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 {
		transport := http.DefaultTransport.(*http.Transport).Clone()
		transport.ExpectContinueTimeout = 100 * time.Millisecond
		client = &http.Client{Transport: transport}
	}

	maxBytes := cfg.MaxResponseBytes
	if maxBytes <= 0 {
		maxBytes = defaultMaxWhisperXResponseBytes
	}

	return &HTTPClient{
		url:              u,
		language:         cfg.Language,
		timeout:          cfg.Timeout,
		retries:          cfg.Retries,
		retryDelay:       cfg.RetryDelay,
		httpClient:       client,
		maxResponseBytes: maxBytes,
		openAudio:        func(path string) (io.ReadCloser, error) { return os.Open(path) },
	}, 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, fileops.WorkspaceFileMode); 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) {
	upload := newMultipartUpload(ctx, audioPath, c.language, c.openAudio)
	defer upload.Close()

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), upload)
	if err != nil {
		return 0, nil, fmt.Errorf("build whisperx request: %w", err)
	}
	req.Header.Set("Content-Type", upload.contentType)
	req.Header.Set("Expect", "100-continue")

	resp, err := c.httpClient.Do(req)
	if err != nil {
		_ = upload.Close()
		if producerErr := upload.Wait(); producerErr != nil {
			return 0, nil, fmt.Errorf("stream whisperx request body: %w", producerErr)
		}
		return 0, nil, fmt.Errorf("perform whisperx request: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		_ = upload.Close()
		if producerErr := upload.Wait(); producerErr != nil {
			return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", producerErr)
		}
		if _, err := readWhisperXResponse(resp.Body, c.maxResponseBytes); err != nil {
			return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err)
		}
		return resp.StatusCode, nil, fmt.Errorf("whisperx returned status %d", resp.StatusCode)
	}

	if err := upload.Wait(); err != nil {
		return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", err)
	}

	data, err := readWhisperXResponse(resp.Body, c.maxResponseBytes)
	if err != nil {
		return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err)
	}
	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 readWhisperXResponse(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("whisperx response exceeds configured limit of %d bytes", maxBytes)
	}
	return data, nil
}

func isHTTPURLScheme(scheme string) bool {
	switch strings.ToLower(scheme) {
	case "http", "https":
		return true
	default:
		return false
	}
}

type multipartUpload struct {
	reader      *io.PipeReader
	writer      *io.PipeWriter
	contentType string
	done        chan struct{}

	mu      sync.Mutex
	audio   io.Closer
	err     error
	aborted bool
}

func newMultipartUpload(ctx context.Context, audioPath, language string, openAudio func(string) (io.ReadCloser, error)) *multipartUpload {
	reader, writer := io.Pipe()
	multipartWriter := multipart.NewWriter(writer)
	upload := &multipartUpload{
		reader:      reader,
		writer:      writer,
		contentType: multipartWriter.FormDataContentType(),
		done:        make(chan struct{}),
	}

	go func() {
		err := upload.write(ctx, multipartWriter, audioPath, language, openAudio)
		if err != nil {
			_ = writer.CloseWithError(err)
		} else {
			_ = writer.Close()
		}
		upload.mu.Lock()
		upload.err = err
		upload.audio = nil
		upload.mu.Unlock()
		close(upload.done)
	}()

	go func() {
		select {
		case <-ctx.Done():
			upload.abort()
		case <-upload.done:
		}
	}()

	return upload
}

func (u *multipartUpload) Read(p []byte) (int, error) {
	return u.reader.Read(p)
}

func (u *multipartUpload) Close() error {
	u.abort()
	return nil
}

func (u *multipartUpload) Wait() error {
	<-u.done
	u.mu.Lock()
	defer u.mu.Unlock()
	return u.err
}

func (u *multipartUpload) write(ctx context.Context, writer *multipart.Writer, audioPath, language string, openAudio func(string) (io.ReadCloser, error)) error {
	fileWriter, err := writer.CreateFormFile("file", filepath.Base(audioPath))
	if err != nil {
		return u.producerError(ctx, fmt.Errorf("create multipart file field: %w", err))
	}

	audioFile, err := openAudio(audioPath)
	if err != nil {
		return u.producerError(ctx, fmt.Errorf("open audio file %q: %w", audioPath, err))
	}
	u.setAudio(audioFile)

	_, copyErr := io.CopyBuffer(fileWriter, &contextReader{ctx: ctx, reader: audioFile}, make([]byte, whisperXUploadBufferSize))
	closeErr := audioFile.Close()
	u.clearAudio(audioFile)
	if copyErr != nil {
		return u.producerError(ctx, fmt.Errorf("copy audio file %q: %w", audioPath, copyErr))
	}
	if closeErr != nil {
		return u.producerError(ctx, fmt.Errorf("close audio file %q: %w", audioPath, closeErr))
	}

	if err := writer.WriteField("language", language); err != nil {
		return u.producerError(ctx, fmt.Errorf("write language form field: %w", err))
	}
	if err := writer.Close(); err != nil {
		return u.producerError(ctx, fmt.Errorf("close multipart writer: %w", err))
	}
	return nil
}

func (u *multipartUpload) producerError(ctx context.Context, err error) error {
	if ctx.Err() != nil {
		return ctx.Err()
	}
	u.mu.Lock()
	aborted := u.aborted
	u.mu.Unlock()
	if aborted {
		return nil
	}
	return err
}

func (u *multipartUpload) setAudio(audio io.Closer) {
	u.mu.Lock()
	u.audio = audio
	aborted := u.aborted
	u.mu.Unlock()
	if aborted {
		_ = audio.Close()
	}
}

func (u *multipartUpload) clearAudio(audio io.Closer) {
	u.mu.Lock()
	if u.audio == audio {
		u.audio = nil
	}
	u.mu.Unlock()
}

func (u *multipartUpload) abort() {
	u.mu.Lock()
	if u.aborted {
		u.mu.Unlock()
		return
	}
	u.aborted = true
	audio := u.audio
	u.mu.Unlock()

	_ = u.reader.Close()
	if audio != nil {
		_ = audio.Close()
	}
}

type contextReader struct {
	ctx    context.Context
	reader io.Reader
}

func (r *contextReader) Read(p []byte) (int, error) {
	select {
	case <-r.ctx.Done():
		return 0, r.ctx.Err()
	default:
		return r.reader.Read(p)
	}
}

func writeFileAtomic(path string, data []byte, perm os.FileMode) error {
	if strings.TrimSpace(path) == "" {
		return fmt.Errorf("path is required")
	}
	if err := fileops.WriteFileAtomic(path, data, perm); err != nil {
		return fmt.Errorf("write file %q: %w", path, err)
	}
	return nil
}