package server

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"hash/crc32"
	"io"
	"math"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"
	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/semantic"
	"github.com/mem9-ai/drive9/pkg/testtidb"
)

func newTestServerWithS3(t *testing.T) (*Server, *s3client.LocalS3Client) {
	t.Helper()
	return newTestServerWithS3Config(t, backend.Options{}, TenantWorkerOptions{})
}

func newTestServerWithS3Config(t *testing.T, backendOpts backend.Options, workerOpts TenantWorkerOptions) (*Server, *s3client.LocalS3Client) {
	t.Helper()
	blobDir, err := os.MkdirTemp("", "dat9-srv-blobs-*")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = os.RemoveAll(blobDir) })

	s3Dir, err := os.MkdirTemp("", "dat9-srv-s3-*")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = os.RemoveAll(s3Dir) })

	initServerTenantSchema(t, testDSN)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })

	s3c, err := s3client.NewLocal(s3Dir, "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}

	b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, backendOpts)
	if err != nil {
		t.Fatal(err)
	}
	return NewWithConfig(Config{Backend: b, TenantWorkers: workerOpts}), s3c
}

type failingCreateMultipartS3 struct {
	*s3client.LocalS3Client
	err error
}

func (s *failingCreateMultipartS3) CreateMultipartUpload(ctx context.Context, key string, algo s3client.ChecksumAlgo, encOpts s3client.EncryptionOpts) (*s3client.MultipartUpload, error) {
	return nil, s.err
}

func newTestServerWithCreateMultipartError(t *testing.T, err error) *Server {
	t.Helper()

	blobDir, err2 := os.MkdirTemp("", "dat9-srv-blobs-*")
	if err2 != nil {
		t.Fatal(err2)
	}
	t.Cleanup(func() { _ = os.RemoveAll(blobDir) })

	s3Dir, err2 := os.MkdirTemp("", "dat9-srv-s3-*")
	if err2 != nil {
		t.Fatal(err2)
	}
	t.Cleanup(func() { _ = os.RemoveAll(s3Dir) })

	initServerTenantSchema(t, testDSN)
	store, err2 := datastore.Open(testDSN)
	if err2 != nil {
		t.Fatal(err2)
	}
	testtidb.ResetDB(t, store.DB())
	t.Cleanup(func() { _ = store.Close() })

	localS3, err2 := s3client.NewLocal(s3Dir, "http://localhost:9091/s3")
	if err2 != nil {
		t.Fatal(err2)
	}
	failingS3 := &failingCreateMultipartS3{LocalS3Client: localS3, err: err}
	b, err2 := backend.NewWithS3ModeAndOptions(store, failingS3, true, backend.Options{})
	if err2 != nil {
		t.Fatal(err2)
	}
	return NewWithConfig(Config{Backend: b})
}

func partChecksumHeader(data []byte) string {
	table := crc32.MakeTable(crc32.Castagnoli)
	parts := s3client.CalcParts(int64(len(data)), s3client.PartSize)
	out := make([]string, 0, len(parts))
	for _, p := range parts {
		start := int64(p.Number-1) * s3client.PartSize
		end := start + p.Size
		v := crc32.Checksum(data[start:end], table)
		b := make([]byte, 4)
		binary.BigEndian.PutUint32(b, v)
		out = append(out, base64.StdEncoding.EncodeToString(b))
	}
	return strings.Join(out, ",")
}

func mustUploadLargeServerFile(t *testing.T, ts *httptest.Server, s *Server, s3c *s3client.LocalS3Client, path string, body []byte) {
	t.Helper()

	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs"+path, bytes.NewReader(body))
	req.ContentLength = int64(len(body))
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusAccepted {
		respBody, _ := io.ReadAll(resp.Body)
		t.Fatalf("initiate upload: got %d, want 202: %s", resp.StatusCode, respBody)
	}

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatalf("decode upload plan: %v", err)
	}

	upload, err := s.fallback.GetUpload(context.Background(), plan.UploadID)
	if err != nil {
		t.Fatalf("GetUpload(%q): %v", plan.UploadID, err)
	}

	for _, p := range plan.Parts {
		start := int64(p.Number-1) * plan.PartSize
		end := start + p.Size
		if _, err := s3c.UploadPart(context.Background(), upload.S3UploadID, p.Number, bytes.NewReader(body[start:end])); err != nil {
			t.Fatalf("upload part %d: %v", p.Number, err)
		}
	}

	req, _ = http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/complete", nil)
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusOK {
		respBody, _ := io.ReadAll(resp.Body)
		t.Fatalf("complete upload: got %d, want 200: %s", resp.StatusCode, respBody)
	}
}

func TestLargeFilePut202(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	// PUT with Content-Length >= 1MB should return 202
	body := make([]byte, 1<<20) // exactly 1MB
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/big.bin", bytes.NewReader(body))
	req.ContentLength = int64(len(body))
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("expected 202, got %d", resp.StatusCode)
	}

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	if plan.UploadID == "" {
		t.Error("expected upload_id")
	}
	if len(plan.Parts) == 0 {
		t.Error("expected parts")
	}
}

func TestDirectS3UploadAndPresignedReadUseS3TrafficPath(t *testing.T) {
	s, s3c := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	body := make([]byte, 1<<20)
	mustUploadLargeServerFile(t, ts, s, s3c, "/billing-path.bin", body)

	client := &http.Client{CheckRedirect: func(_ *http.Request, _ []*http.Request) error {
		return http.ErrUseLastResponse
	}}
	resp, err := client.Get(ts.URL + "/v1/fs/billing-path.bin")
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusFound {
		t.Fatalf("read status = %d, want %d", resp.StatusCode, http.StatusFound)
	}

	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	for _, direction := range []string{"read", "write"} {
		want := `drive9_tenant_file_bytes_total{direction="` + direction + `",tenant_id="local",tidbcloud_org_id="guest",traffic_path="s3"}`
		if !strings.Contains(text, want) {
			t.Fatalf("missing %s S3 traffic metric %q:\n%s", direction, want, text)
		}
		wantOps := `drive9_tenant_file_ops_total{direction="` + direction + `",tenant_id="local",tidbcloud_org_id="guest",traffic_path="s3"}`
		if !strings.Contains(text, wantOps) {
			t.Fatalf("missing %s S3 ops metric %q:\n%s", direction, wantOps, text)
		}
	}
}

func TestUploadCompleteRejectsInvalidJSONTags(t *testing.T) {
	s, s3c := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	body := make([]byte, 1<<20)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/bad-tags.bin", bytes.NewReader(body))
	req.ContentLength = int64(len(body))
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusAccepted {
		respBody, _ := io.ReadAll(resp.Body)
		t.Fatalf("initiate upload: got %d, want 202: %s", resp.StatusCode, respBody)
	}

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatalf("decode upload plan: %v", err)
	}

	upload, err := s.fallback.GetUpload(context.Background(), plan.UploadID)
	if err != nil {
		t.Fatalf("GetUpload(%q): %v", plan.UploadID, err)
	}
	for _, p := range plan.Parts {
		start := int64(p.Number-1) * plan.PartSize
		end := start + p.Size
		if _, err := s3c.UploadPart(context.Background(), upload.S3UploadID, p.Number, bytes.NewReader(body[start:end])); err != nil {
			t.Fatalf("upload part %d: %v", p.Number, err)
		}
	}

	completeBody, err := json.Marshal(map[string]any{
		"tags": map[string]string{"owner=id": "alice"},
	})
	if err != nil {
		t.Fatalf("marshal complete body: %v", err)
	}

	req, _ = http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/complete", bytes.NewReader(completeBody))
	req.Header.Set("Content-Type", "application/json")
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusBadRequest {
		respBody, _ := io.ReadAll(resp.Body)
		t.Fatalf("complete upload: got %d, want 400: %s", resp.StatusCode, respBody)
	}
	respBody, _ := io.ReadAll(resp.Body)
	if !strings.Contains(string(respBody), "contains '='") {
		t.Fatalf("complete upload response = %q, want invalid tag key error", respBody)
	}
}

func TestUploadInitiateByBody202(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	body := make([]byte, 1<<20) // exactly 1MB
	reqBody := map[string]any{
		"path":           "/big-body.bin",
		"total_size":     len(body),
		"part_checksums": strings.Split(partChecksumHeader(body), ","),
	}
	p, _ := json.Marshal(reqBody)
	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/initiate", bytes.NewReader(p))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusAccepted {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 202, got %d: %s", resp.StatusCode, b)
	}

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	if plan.UploadID == "" {
		t.Error("expected upload_id")
	}
	if len(plan.Parts) == 0 {
		t.Error("expected parts")
	}
}

func TestUploadInitiatePreservesSpaceOnlyPath(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	body := make([]byte, 1<<20)
	tests := []struct {
		name     string
		endpoint string
		payload  map[string]any
	}{
		{
			name:     "v1",
			endpoint: "/v1/uploads/initiate",
			payload: map[string]any{
				"path":           "  ",
				"total_size":     len(body),
				"part_checksums": strings.Split(partChecksumHeader(body), ","),
			},
		},
		{
			name:     "v2",
			endpoint: "/v2/uploads/initiate",
			payload: map[string]any{
				"path":       "  ",
				"total_size": len(body),
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			encoded, err := json.Marshal(tt.payload)
			if err != nil {
				t.Fatal(err)
			}
			req := mustNewRequest(t, http.MethodPost, ts.URL+tt.endpoint, bytes.NewReader(encoded))
			req.Header.Set("Content-Type", "application/json")
			resp, err := http.DefaultClient.Do(req)
			if err != nil {
				t.Fatal(err)
			}
			defer func() { _ = resp.Body.Close() }()
			if resp.StatusCode != http.StatusAccepted {
				respBody, _ := io.ReadAll(resp.Body)
				t.Errorf("status = %d, want %d: %s", resp.StatusCode, http.StatusAccepted, respBody)
			}
		})
	}
}

func TestUploadInitiateSanitizesCreateMultipartProviderError(t *testing.T) {
	leakedErr := errors.New("operation error S3: CreateMultipartUpload, https response error StatusCode: 403, RequestID: QCRQPQAG0J880ZFK, HostID: KAjMF1T1LtFzMU/V7Tzp0rt14AqR7irO2HJtlHdXBCQioAv14tJsNkbUWvBZbSGYcFxqroX6fdo=, api error AccessDenied: User: arn:aws:sts::401696231252:assumed-role/prod-dat9-server-irsa/1777968029598718639 is not authorized to perform: kms:GenerateDataKey on resource: arn:aws:kms:ap-southeast-1:401696231252:key/fca526d7-bbff-44d4-9eee-7dd44053e85b")
	s := newTestServerWithCreateMultipartError(t, leakedErr)
	ts := httptest.NewServer(s)
	defer ts.Close()

	body := make([]byte, 1<<20)
	v1Body, _ := json.Marshal(map[string]any{
		"path":           "/leak-v1.bin",
		"total_size":     len(body),
		"part_checksums": strings.Split(partChecksumHeader(body), ","),
	})
	v2Body, _ := json.Marshal(map[string]any{
		"path":       "/leak-v2.bin",
		"total_size": len(body),
	})

	tests := []struct {
		name        string
		req         *http.Request
		contentType string
	}{
		{
			name:        "large put",
			req:         mustNewRequest(t, http.MethodPut, ts.URL+"/v1/fs/leak-put.bin", bytes.NewReader(body)),
			contentType: "",
		},
		{
			name:        "v1 initiate",
			req:         mustNewRequest(t, http.MethodPost, ts.URL+"/v1/uploads/initiate", bytes.NewReader(v1Body)),
			contentType: "application/json",
		},
		{
			name:        "v2 initiate",
			req:         mustNewRequest(t, http.MethodPost, ts.URL+"/v2/uploads/initiate", bytes.NewReader(v2Body)),
			contentType: "application/json",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if tt.contentType != "" {
				tt.req.Header.Set("Content-Type", tt.contentType)
			}
			if tt.req.Method == http.MethodPut {
				tt.req.ContentLength = int64(len(body))
				tt.req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
			}
			resp, err := http.DefaultClient.Do(tt.req)
			if err != nil {
				t.Fatal(err)
			}
			defer func() { _ = resp.Body.Close() }()
			if resp.StatusCode != http.StatusInternalServerError {
				respBody, _ := io.ReadAll(resp.Body)
				t.Fatalf("status = %d, want 500: %s", resp.StatusCode, respBody)
			}
			respBody, _ := io.ReadAll(resp.Body)
			text := string(respBody)
			if !strings.Contains(text, internalStorageErrorMessage) {
				t.Fatalf("response = %q, want sanitized storage error", text)
			}
			for _, leaked := range []string{
				"arn:aws",
				"RequestID",
				"HostID",
				"GenerateDataKey",
				"prod-dat9-server-irsa",
				"operation error S3",
				"fca526d7-bbff-44d4-9eee-7dd44053e85b",
			} {
				if strings.Contains(text, leaked) {
					t.Fatalf("response leaked %q: %s", leaked, text)
				}
			}
		})
	}
}

func TestV2PresignValidationErrorRemainsClientError(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	plan := mustInitiateV2Upload(t, ts, "/v2-presign-invalid.bin", 20<<20)

	body, _ := json.Marshal(map[string]any{"part_number": plan.TotalParts + 1})
	req := mustNewRequest(t, http.MethodPost, ts.URL+"/v2/uploads/"+plan.UploadID+"/presign", bytes.NewReader(body))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusBadRequest {
		respBody, _ := io.ReadAll(resp.Body)
		t.Fatalf("status = %d, want 400: %s", resp.StatusCode, respBody)
	}
	respBody, _ := io.ReadAll(resp.Body)
	text := string(respBody)
	if !strings.Contains(text, "invalid part number") {
		t.Fatalf("response = %q, want invalid part number", text)
	}
	if strings.Contains(text, internalStorageErrorMessage) {
		t.Fatalf("response = %q, want client error not internal storage error", text)
	}
}

func TestV2CompleteValidationErrorRemainsClientError(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	plan := mustInitiateV2Upload(t, ts, "/v2-complete-invalid.bin", 20<<20)
	presignBody, _ := json.Marshal(map[string]any{"part_number": 1})
	req := mustNewRequest(t, http.MethodPost, ts.URL+"/v2/uploads/"+plan.UploadID+"/presign", bytes.NewReader(presignBody))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("presign status = %d, want 200", resp.StatusCode)
	}

	completeBody, _ := json.Marshal(map[string]any{
		"parts": []map[string]any{{"number": 1, "etag": "etag-1"}},
	})
	req = mustNewRequest(t, http.MethodPost, ts.URL+"/v2/uploads/"+plan.UploadID+"/complete", bytes.NewReader(completeBody))
	req.Header.Set("Content-Type", "application/json")
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusBadRequest {
		respBody, _ := io.ReadAll(resp.Body)
		t.Fatalf("status = %d, want 400: %s", resp.StatusCode, respBody)
	}
	respBody, _ := io.ReadAll(resp.Body)
	text := string(respBody)
	if !strings.Contains(text, "part count mismatch") {
		t.Fatalf("response = %q, want part count mismatch", text)
	}
	if strings.Contains(text, internalStorageErrorMessage) {
		t.Fatalf("response = %q, want client error not internal storage error", text)
	}
}

func mustInitiateV2Upload(t *testing.T, ts *httptest.Server, path string, totalSize int64) backend.UploadPlanV2 {
	t.Helper()
	reqBody, _ := json.Marshal(map[string]any{
		"path":       path,
		"total_size": totalSize,
	})
	req := mustNewRequest(t, http.MethodPost, ts.URL+"/v2/uploads/initiate", bytes.NewReader(reqBody))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusAccepted {
		body, _ := io.ReadAll(resp.Body)
		t.Fatalf("initiate status = %d, want 202: %s", resp.StatusCode, body)
	}
	var plan backend.UploadPlanV2
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	return plan
}

func mustNewRequest(t *testing.T, method, url string, body io.Reader) *http.Request {
	t.Helper()
	req, err := http.NewRequest(method, url, body)
	if err != nil {
		t.Fatal(err)
	}
	return req
}

func TestV1UploadInitiateReturnsAllPresignedParts(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	totalSize := 20 << 20
	body := make([]byte, totalSize)
	reqBody := map[string]any{
		"path":           "/all-parts.bin",
		"total_size":     totalSize,
		"part_checksums": strings.Split(partChecksumHeader(body), ","),
	}
	p, _ := json.Marshal(reqBody)
	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/initiate", bytes.NewReader(p))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusAccepted {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 202, got %d: %s", resp.StatusCode, b)
	}

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	if plan.PartSize != s3client.PartSize {
		t.Fatalf("part size = %d, want %d", plan.PartSize, s3client.PartSize)
	}
	if len(plan.Parts) != 3 {
		t.Fatalf("expected 3 presigned parts, got %d", len(plan.Parts))
	}
	wantSizes := []int64{s3client.PartSize, s3client.PartSize, 4 << 20}
	for i, part := range plan.Parts {
		if part.Number != i+1 {
			t.Fatalf("part %d number = %d, want %d", i, part.Number, i+1)
		}
		if part.URL == "" {
			t.Fatalf("part %d missing presigned URL", part.Number)
		}
		if part.Size != wantSizes[i] {
			t.Fatalf("part %d size = %d, want %d", part.Number, part.Size, wantSizes[i])
		}
	}
}

func TestV1LargeFilePutRequiresChecksumHeader(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	body := make([]byte, 1<<20)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/missing-checksums.bin", bytes.NewReader(body))
	req.ContentLength = int64(len(body))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusBadRequest {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 400, got %d: %s", resp.StatusCode, b)
	}
	b, _ := io.ReadAll(resp.Body)
	if !strings.Contains(string(b), "missing X-Dat9-Part-Checksums header") {
		t.Fatalf("expected missing checksum header error, got %s", b)
	}
}

func TestV1UploadInitiateByBodyRequiresPartChecksums(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	reqBody := map[string]any{
		"path":       "/missing-body-checksums.bin",
		"total_size": 1 << 20,
	}
	p, _ := json.Marshal(reqBody)
	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/initiate", bytes.NewReader(p))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusBadRequest {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 400, got %d: %s", resp.StatusCode, b)
	}
	b, _ := io.ReadAll(resp.Body)
	if !strings.Contains(string(b), "missing part_checksums") {
		t.Fatalf("expected missing part_checksums error, got %s", b)
	}
}

func TestSmallFilePut200(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	// PUT with Content-Length < 1MB should return 200 (proxied)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/small.txt", bytes.NewReader([]byte("hello")))
	req.ContentLength = 5
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		t.Fatalf("expected 200, got %d", resp.StatusCode)
	}
}

func TestAutoImagePutWritesContentTextEndToEnd(t *testing.T) {
	s, _ := newTestServerWithS3Config(t, backend.Options{
		DatabaseAutoEmbedding: true,
		AsyncImageExtract: backend.AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: staticServerImageExtractor{text: "caption from http put"},
		},
	}, TenantWorkerOptions{
		Workers:       1,
		PollInterval:  10 * time.Millisecond,
		LeaseDuration: 200 * time.Millisecond,
	})
	ts := httptest.NewServer(s)
	defer ts.Close()

	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/e2e-put.png", bytes.NewReader(fakeImagePNG))
	req.ContentLength = int64(len(fakeImagePNG))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 200, got %d: %s", resp.StatusCode, body)
	}

	file := mustServerFile(t, s.fallback, "/e2e-put.png")
	waitForContentTextOnServer(t, s.fallback, "/e2e-put.png", "caption from http put", 3*time.Second)
	waitForTaskStatus(t, s.fallback, file.FileID, 1, string(semantic.TaskSucceeded), 3*time.Second)

	tasks := loadSemanticTaskRowsForResource(t, s.fallback, file.FileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1", len(tasks))
	}
	if tasks[0].TaskType != string(semantic.TaskTypeImgExtractText) || tasks[0].Status != string(semantic.TaskSucceeded) {
		t.Fatalf("unexpected semantic task rows: %+v", tasks)
	}
}

func TestAutoVideoPutWritesVisualContentTextEndToEnd(t *testing.T) {
	s, _ := newTestServerWithS3Config(t, backend.Options{
		TenantID:              "test-tenant",
		DatabaseAutoEmbedding: true,
		AsyncVideoExtract: backend.AsyncVideoExtractOptions{
			Enabled:         true,
			Extractor:       staticServerVideoExtractor{text: "a golden retriever running through a sunlit park with oak trees"},
			TenantAllowlist: map[string]struct{}{"test-tenant": {}},
		},
	}, TenantWorkerOptions{
		Workers:       1,
		PollInterval:  10 * time.Millisecond,
		LeaseDuration: 200 * time.Millisecond,
	})
	ts := httptest.NewServer(s)
	defer ts.Close()

	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/e2e-video.mp4", bytes.NewReader([]byte("fake-mp4-bytes")))
	req.ContentLength = int64(len("fake-mp4-bytes"))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 200, got %d: %s", resp.StatusCode, body)
	}

	file := mustServerFile(t, s.fallback, "/e2e-video.mp4")

	// Wait for the visual description to be written back as content_text.
	// "golden retriever" is a visual-only term that proves the Vision extractor
	// path was invoked, not the audio transcript path.
	waitForContentTextOnServer(t, s.fallback, "/e2e-video.mp4", "golden retriever", 3*time.Second)
	waitForTaskStatus(t, s.fallback, file.FileID, 1, string(semantic.TaskSucceeded), 3*time.Second)

	// Verify exactly one task was created and it is video_extract_visual (not audio_extract_text).
	tasks := loadSemanticTaskRowsForResource(t, s.fallback, file.FileID)
	if len(tasks) != 1 {
		t.Fatalf("semantic task count=%d, want 1 (video only, no audio)", len(tasks))
	}
	if tasks[0].TaskType != string(semantic.TaskTypeVideoExtractVisual) {
		t.Fatalf("task type=%q, want %q", tasks[0].TaskType, semantic.TaskTypeVideoExtractVisual)
	}
	if tasks[0].Status != string(semantic.TaskSucceeded) {
		t.Fatalf("task status=%q, want %q", tasks[0].Status, semantic.TaskSucceeded)
	}
}

func TestVideoQuotaSkipsEnqueueAtLimit(t *testing.T) {
	s, _ := newTestServerWithS3Config(t, backend.Options{
		TenantID:              "test-tenant",
		DatabaseAutoEmbedding: true,
		AsyncVideoExtract: backend.AsyncVideoExtractOptions{
			Enabled:          true,
			Extractor:        staticServerVideoExtractor{text: "visual content"},
			TenantAllowlist:  map[string]struct{}{"test-tenant": {}},
			MaxVideoLLMFiles: 1, // allow only 1 video
		},
	}, TenantWorkerOptions{
		Workers:       1,
		PollInterval:  10 * time.Millisecond,
		LeaseDuration: 200 * time.Millisecond,
	})
	ts := httptest.NewServer(s)
	defer ts.Close()

	// Upload first video — should enqueue and succeed.
	req1, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/first.mp4", bytes.NewReader([]byte("vid1")))
	req1.ContentLength = 4
	resp1, err := http.DefaultClient.Do(req1)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp1.Body.Close()
	if resp1.StatusCode != http.StatusOK {
		t.Fatalf("first upload: status=%d", resp1.StatusCode)
	}
	file1 := mustServerFile(t, s.fallback, "/first.mp4")
	waitForTaskStatus(t, s.fallback, file1.FileID, 1, string(semantic.TaskSucceeded), 3*time.Second)

	// Upload second video — upload succeeds but no video_extract_visual task.
	req2, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/second.mp4", bytes.NewReader([]byte("vid2")))
	req2.ContentLength = 4
	resp2, err := http.DefaultClient.Do(req2)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp2.Body.Close()
	if resp2.StatusCode != http.StatusOK {
		t.Fatalf("second upload: status=%d, want 200 (upload must succeed)", resp2.StatusCode)
	}
	file2 := mustServerFile(t, s.fallback, "/second.mp4")
	// Give a short window for any mistaken enqueue to appear.
	time.Sleep(200 * time.Millisecond)
	tasks2 := loadSemanticTaskRowsForResource(t, s.fallback, file2.FileID)
	if len(tasks2) != 0 {
		t.Fatalf("second video should have 0 tasks (quota), got %d: %+v", len(tasks2), tasks2)
	}

	// Re-upload first video (overwrite) — quota counts per-extraction-attempt,
	// so the re-upload also consumes quota. With quota=1, already used by the
	// first upload, this re-upload should NOT enqueue a new extraction.
	req3, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/first.mp4", bytes.NewReader([]byte("vid1-v2")))
	req3.ContentLength = 7
	resp3, err := http.DefaultClient.Do(req3)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp3.Body.Close()
	if resp3.StatusCode != http.StatusOK {
		t.Fatalf("re-upload: status=%d", resp3.StatusCode)
	}
	// The first upload consumed the only quota slot. Re-upload must not
	// create a second video_extract_visual task.
	time.Sleep(200 * time.Millisecond)
	tasks3 := loadSemanticTaskRowsForResource(t, s.fallback, file1.FileID)
	if len(tasks3) != 1 {
		t.Fatalf("re-upload of first video should still have 1 task (quota exhausted), got %d: %+v", len(tasks3), tasks3)
	}
}

func TestUploadCompleteEndpoint(t *testing.T) {
	s, s3c := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()
	checksum := strings.Repeat("a", 64)

	// Initiate via PUT 202
	body := make([]byte, 1<<20)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/complete-test.bin", bytes.NewReader(body))
	req.ContentLength = int64(len(body))
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()

	// Get the upload to find s3_upload_id
	upload, _ := s.fallback.GetUpload(context.Background(), plan.UploadID)

	// Upload all parts via S3 client directly
	for _, p := range plan.Parts {
		start := int64(p.Number-1) * plan.PartSize
		end := start + p.Size
		if end > int64(len(body)) {
			end = int64(len(body))
		}
		if _, err := s3c.UploadPart(context.Background(), upload.S3UploadID, p.Number, bytes.NewReader(body[start:end])); err != nil {
			t.Fatalf("upload part %d: %v", p.Number, err)
		}
	}

	// POST /v1/uploads/{id}/complete
	completeBody, _ := json.Marshal(map[string]string{"checksum_sha256": checksum})
	req, _ = http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/complete", bytes.NewReader(completeBody))
	req.Header.Set("Content-Type", "application/json")
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		t.Fatalf("complete: expected 200, got %d", resp.StatusCode)
	}

	req, _ = http.NewRequest(http.MethodHead, ts.URL+"/v1/fs/complete-test.bin", nil)
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("stat: expected 200, got %d", resp.StatusCode)
	}
	if got := resp.Header.Get("X-Dat9-Checksum-SHA256"); got != checksum {
		t.Fatalf("stat checksum = %q, want %q", got, checksum)
	}
}

func TestV2UploadCompleteEndpointPersistsChecksum(t *testing.T) {
	s, s3c := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	ctx := context.Background()
	path := "/v2-checksum.bin"
	data := []byte("v2 checksum content")
	checksum := strings.Repeat("b", 64)
	plan := mustInitiateV2Upload(t, ts, path, int64(len(data)))
	if _, err := s.fallback.PresignPart(ctx, plan.UploadID, 1, nil); err != nil {
		t.Fatal(err)
	}
	upload, err := s.fallback.GetUpload(ctx, plan.UploadID)
	if err != nil {
		t.Fatal(err)
	}
	etag, err := s3c.UploadPart(ctx, upload.S3UploadID, 1, bytes.NewReader(data))
	if err != nil {
		t.Fatal(err)
	}
	completeBody, err := json.Marshal(map[string]any{
		"parts":           []backend.CompletePart{{Number: 1, ETag: etag}},
		"checksum_sha256": checksum,
	})
	if err != nil {
		t.Fatal(err)
	}
	req := mustNewRequest(t, http.MethodPost, ts.URL+"/v2/uploads/"+plan.UploadID+"/complete", bytes.NewReader(completeBody))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("complete status = %d, want 200", resp.StatusCode)
	}

	req = mustNewRequest(t, http.MethodHead, ts.URL+"/v1/fs"+path, nil)
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("HEAD status = %d, want 200", resp.StatusCode)
	}
	if got := resp.Header.Get("X-Dat9-Checksum-SHA256"); got != checksum {
		t.Fatalf("HEAD checksum = %q, want %q", got, checksum)
	}
}

func TestScopedMultipartCompleteRejectsTrustedChecksumButAllowsOmitted(t *testing.T) {
	checksum := strings.Repeat("c", 64)
	newScope := func(b *backend.Dat9Backend, path string) *TenantScope {
		return &TenantScope{TenantID: "tenant-scoped", Backend: b, IsScoped: true, FSScopes: []FSScope{{Prefix: path, Ops: map[FSOp]bool{FSOpWrite: true}}}}
	}

	t.Run("v1", func(t *testing.T) {
		s, s3c := newTestServerWithS3(t)
		data := []byte("v1-scoped")
		plan, err := s.fallback.InitiateUploadWithChecksums(context.Background(), "/scoped-v1.bin", int64(len(data)), strings.Split(partChecksumHeader(data), ","))
		if err != nil {
			t.Fatal(err)
		}
		upload, err := s.fallback.GetUpload(context.Background(), plan.UploadID)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := s3c.UploadPart(context.Background(), upload.S3UploadID, 1, bytes.NewReader(data)); err != nil {
			t.Fatal(err)
		}
		scope := newScope(s.fallback, "/scoped-v1.bin")
		req := httptest.NewRequest(http.MethodPost, "/v1/uploads/"+plan.UploadID+"/complete", strings.NewReader(`{"checksum_sha256":"`+checksum+`"}`))
		req = req.WithContext(withScope(req.Context(), scope))
		rr := httptest.NewRecorder()
		s.handleUploadComplete(rr, req, plan.UploadID)
		if rr.Code != http.StatusForbidden {
			t.Fatalf("checksum status=%d body=%s", rr.Code, rr.Body.String())
		}

		req = httptest.NewRequest(http.MethodPost, "/v1/uploads/"+plan.UploadID+"/complete", http.NoBody)
		req = req.WithContext(withScope(req.Context(), scope))
		rr = httptest.NewRecorder()
		s.handleUploadComplete(rr, req, plan.UploadID)
		if rr.Code != http.StatusOK {
			t.Fatalf("omitted checksum status=%d body=%s", rr.Code, rr.Body.String())
		}
	})

	t.Run("v2", func(t *testing.T) {
		s, s3c := newTestServerWithS3(t)
		data := []byte("v2-scoped")
		plan, err := s.fallback.InitiateUploadV2(context.Background(), "/scoped-v2.bin", int64(len(data)))
		if err != nil {
			t.Fatal(err)
		}
		if _, err := s.fallback.PresignPart(context.Background(), plan.UploadID, 1, nil); err != nil {
			t.Fatal(err)
		}
		upload, err := s.fallback.GetUpload(context.Background(), plan.UploadID)
		if err != nil {
			t.Fatal(err)
		}
		etag, err := s3c.UploadPart(context.Background(), upload.S3UploadID, 1, bytes.NewReader(data))
		if err != nil {
			t.Fatal(err)
		}
		parts := fmt.Sprintf(`[{"number":1,"etag":%q}]`, etag)
		scope := newScope(s.fallback, "/scoped-v2.bin")
		req := httptest.NewRequest(http.MethodPost, "/v2/uploads/"+plan.UploadID+"/complete", strings.NewReader(`{"parts":`+parts+`,"checksum_sha256":"`+checksum+`"}`))
		req = req.WithContext(withScope(req.Context(), scope))
		rr := httptest.NewRecorder()
		s.handleV2UploadComplete(rr, req, plan.UploadID)
		if rr.Code != http.StatusForbidden {
			t.Fatalf("checksum status=%d body=%s", rr.Code, rr.Body.String())
		}

		req = httptest.NewRequest(http.MethodPost, "/v2/uploads/"+plan.UploadID+"/complete", strings.NewReader(`{"parts":`+parts+`}`))
		req = req.WithContext(withScope(req.Context(), scope))
		rr = httptest.NewRecorder()
		s.handleV2UploadComplete(rr, req, plan.UploadID)
		if rr.Code != http.StatusOK {
			t.Fatalf("omitted checksum status=%d body=%s", rr.Code, rr.Body.String())
		}
	})
}

func TestUploadCompleteRejectsMalformedChecksum(t *testing.T) {
	t.Run("v1", func(t *testing.T) {
		s, _ := newTestServerWithS3(t)
		ts := httptest.NewServer(s)
		defer ts.Close()

		body := make([]byte, 1<<20)
		req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/bad-checksum-v1.bin", bytes.NewReader(body))
		req.ContentLength = int64(len(body))
		req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		var plan backend.UploadPlan
		if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
			t.Fatal(err)
		}
		_ = resp.Body.Close()

		req, _ = http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/complete", strings.NewReader(`{"checksum_sha256":"bad"}`))
		req.Header.Set("Content-Type", "application/json")
		resp, err = http.DefaultClient.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		defer func() { _ = resp.Body.Close() }()
		if resp.StatusCode != http.StatusBadRequest {
			t.Fatalf("status = %d, want 400", resp.StatusCode)
		}
		respBody, _ := io.ReadAll(resp.Body)
		if !strings.Contains(string(respBody), "checksum_sha256") {
			t.Fatalf("response = %q, want checksum validation error", respBody)
		}
	})

	t.Run("v2", func(t *testing.T) {
		s, _ := newTestServerWithS3(t)
		ts := httptest.NewServer(s)
		defer ts.Close()

		plan := mustInitiateV2Upload(t, ts, "/bad-checksum-v2.bin", 20<<20)
		presignBody := strings.NewReader(`{"part_number":1}`)
		presignReq := mustNewRequest(t, http.MethodPost, ts.URL+"/v2/uploads/"+plan.UploadID+"/presign", presignBody)
		presignReq.Header.Set("Content-Type", "application/json")
		presignResp, err := http.DefaultClient.Do(presignReq)
		if err != nil {
			t.Fatal(err)
		}
		_ = presignResp.Body.Close()
		if presignResp.StatusCode != http.StatusOK {
			t.Fatalf("presign status = %d, want 200", presignResp.StatusCode)
		}
		body := fmt.Sprintf(`{"parts":[{"number":1,"etag":"etag"}],"checksum_sha256":%q}`, strings.Repeat("A", 64))
		req := mustNewRequest(t, http.MethodPost, ts.URL+"/v2/uploads/"+plan.UploadID+"/complete", strings.NewReader(body))
		req.Header.Set("Content-Type", "application/json")
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		defer func() { _ = resp.Body.Close() }()
		if resp.StatusCode != http.StatusBadRequest {
			t.Fatalf("status = %d, want 400", resp.StatusCode)
		}
		respBody, _ := io.ReadAll(resp.Body)
		if !strings.Contains(string(respBody), "checksum_sha256") {
			t.Fatalf("response = %q, want checksum validation error", respBody)
		}
	})
}

func TestValidateWholeChecksumSHA256RejectsUppercaseHex(t *testing.T) {
	if err := validateWholeChecksumSHA256(strings.Repeat("a", 64)); err != nil {
		t.Fatalf("lowercase checksum rejected: %v", err)
	}
	if err := validateWholeChecksumSHA256(strings.Repeat("A", 64)); err == nil {
		t.Fatal("uppercase hexadecimal checksum unexpectedly accepted")
	}
}

func TestUploadResumeEndpoint(t *testing.T) {
	s, s3c := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	// Initiate a 20MB upload (3 parts)
	totalSize := int64(20 << 20)
	body := make([]byte, totalSize)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/resume-test.bin", bytes.NewReader(body))
	req.ContentLength = totalSize
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()

	upload, _ := s.fallback.GetUpload(context.Background(), plan.UploadID)

	// Upload only part 1
	if _, err := s3c.UploadPart(context.Background(), upload.S3UploadID, 1, bytes.NewReader(make([]byte, upload.PartSize))); err != nil {
		t.Fatal(err)
	}

	// POST /v1/uploads/{id}/resume
	req, _ = http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/resume", nil)
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		t.Fatalf("resume: expected 200, got %d: %s", resp.StatusCode, body)
	}

	var resumed backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&resumed); err != nil {
		t.Fatal(err)
	}
	if len(resumed.Parts) != 2 {
		t.Errorf("expected 2 missing parts, got %d", len(resumed.Parts))
	}
}

func TestUploadResumeEndpointByBody(t *testing.T) {
	s, s3c := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	totalSize := int64(20 << 20)
	body := make([]byte, totalSize)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/resume-body-test.bin", bytes.NewReader(body))
	req.ContentLength = totalSize
	checksumsHeader := partChecksumHeader(body)
	req.Header.Set("X-Dat9-Part-Checksums", checksumsHeader)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()

	upload, _ := s.fallback.GetUpload(context.Background(), plan.UploadID)
	if _, err := s3c.UploadPart(context.Background(), upload.S3UploadID, 1, bytes.NewReader(make([]byte, upload.PartSize))); err != nil {
		t.Fatal(err)
	}

	resumePayload, _ := json.Marshal(map[string]any{"part_checksums": strings.Split(checksumsHeader, ",")})
	req, _ = http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/resume", bytes.NewReader(resumePayload))
	req.Header.Set("Content-Type", "application/json")
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusOK {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("resume by body: expected 200, got %d: %s", resp.StatusCode, b)
	}

	var resumed backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&resumed); err != nil {
		t.Fatal(err)
	}
	if len(resumed.Parts) != 2 {
		t.Errorf("expected 2 missing parts, got %d", len(resumed.Parts))
	}
}

func TestV1UploadResumeRequiresChecksums(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	totalSize := int64(20 << 20)
	body := make([]byte, totalSize)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/resume-missing-checksums.bin", bytes.NewReader(body))
	req.ContentLength = totalSize
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()

	req, _ = http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/resume", nil)
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusBadRequest {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 400, got %d: %s", resp.StatusCode, b)
	}
	b, _ := io.ReadAll(resp.Body)
	if !strings.Contains(string(b), "missing X-Dat9-Part-Checksums header") {
		t.Fatalf("expected missing checksum header error, got %s", b)
	}
}

func TestV1UploadResumeByBodyRequiresPartChecksums(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	totalSize := int64(20 << 20)
	body := make([]byte, totalSize)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/resume-missing-body-checksums.bin", bytes.NewReader(body))
	req.ContentLength = totalSize
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()

	req, _ = http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/resume", strings.NewReader(`{}`))
	req.Header.Set("Content-Type", "application/json")
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusBadRequest {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 400, got %d: %s", resp.StatusCode, b)
	}
	b, _ := io.ReadAll(resp.Body)
	if !strings.Contains(string(b), "missing part_checksums") {
		t.Fatalf("expected missing part_checksums error, got %s", b)
	}
}

func TestV1PatchUsesFixedPartSizeBoundary(t *testing.T) {
	s, s3c := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	totalSize := int64(20 << 20)
	body := make([]byte, totalSize)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/patch-fixed-boundary.bin", bytes.NewReader(body))
	req.ContentLength = totalSize
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}

	var uploadPlan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&uploadPlan); err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()

	upload, err := s.fallback.GetUpload(context.Background(), uploadPlan.UploadID)
	if err != nil {
		t.Fatal(err)
	}
	for _, part := range uploadPlan.Parts {
		start := int64(part.Number-1) * s3client.PartSize
		end := start + part.Size
		if _, err := s3c.UploadPart(context.Background(), upload.S3UploadID, part.Number, bytes.NewReader(body[start:end])); err != nil {
			t.Fatalf("upload part %d: %v", part.Number, err)
		}
	}

	req, _ = http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+uploadPlan.UploadID+"/complete", nil)
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("complete: expected 200, got %d", resp.StatusCode)
	}

	patchReq := map[string]any{
		"new_size":    totalSize,
		"dirty_parts": []int{2},
	}
	p, _ := json.Marshal(patchReq)
	req, _ = http.NewRequest(http.MethodPatch, ts.URL+"/v1/fs/patch-fixed-boundary.bin", bytes.NewReader(p))
	req.Header.Set("Content-Type", "application/json")
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusAccepted {
		b, _ := io.ReadAll(resp.Body)
		t.Fatalf("patch: expected 202, got %d: %s", resp.StatusCode, b)
	}

	var plan backend.PatchPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	if plan.PartSize != s3client.PartSize {
		t.Fatalf("patch part size = %d, want %d", plan.PartSize, s3client.PartSize)
	}
	if len(plan.UploadParts) != 1 {
		t.Fatalf("upload parts = %d, want 1", len(plan.UploadParts))
	}
	part := plan.UploadParts[0]
	if part.Number != 2 {
		t.Fatalf("upload part number = %d, want 2", part.Number)
	}
	if part.Size != s3client.PartSize {
		t.Fatalf("upload part size = %d, want %d", part.Size, s3client.PartSize)
	}
	if part.ReadURL == "" {
		t.Fatal("expected read_url for dirty part within original file")
	}
	if got := part.ReadHeaders["Range"]; got != fmt.Sprintf("bytes=%d-%d", s3client.PartSize, 2*s3client.PartSize-1) {
		t.Fatalf("range header = %q, want %q", got, fmt.Sprintf("bytes=%d-%d", s3client.PartSize, 2*s3client.PartSize-1))
	}
	if len(plan.CopiedParts) != 2 || plan.CopiedParts[0] != 1 || plan.CopiedParts[1] != 3 {
		t.Fatalf("copied parts = %v, want [1 3]", plan.CopiedParts)
	}
}

func TestAppendEndpointUsesTailDirtyPartOnly(t *testing.T) {
	s, s3c := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	origSize := int64(2*s3client.PartSize + s3client.PartSize/2)
	body := make([]byte, origSize)
	mustUploadLargeServerFile(t, ts, s, s3c, "/append-tail.bin", body)

	appendReq := map[string]any{
		"append_size": s3client.PartSize + s3client.PartSize/4,
		"part_size":   s3client.PartSize,
	}
	reqBody, _ := json.Marshal(appendReq)
	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/fs/append-tail.bin?append", bytes.NewReader(reqBody))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusAccepted {
		respBody, _ := io.ReadAll(resp.Body)
		t.Fatalf("append initiate: got %d, want 202: %s", resp.StatusCode, respBody)
	}

	var plan backend.AppendPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatalf("decode append plan: %v", err)
	}
	if plan.BaseSize != origSize {
		t.Fatalf("base size = %d, want %d", plan.BaseSize, origSize)
	}
	if plan.PartSize != s3client.PartSize {
		t.Fatalf("part size = %d, want %d", plan.PartSize, s3client.PartSize)
	}
	if len(plan.UploadParts) != 2 {
		t.Fatalf("upload parts = %d, want 2", len(plan.UploadParts))
	}
	if plan.UploadParts[0].Number != 3 || plan.UploadParts[1].Number != 4 {
		t.Fatalf("upload part numbers = [%d %d], want [3 4]", plan.UploadParts[0].Number, plan.UploadParts[1].Number)
	}
	if plan.UploadParts[0].Size != s3client.PartSize {
		t.Fatalf("part 3 size = %d, want %d", plan.UploadParts[0].Size, s3client.PartSize)
	}
	if plan.UploadParts[1].Size != 3*s3client.PartSize/4 {
		t.Fatalf("part 4 size = %d, want %d", plan.UploadParts[1].Size, 3*s3client.PartSize/4)
	}
	if plan.UploadParts[0].ReadURL == "" {
		t.Fatal("expected read_url for tail part")
	}
	wantRange := fmt.Sprintf("bytes=%d-%d", 2*s3client.PartSize, origSize-1)
	if got := plan.UploadParts[0].ReadHeaders["Range"]; got != wantRange {
		t.Fatalf("tail part range = %q, want %q", got, wantRange)
	}
	if plan.UploadParts[1].ReadURL != "" {
		t.Fatal("new append-only part should not have read_url")
	}
	if len(plan.CopiedParts) != 2 || plan.CopiedParts[0] != 1 || plan.CopiedParts[1] != 2 {
		t.Fatalf("copied parts = %v, want [1 2]", plan.CopiedParts)
	}
}

func TestLargeUploadOverwritesExistingSmallFile(t *testing.T) {
	s, s3c := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	// Seed an existing small file at the target path.
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/overwrite.bin", bytes.NewReader([]byte("small")))
	req.ContentLength = 5
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("small seed: expected 200, got %d", resp.StatusCode)
	}

	// Initiate a large upload to the same path.
	totalSize := int64(2 << 20)
	largeBody := make([]byte, totalSize)
	req, _ = http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/overwrite.bin", bytes.NewReader(largeBody))
	req.ContentLength = totalSize
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(largeBody))
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatalf("decode plan: %v", err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("initiate overwrite: expected 202, got %d", resp.StatusCode)
	}

	upload, err := s.fallback.GetUpload(context.Background(), plan.UploadID)
	if err != nil {
		t.Fatal(err)
	}

	// Upload all parts through the local S3 stand-in.
	for _, p := range plan.Parts {
		start := int64(p.Number-1) * plan.PartSize
		end := start + p.Size
		if _, err := s3c.UploadPart(context.Background(), upload.S3UploadID, p.Number,
			bytes.NewReader(make([]byte, end-start))); err != nil {
			t.Fatalf("upload part %d: %v", p.Number, err)
		}
	}

	// Complete should now overwrite the existing small-file node.
	req, _ = http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/complete", nil)
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("complete overwrite: expected 200, got %d", resp.StatusCode)
	}

	nf, err := s.fallback.Store().Stat(context.Background(), "/overwrite.bin")
	if err != nil {
		t.Fatal(err)
	}
	if nf.File == nil || nf.File.StorageType != datastore.StorageS3 {
		t.Fatalf("expected overwrite.bin to point at S3-backed file, got %+v", nf.File)
	}
	if nf.File.SizeBytes != totalSize {
		t.Fatalf("expected size %d, got %d", totalSize, nf.File.SizeBytes)
	}
}

func TestAutoImageMultipartOverwriteWritesContentTextEndToEnd(t *testing.T) {
	s, s3c := newTestServerWithS3Config(t, backend.Options{
		DatabaseAutoEmbedding: true,
		AsyncImageExtract: backend.AsyncImageExtractOptions{
			Enabled:   true,
			Workers:   1,
			QueueSize: 8,
			Extractor: staticServerImageExtractor{text: "caption from multipart overwrite"},
		},
	}, TenantWorkerOptions{
		Workers:       1,
		PollInterval:  10 * time.Millisecond,
		LeaseDuration: 200 * time.Millisecond,
	})
	ts := httptest.NewServer(s)
	defer ts.Close()

	seedReq, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/overwrite-auto.png", bytes.NewReader(fakeImagePNG))
	seedReq.ContentLength = int64(len(fakeImagePNG))
	seedResp, err := http.DefaultClient.Do(seedReq)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = seedResp.Body.Close() }()
	if seedResp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(seedResp.Body)
		t.Fatalf("seed put: expected 200, got %d: %s", seedResp.StatusCode, body)
	}

	original := mustServerFile(t, s.fallback, "/overwrite-auto.png")
	waitForContentTextOnServer(t, s.fallback, "/overwrite-auto.png", "caption from multipart overwrite", 3*time.Second)
	waitForTaskStatus(t, s.fallback, original.FileID, 1, string(semantic.TaskSucceeded), 3*time.Second)

	totalSize := int64(2 << 20)
	body := make([]byte, totalSize)
	// Prefix the multipart content with the PNG magic bytes so the extraction
	// path's raster-image sniff accepts it.
	copy(body, fakeImagePNG)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/overwrite-auto.png", bytes.NewReader(body))
	req.ContentLength = totalSize
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatalf("decode plan: %v", err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("initiate overwrite: expected 202, got %d", resp.StatusCode)
	}

	upload, err := s.fallback.GetUpload(context.Background(), plan.UploadID)
	if err != nil {
		t.Fatal(err)
	}
	for _, part := range plan.Parts {
		start := int64(part.Number-1) * plan.PartSize
		end := start + part.Size
		if _, err := s3c.UploadPart(context.Background(), upload.S3UploadID, part.Number, bytes.NewReader(body[start:end])); err != nil {
			t.Fatalf("upload part %d: %v", part.Number, err)
		}
	}

	completeReq, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/"+plan.UploadID+"/complete", nil)
	completeResp, err := http.DefaultClient.Do(completeReq)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = completeResp.Body.Close() }()
	if completeResp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(completeResp.Body)
		t.Fatalf("complete overwrite: expected 200, got %d: %s", completeResp.StatusCode, body)
	}

	updated := mustServerFile(t, s.fallback, "/overwrite-auto.png")
	if updated.FileID != original.FileID {
		t.Fatalf("overwrite file_id=%q, want %q", updated.FileID, original.FileID)
	}
	if updated.Revision != 2 {
		t.Fatalf("revision=%d, want 2", updated.Revision)
	}
	waitForContentTextOnServer(t, s.fallback, "/overwrite-auto.png", "caption from multipart overwrite", 3*time.Second)
	waitForTaskStatus(t, s.fallback, updated.FileID, 2, string(semantic.TaskSucceeded), 3*time.Second)

	tasks := loadSemanticTaskRowsForResource(t, s.fallback, updated.FileID)
	if len(tasks) != 2 {
		t.Fatalf("semantic task count=%d, want 2", len(tasks))
	}
	for _, task := range tasks {
		if task.TaskType != string(semantic.TaskTypeImgExtractText) || task.Status != string(semantic.TaskSucceeded) {
			t.Fatalf("unexpected semantic task rows: %+v", tasks)
		}
	}
}

func TestListUploadsEndpoint(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	// Create one upload
	body := make([]byte, 1<<20)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/list-test.bin", bytes.NewReader(body))
	req.ContentLength = int64(len(body))
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, _ := http.DefaultClient.Do(req)
	_ = resp.Body.Close()

	// GET /v1/uploads?path=/list-test.bin&status=UPLOADING
	resp, err := http.Get(ts.URL + "/v1/uploads?path=/list-test.bin&status=UPLOADING")
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusOK {
		t.Fatalf("expected 200, got %d", resp.StatusCode)
	}

	var result struct {
		Uploads []struct {
			UploadID   string `json:"upload_id"`
			PartsTotal int    `json:"parts_total"`
			Status     string `json:"status"`
		} `json:"uploads"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		t.Fatal(err)
	}
	if len(result.Uploads) != 1 {
		t.Errorf("expected 1 upload, got %d", len(result.Uploads))
	}
	if result.Uploads[0].PartsTotal != 1 {
		t.Errorf("expected parts_total=1, got %d", result.Uploads[0].PartsTotal)
	}
}

func TestListUploadsEndpointRootPathReturnsNotFound(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	resp, err := http.Get(ts.URL + "/v1/uploads?path=/&status=UPLOADING")
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode != http.StatusNotFound {
		t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusNotFound)
	}
}

// A second initiate for the same path supersedes the previous active session
// (which a crashed client can never abort) instead of 409ing until expires_at.
func TestSecondUploadSamePathSupersedesFirst(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	// First upload should succeed with 202
	body := make([]byte, 1<<20)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/dup-test.bin", bytes.NewReader(body))
	req.ContentLength = int64(len(body))
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, _ := http.DefaultClient.Do(req)
	var plan1 backend.UploadPlan
	_ = json.NewDecoder(resp.Body).Decode(&plan1)
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("first upload: expected 202, got %d", resp.StatusCode)
	}

	// Second upload for same path supersedes the dangling first session.
	req, _ = http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/dup-test.bin", bytes.NewReader(body))
	req.ContentLength = int64(len(body))
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, _ = http.DefaultClient.Do(req)
	var plan2 backend.UploadPlan
	_ = json.NewDecoder(resp.Body).Decode(&plan2)
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusAccepted {
		t.Fatalf("second upload: expected 202 (supersede), got %d", resp.StatusCode)
	}
	if plan2.UploadID == plan1.UploadID {
		t.Fatal("expected a new upload record for the superseding initiate")
	}

	// Only the superseding session remains active.
	listResp, err := http.Get(ts.URL + "/v1/uploads?path=/dup-test.bin&status=UPLOADING")
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = listResp.Body.Close() }()
	var result struct {
		Uploads []struct {
			UploadID string `json:"upload_id"`
		} `json:"uploads"`
	}
	if err := json.NewDecoder(listResp.Body).Decode(&result); err != nil {
		t.Fatal(err)
	}
	if len(result.Uploads) != 1 || result.Uploads[0].UploadID != plan2.UploadID {
		t.Fatalf("active uploads = %+v, want only %s", result.Uploads, plan2.UploadID)
	}
}

func TestAbortUploadEndpoint(t *testing.T) {
	s, _ := newTestServerWithS3(t)
	ts := httptest.NewServer(s)
	defer ts.Close()

	// Create upload
	body := make([]byte, 1<<20)
	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/abort-test.bin", bytes.NewReader(body))
	req.ContentLength = int64(len(body))
	req.Header.Set("X-Dat9-Part-Checksums", partChecksumHeader(body))
	resp, _ := http.DefaultClient.Do(req)

	var plan backend.UploadPlan
	if err := json.NewDecoder(resp.Body).Decode(&plan); err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()

	// DELETE /v1/uploads/{id}
	req, _ = http.NewRequest(http.MethodDelete, ts.URL+"/v1/uploads/"+plan.UploadID, nil)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		t.Fatalf("abort: expected 200, got %d", resp.StatusCode)
	}

	// Verify upload is aborted
	upload, _ := s.fallback.GetUpload(context.Background(), plan.UploadID)
	if upload.Status != datastore.UploadAborted {
		t.Errorf("expected ABORTED, got %s", upload.Status)
	}
}

func TestParsePartChecksumsHeaderValidation(t *testing.T) {
	if _, err := parsePartChecksumsHeader(""); err != nil {
		t.Fatalf("empty header should be allowed: %v", err)
	}

	_, err := parsePartChecksumsHeader("not-base64")
	if err == nil || !strings.Contains(err.Error(), "invalid base64") {
		t.Fatalf("expected base64 error, got %v", err)
	}

	short := base64.StdEncoding.EncodeToString([]byte("short"))
	_, err = parsePartChecksumsHeader(short)
	if err == nil || !strings.Contains(err.Error(), "expected 4") {
		t.Fatalf("expected decoded length error, got %v", err)
	}
}

func TestUploadRespectsMaxUploadBytes(t *testing.T) {
	base, _ := newTestServerWithS3(t)
	s := NewWithConfig(Config{Backend: base.fallback, MaxUploadBytes: 10})
	ts := httptest.NewServer(s)
	defer ts.Close()

	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/too-big.bin", bytes.NewReader([]byte("12345678901")))
	req.ContentLength = 11
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusRequestEntityTooLarge {
		body, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 413, got %d: %s", resp.StatusCode, body)
	}
}

func TestNewWithConfigUsesDefaultMaxUploadBytes(t *testing.T) {
	base, _ := newTestServerWithS3(t)
	s := NewWithConfig(Config{Backend: base.fallback})
	if s.maxUploadBytes != DefaultMaxUploadBytes {
		t.Fatalf("default maxUploadBytes = %d, want %d", s.maxUploadBytes, DefaultMaxUploadBytes)
	}
}

func TestNewWithConfigUsesDefaultTenantPoolMaxSize(t *testing.T) {
	s := NewWithConfig(Config{})
	if s.tenantPoolMaxSize != DefaultTenantPoolMaxSize {
		t.Fatalf("default tenantPoolMaxSize = %d, want %d", s.tenantPoolMaxSize, DefaultTenantPoolMaxSize)
	}
}

func TestNewWithConfigUsesDefaultTenantPoolRefillFreeRatio(t *testing.T) {
	s := NewWithConfig(Config{})
	if s.tenantPoolRefillFreeRatio != DefaultTenantPoolRefillFreeRatio {
		t.Fatalf("default tenantPoolRefillFreeRatio = %f, want %f", s.tenantPoolRefillFreeRatio, DefaultTenantPoolRefillFreeRatio)
	}
}

func TestNewWithConfigRejectsNaNTenantPoolRefillFreeRatio(t *testing.T) {
	s := NewWithConfig(Config{TenantPoolRefillFreeRatio: math.NaN()})
	if s.tenantPoolRefillFreeRatio != DefaultTenantPoolRefillFreeRatio {
		t.Fatalf("NaN tenantPoolRefillFreeRatio = %f, want %f", s.tenantPoolRefillFreeRatio, DefaultTenantPoolRefillFreeRatio)
	}
}

func TestNewWithConfigUsesSharedDBReopenRatio(t *testing.T) {
	s := NewWithConfig(Config{SharedDBReopenRatio: 0.65})
	if s.sharedDBReopenRatio != 0.65 {
		t.Fatalf("sharedDBReopenRatio = %f, want 0.65", s.sharedDBReopenRatio)
	}
}

func TestNewWithConfigUsesDefaultSharedDBReopenRatio(t *testing.T) {
	s := NewWithConfig(Config{})
	if s.sharedDBReopenRatio != DefaultSharedDBReopenRatio {
		t.Fatalf("default sharedDBReopenRatio = %f, want %f", s.sharedDBReopenRatio, DefaultSharedDBReopenRatio)
	}
}

func TestNewWithConfigRejectsInvalidSharedDBReopenRatio(t *testing.T) {
	for _, ratio := range []float64{0, 1, math.NaN(), math.Inf(1)} {
		s := NewWithConfig(Config{SharedDBReopenRatio: ratio})
		if s.sharedDBReopenRatio != DefaultSharedDBReopenRatio {
			t.Fatalf("invalid sharedDBReopenRatio %v = %f, want %f", ratio, s.sharedDBReopenRatio, DefaultSharedDBReopenRatio)
		}
	}
}

func TestDeclaredContentLengthOverMaxRejected(t *testing.T) {
	base, _ := newTestServerWithS3(t)
	s := NewWithConfig(Config{Backend: base.fallback, MaxUploadBytes: 10})
	ts := httptest.NewServer(s)
	defer ts.Close()

	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/declared-too-big.bin", http.NoBody)
	req.Header.Set("X-Dat9-Content-Length", "11")
	req.ContentLength = 0
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusRequestEntityTooLarge {
		body, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 413, got %d: %s", resp.StatusCode, body)
	}
}

func TestContentLengthHeaderMismatchRejected(t *testing.T) {
	base, _ := newTestServerWithS3(t)
	s := NewWithConfig(Config{Backend: base.fallback, MaxUploadBytes: 1 << 20})
	ts := httptest.NewServer(s)
	defer ts.Close()

	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/mismatch.bin", bytes.NewReader([]byte("1234")))
	req.ContentLength = 4
	req.Header.Set("X-Dat9-Content-Length", "5")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusBadRequest {
		body, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 400, got %d: %s", resp.StatusCode, body)
	}
}

func TestWriteReturns507WhenTenantStorageQuotaExceeded(t *testing.T) {
	s, _ := newTestServerWithS3Config(t, backend.Options{MaxTenantStorageBytes: 10}, TenantWorkerOptions{})
	ts := httptest.NewServer(s)
	defer ts.Close()

	req, _ := http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/fill.txt", bytes.NewReader([]byte("1234567890")))
	req.ContentLength = 10
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("expected first write 200, got %d", resp.StatusCode)
	}

	req, _ = http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/overflow.txt", bytes.NewReader([]byte("1")))
	req.ContentLength = 1
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusInsufficientStorage {
		body, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 507, got %d: %s", resp.StatusCode, body)
	}
}

func TestUploadInitiateReturns507WhenTenantStorageQuotaExceeded(t *testing.T) {
	s, _ := newTestServerWithS3Config(t, backend.Options{MaxTenantStorageBytes: 70_000}, TenantWorkerOptions{})
	if _, err := s.fallback.Write("/filled.bin", bytes.Repeat([]byte("a"), 60_000), 0, filesystem.WriteFlagCreate); err != nil {
		t.Fatal(err)
	}
	ts := httptest.NewServer(s)
	defer ts.Close()

	body := bytes.Repeat([]byte("b"), 20_000)
	reqBody := map[string]any{
		"path":           "/quota-over.bin",
		"total_size":     len(body),
		"part_checksums": strings.Split(partChecksumHeader(body), ","),
	}
	p, _ := json.Marshal(reqBody)
	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/uploads/initiate", bytes.NewReader(p))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusInsufficientStorage {
		body, _ := io.ReadAll(resp.Body)
		t.Fatalf("expected 507, got %d: %s", resp.StatusCode, body)
	}
}