Support multipart uploads > 5gb

Author: itspriddle

.surface

@@ -238,6 +238,7 @@ FLAG vector db import-session create --drop-tables type=bool
 FLAG vector db import-session create --filename type=string
 FLAG vector db import-session create --search-replace-from type=string
 FLAG vector db import-session create --search-replace-to type=string
+FLAG vector db import-session run --parts type=string
 FLAG vector deploy list --page type=int
 FLAG vector deploy list --per-page type=int
 FLAG vector deploy rollback --poll-interval type=duration

internal/api/client.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"os"
@@ -87,6 +88,34 @@ func (c *Client) PutFile(ctx context.Context, url string, file *os.File) (*http.
 	return resp, nil
 }
 
+// PutFilePart uploads a file part via PUT to the given URL (typically a presigned
+// S3 URL for multipart uploads). It returns the ETag header from the response.
+// Unlike other methods, this does not add Authorization or Accept headers.
+func (c *Client) PutFilePart(ctx context.Context, url string, body io.Reader, contentLength int64) (string, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, body)
+	if err != nil {
+		return "", fmt.Errorf("creating file part upload request: %w", err)
+	}
+	req.ContentLength = contentLength
+	req.Header.Set("User-Agent", c.UserAgent)
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("executing file part upload: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return "", ParseErrorResponse(resp)
+	}
+
+	etag := resp.Header.Get("Etag")
+	if etag == "" {
+		return "", fmt.Errorf("S3 response missing ETag header")
+	}
+	return etag, nil
+}
+
 // jsonRequest is a helper that JSON-encodes a body and sends a request.
 // When body is nil, the request is sent with no body and no Content-Type header.
 func (c *Client) jsonRequest(ctx context.Context, method, path string, body any) (*http.Response, error) {

internal/api/client_test.go

@@ -1,6 +1,7 @@
 package api
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"io"
@@ -199,6 +200,68 @@ func TestClient_PutFile(t *testing.T) {
 	assert.Equal(t, "file-content", string(gotBody))
 }
 
+func TestClient_PutFilePart(t *testing.T) {
+	var gotMethod, gotUserAgent string
+	var gotBody []byte
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotMethod = r.Method
+		gotUserAgent = r.Header.Get("User-Agent")
+		gotBody, _ = io.ReadAll(r.Body)
+		// PutFilePart should not add Authorization or Accept headers (presigned S3 URL).
+		assert.Empty(t, r.Header.Get("Authorization"))
+		assert.Empty(t, r.Header.Get("Accept"))
+		w.Header().Set("Etag", `"abc123"`)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+
+	content := []byte("chunk-data-here")
+	reader := bytes.NewReader(content)
+
+	c := NewClient("https://api.example.com", "tok", "vector-cli/test")
+	etag, err := c.PutFilePart(context.Background(), srv.URL+"/upload/part1", reader, int64(len(content)))
+	require.NoError(t, err)
+
+	assert.Equal(t, http.MethodPut, gotMethod)
+	assert.Equal(t, "vector-cli/test", gotUserAgent)
+	assert.Equal(t, "chunk-data-here", string(gotBody))
+	assert.Equal(t, `"abc123"`, etag)
+}
+
+func TestClient_PutFilePartErrorResponse(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+	}))
+	defer srv.Close()
+
+	content := []byte("data")
+	reader := bytes.NewReader(content)
+
+	c := NewClient("https://api.example.com", "tok", "")
+	_, err := c.PutFilePart(context.Background(), srv.URL+"/upload", reader, int64(len(content)))
+	require.Error(t, err)
+
+	apiErr, ok := err.(*APIError)
+	require.True(t, ok, "error should be *APIError")
+	assert.Equal(t, 403, apiErr.HTTPStatus)
+}
+
+func TestClient_PutFilePartMissingEtag(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Return 200 but no ETag header
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+
+	content := []byte("data")
+	reader := bytes.NewReader(content)
+
+	c := NewClient("https://api.example.com", "tok", "")
+	_, err := c.PutFilePart(context.Background(), srv.URL+"/upload", reader, int64(len(content)))
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "missing ETag")
+}
+
 func TestClient_ErrorResponse(t *testing.T) {
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "application/json")

internal/commands/archive.go

@@ -9,9 +9,57 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"github.com/built-fast/vector-cli/internal/appctx"
 	"github.com/built-fast/vector-cli/internal/output"
 )
 
+func uploadMultipart(cmd *cobra.Command, app *appctx.App, file *os.File, fileSize int64, filename string, importID string, uploadParts []any) ([]map[string]any, error) {
+	w := cmd.ErrOrStderr()
+	partCount := int64(len(uploadParts))
+	baseSize := fileSize / partCount
+	lastSize := fileSize - baseSize*(partCount-1)
+
+	sizeMB := float64(fileSize) / (1024 * 1024)
+	_, _ = fmt.Fprintf(w, "Uploading %s (%.1f MB) in %d parts...\n", filename, sizeMB, partCount)
+
+	completedParts := make([]map[string]any, 0, partCount)
+
+	for i, part := range uploadParts {
+		partMap, ok := part.(map[string]any)
+		if !ok {
+			return nil, fmt.Errorf("invalid upload part at index %d", i)
+		}
+
+		partNumber := int(getFloat(partMap, "part_number"))
+		partURL := getString(partMap, "url")
+		if partURL == "" {
+			return nil, fmt.Errorf("upload part %d missing URL", partNumber)
+		}
+
+		chunkSize := baseSize
+		if i == len(uploadParts)-1 {
+			chunkSize = lastSize
+		}
+		offset := baseSize * int64(i)
+
+		_, _ = fmt.Fprintf(w, "  Uploading part %d/%d...\n", partNumber, partCount)
+
+		section := io.NewSectionReader(file, offset, chunkSize)
+		etag, err := app.Client.PutFilePart(cmd.Context(), partURL, section, chunkSize)
+		if err != nil {
+			return nil, fmt.Errorf("failed to upload part %d: %w", partNumber, err)
+		}
+
+		completedParts = append(completedParts, map[string]any{
+			"part_number": partNumber,
+			"etag":        etag,
+		})
+	}
+
+	_, _ = fmt.Fprintln(w, "Upload complete.")
+	return completedParts, nil
+}
+
 // NewArchiveCmd creates the archive command group.
 func NewArchiveCmd() *cobra.Command {
 	cmd := &cobra.Command{
@@ -116,40 +164,59 @@ func newArchiveImportCmd() *cobra.Command {
 			}
 
 			importID := getString(item, "id")
-			uploadURL := getString(item, "upload_url")
-
-			if importID == "" || uploadURL == "" {
-				return fmt.Errorf("import session response missing upload URL or import ID")
+			if importID == "" {
+				return fmt.Errorf("import session response missing import ID")
 			}
 
-			// Step 2: Upload file to presigned URL
-			sizeMB := float64(fileSize) / (1024 * 1024)
-			_, _ = fmt.Fprintf(w, "Uploading %s (%.1f MB)...\n", filename, sizeMB)
+			// Step 2: Upload file
+			var runBody any
 
-			uploadResp, err := app.Client.PutFile(cmd.Context(), uploadURL, file)
-			if err != nil {
-				return fmt.Errorf("failed to upload file: %w", err)
-			}
-			defer func() { _ = uploadResp.Body.Close() }()
+			if getBool(item, "is_multipart") {
+				uploadParts := getSlice(item, "upload_parts")
+				if len(uploadParts) == 0 {
+					return fmt.Errorf("multipart import session response missing upload parts")
+				}
 
-			_, _ = fmt.Fprintln(w, "Upload complete.")
+				completedParts, uploadErr := uploadMultipart(cmd, app, file, fileSize, filename, importID, uploadParts)
+				if uploadErr != nil {
+					return uploadErr
+				}
+
+				runBody = map[string]any{"parts": completedParts}
+			} else {
+				uploadURL := getString(item, "upload_url")
+				if uploadURL == "" {
+					return fmt.Errorf("import session response missing upload URL")
+				}
+
+				sizeMB := float64(fileSize) / (1024 * 1024)
+				_, _ = fmt.Fprintf(w, "Uploading %s (%.1f MB)...\n", filename, sizeMB)
+
+				uploadResp, uploadErr := app.Client.PutFile(cmd.Context(), uploadURL, file)
+				if uploadErr != nil {
+					return fmt.Errorf("failed to upload file: %w", uploadErr)
+				}
+				defer func() { _ = uploadResp.Body.Close() }()
+
+				_, _ = fmt.Fprintln(w, "Upload complete.")
+			}
 
 			// Step 3: Trigger import
 			_, _ = fmt.Fprintln(w, "Starting import...")
 
 			runEndpoint := fmt.Sprintf("%s/%s/run", importsPath(siteID), importID)
-			runResp, err := app.Client.Post(cmd.Context(), runEndpoint, nil)
+			runResp, err := app.Client.Post(cmd.Context(), runEndpoint, runBody)
 			if err != nil {
 				return fmt.Errorf("failed to start import: %w", err)
 			}
 			defer func() { _ = runResp.Body.Close() }()
 
-			runBody, err := io.ReadAll(runResp.Body)
+			runRespBody, err := io.ReadAll(runResp.Body)
 			if err != nil {
 				return fmt.Errorf("failed to start import: %w", err)
 			}
 
-			runData, err := parseResponseData(runBody)
+			runData, err := parseResponseData(runRespBody)
 			if err != nil {
 				return fmt.Errorf("failed to start import: %w", err)
 			}