Make WhisperX uploads replayable

internal/adapters/whisperx/http.go

@@ -6,14 +6,11 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"mime/multipart"
 	"net"
 	"net/http"
 	"net/url"
 	"os"
-	"path/filepath"
 	"strings"
-	"sync"
 	"time"
 
 	"gitea.maximumdirect.net/eric/narratio/internal/fileops"
@@ -21,7 +18,6 @@ import (
 
 const (
 	defaultMaxWhisperXResponseBytes int64 = 10 * 1024 * 1024
-	whisperXUploadBufferSize              = 32 * 1024
 )
 
 // HTTPClientConfig contains parsed, deterministic WhisperX HTTP client settings.
@@ -192,21 +188,32 @@ func (c *HTTPClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra
 }
 
 func (c *HTTPClient) doTranscribeAttempt(ctx context.Context, audioPath string) (int, []byte, error) {
-	upload := newMultipartUpload(ctx, audioPath, c.language, c.openAudio)
-	defer upload.Close()
+	upload, err := newMultipartBodyFactory(ctx, audioPath, c.language, c.openAudio)
+	if err != nil {
+		return 0, nil, fmt.Errorf("prepare whisperx request body: %w", err)
+	}
+	defer func() { _ = upload.Close() }()
+	body, err := upload.NewBody()
+	if err != nil {
+		return 0, nil, fmt.Errorf("open whisperx request body: %w", err)
+	}
 
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), upload)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), body)
 	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")
+	req.GetBody = upload.NewBody
 
 	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)
+		if sourceErr := upload.SourceError(true); sourceErr != nil {
+			return 0, nil, fmt.Errorf("stream whisperx request body: %w", sourceErr)
+		}
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			return 0, nil, fmt.Errorf("perform whisperx request: %w", ctxErr)
 		}
 		return 0, nil, fmt.Errorf("perform whisperx request: %w", err)
 	}
@@ -214,8 +221,8 @@ func (c *HTTPClient) doTranscribeAttempt(ctx context.Context, audioPath string)
 
 	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 sourceErr := upload.SourceError(false); sourceErr != nil {
+			return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", sourceErr)
 		}
 		if _, err := readWhisperXResponse(resp.Body, c.maxResponseBytes); err != nil {
 			return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err)
@@ -223,13 +230,19 @@ func (c *HTTPClient) doTranscribeAttempt(ctx context.Context, audioPath string)
 		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, responseErr := readWhisperXResponse(resp.Body, c.maxResponseBytes)
+	_ = upload.Close()
+	if sourceErr := upload.SourceError(false); sourceErr != nil {
+		if responseErr != nil {
+			return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", errors.Join(sourceErr, responseErr))
+		}
+		return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", sourceErr)
 	}
-
-	data, err := readWhisperXResponse(resp.Body, c.maxResponseBytes)
-	if err != nil {
-		return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err)
+	if err := ctx.Err(); err != nil {
+		return resp.StatusCode, nil, fmt.Errorf("complete whisperx request: %w", err)
+	}
+	if responseErr != nil {
+		return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", responseErr)
 	}
 	return resp.StatusCode, data, nil
 }
@@ -286,161 +299,6 @@ func isHTTPURLScheme(scheme string) bool {
 	}
 }
 
-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")