package server

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strconv"
	"testing"

	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/tenant"
	"github.com/mem9-ai/drive9/pkg/testtidb"
)

func TestParseAppendLogRange(t *testing.T) {
	tests := []struct {
		name        string
		raw         string
		size        int64
		wantStart   int64
		wantEnd     int64
		wantPartial bool
		wantErr     bool
	}{
		{name: "whole", size: 10, wantStart: 0, wantEnd: 9},
		{name: "bounded", raw: "bytes=2-5", size: 10, wantStart: 2, wantEnd: 5, wantPartial: true},
		{name: "open ended", raw: "bytes=7-", size: 10, wantStart: 7, wantEnd: 9, wantPartial: true},
		{name: "suffix", raw: "bytes=-3", size: 10, wantStart: 7, wantEnd: 9, wantPartial: true},
		{name: "oversized suffix", raw: "bytes=-20", size: 10, wantStart: 0, wantEnd: 9, wantPartial: true},
		{name: "clamped end", raw: "bytes=7-20", size: 10, wantStart: 7, wantEnd: 9, wantPartial: true},
		{name: "empty", size: 0, wantStart: 0, wantEnd: -1},
		{name: "empty range", raw: "bytes=0-0", size: 0, wantErr: true},
		{name: "multiple", raw: "bytes=0-1,3-4", size: 10, wantErr: true},
		{name: "invalid suffix", raw: "bytes=-0", size: 10, wantErr: true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			start, end, partial, err := parseAppendLogRange(tt.raw, tt.size)
			if (err != nil) != tt.wantErr {
				t.Fatalf("parseAppendLogRange(%q, %d) error = %v, wantErr=%v", tt.raw, tt.size, err, tt.wantErr)
			}
			if tt.wantErr {
				return
			}
			if start != tt.wantStart || end != tt.wantEnd || partial != tt.wantPartial {
				t.Fatalf("parseAppendLogRange(%q, %d) = (%d, %d, %v), want (%d, %d, %v)", tt.raw, tt.size, start, end, partial, tt.wantStart, tt.wantEnd, tt.wantPartial)
			}
		})
	}
}

func TestScopedAppendLogPostQueryAllowed(t *testing.T) {
	if !isScopedFSPostQueryAllowed(url.Values{"append-log": {""}}) {
		t.Fatal("scoped append-log POST was rejected")
	}
	if isScopedFSPostQueryAllowed(url.Values{"append-log": {""}, "unexpected": {"1"}}) {
		t.Fatal("scoped append-log POST accepted an unknown query key")
	}
}

func newAppendLogServer(t *testing.T) *Server {
	t.Helper()
	ctx := context.Background()
	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(t.TempDir(), "http://localhost:9091/s3")
	if err != nil {
		t.Fatal(err)
	}
	b, err := backend.NewWithS3ModeAndOptions(store, s3c, true, backend.Options{
		AppendLogS3:        s3c,
		InlineThreshold:    1,
		TenantID:           "append-log-server-test",
		StorageNamespaceID: "append-log-server-test",
	})
	if err != nil {
		t.Fatal(err)
	}
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	testtidb.ResetMetaDB(t, metaStore.DB())
	t.Cleanup(func() { _ = metaStore.Close() })
	b.SetMetaQuotaStore(ctx, "append-log-server-test", tenant.NewMetaQuotaAdapter(metaStore))
	s := NewWithConfig(Config{Backend: b})
	t.Cleanup(s.Close)
	return s
}

type appendLogErrorReader struct {
	err error
}

func (r appendLogErrorReader) Read([]byte) (int, error) {
	return 0, r.err
}

func TestAppendLogHTTPCreateReadRewriteAndUnsupportedPlans(t *testing.T) {
	s := newAppendLogServer(t)
	ts := httptest.NewServer(s)
	defer ts.Close()
	writeBytesBefore := tenantFileBytesMetricValue(t, "write", "s3")
	writeOpsBefore := tenantFileOpsMetricValue(t, "write", "s3")
	readBytesBefore := tenantFileBytesMetricValue(t, "read", "s3")
	readOpsBefore := tenantFileOpsMetricValue(t, "read", "s3")

	appendRequest := func(body string, revision, size int64) *http.Response {
		t.Helper()
		req, err := http.NewRequest(http.MethodPost, ts.URL+"/v1/fs/wal?append-log", bytes.NewReader([]byte(body)))
		if err != nil {
			t.Fatal(err)
		}
		req.Header.Set("X-Dat9-Expected-Revision", strconv.FormatInt(revision, 10))
		req.Header.Set("X-Dat9-Expected-Size", strconv.FormatInt(size, 10))
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		return resp
	}

	resp := appendRequest("head", 0, 0)
	var created struct {
		Revision int64 `json:"revision"`
		Size     int64 `json:"size_bytes"`
	}
	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		_ = resp.Body.Close()
		t.Fatalf("append-log create status=%d body=%s", resp.StatusCode, body)
	}
	if err := json.NewDecoder(resp.Body).Decode(&created); err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if created.Revision != 1 || created.Size != 4 {
		t.Fatalf("create response = %+v, want revision=1 size=4", created)
	}

	resp = appendRequest("tail", 1, 4)
	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		_ = resp.Body.Close()
		t.Fatalf("append-log append status=%d body=%s", resp.StatusCode, body)
	}
	_ = resp.Body.Close()
	requireTenantFileBytesDelta(t, writeBytesBefore, 8, "write", "s3")
	requireTenantFileOpsDelta(t, writeOpsBefore, 2, "write", "s3")

	resp, err := http.Get(ts.URL + "/v1/fs/wal")
	if err != nil {
		t.Fatal(err)
	}
	data, _ := io.ReadAll(resp.Body)
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusOK || string(data) != "headtail" {
		t.Fatalf("append-log GET = status=%d body=%q, want 200 headtail", resp.StatusCode, data)
	}

	req, err := http.NewRequest(http.MethodGet, ts.URL+"/v1/fs/wal", nil)
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("Range", "bytes=-4")
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	data, _ = io.ReadAll(resp.Body)
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusPartialContent || string(data) != "tail" || resp.Header.Get("Content-Range") != "bytes 4-7/8" {
		t.Fatalf("append-log range GET = status=%d range=%q body=%q", resp.StatusCode, resp.Header.Get("Content-Range"), data)
	}
	requireTenantFileBytesDelta(t, readBytesBefore, 12, "read", "s3")
	requireTenantFileOpsDelta(t, readOpsBefore, 2, "read", "s3")

	resp, err = http.Head(ts.URL + "/v1/fs/wal")
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if resp.Header.Get("X-Dat9-Content-Layout") != "append_log" || resp.Header.Get("X-Dat9-Revision") != "2" {
		t.Fatalf("append-log HEAD layout=%q revision=%q", resp.Header.Get("X-Dat9-Content-Layout"), resp.Header.Get("X-Dat9-Revision"))
	}

	req, err = http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/wal", bytes.NewReader([]byte("reset")))
	if err != nil {
		t.Fatal(err)
	}
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	if resp.StatusCode != http.StatusBadRequest {
		body, _ := io.ReadAll(resp.Body)
		_ = resp.Body.Close()
		t.Fatalf("unconditional append-log PUT status=%d body=%s", resp.StatusCode, body)
	}
	_ = resp.Body.Close()

	req, err = http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/wal", bytes.NewReader([]byte("reset")))
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("X-Dat9-Expected-Revision", "2")
	req.Header.Add("X-Dat9-Tag", "owner=alice")
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	if resp.StatusCode != http.StatusBadRequest {
		body, _ := io.ReadAll(resp.Body)
		_ = resp.Body.Close()
		t.Fatalf("tagged append-log PUT status=%d body=%s, want 400", resp.StatusCode, body)
	}
	_ = resp.Body.Close()

	req, err = http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/wal", bytes.NewReader([]byte("reset")))
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("X-Dat9-Expected-Revision", "2")
	req.Header.Set("X-Dat9-Description", "replacement description")
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	if resp.StatusCode != http.StatusBadRequest {
		body, _ := io.ReadAll(resp.Body)
		_ = resp.Body.Close()
		t.Fatalf("described append-log PUT status=%d body=%s, want 400", resp.StatusCode, body)
	}
	_ = resp.Body.Close()

	req, err = http.NewRequest(http.MethodPut, ts.URL+"/v1/fs/wal", bytes.NewReader([]byte("reset")))
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("X-Dat9-Expected-Revision", "2")
	resp, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		_ = resp.Body.Close()
		t.Fatalf("conditional append-log PUT status=%d body=%s", resp.StatusCode, body)
	}
	_ = resp.Body.Close()
	requireTenantFileBytesDelta(t, writeBytesBefore, 13, "write", "s3")
	requireTenantFileOpsDelta(t, writeOpsBefore, 3, "write", "s3")

	for _, endpoint := range []string{"/v1/fs/wal?append", "/v1/fs/wal?patch"} {
		method := http.MethodPost
		if endpoint == "/v1/fs/wal?patch" {
			method = http.MethodPatch
		}
		req, err = http.NewRequest(method, ts.URL+endpoint, nil)
		if err != nil {
			t.Fatal(err)
		}
		resp, err = http.DefaultClient.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		if resp.StatusCode != http.StatusBadRequest {
			body, _ := io.ReadAll(resp.Body)
			_ = resp.Body.Close()
			t.Fatalf("append-log unsupported endpoint %s status=%d body=%s", endpoint, resp.StatusCode, body)
		}
		_ = resp.Body.Close()
	}
}

func TestAppendLogEmptyReadRecordsOperation(t *testing.T) {
	s := newAppendLogServer(t)
	ctx := context.Background()
	if _, err := s.fallback.AppendLog(ctx, "/empty", bytes.NewReader(nil), 0, 0, 0); err != nil {
		t.Fatalf("create empty append-log: %v", err)
	}
	readBytesBefore := tenantFileBytesMetricValue(t, "read", "s3")
	readOpsBefore := tenantFileOpsMetricValue(t, "read", "s3")

	ts := httptest.NewServer(s)
	defer ts.Close()
	resp, err := http.Get(ts.URL + "/v1/fs/empty")
	if err != nil {
		t.Fatal(err)
	}
	data, _ := io.ReadAll(resp.Body)
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusOK || len(data) != 0 {
		t.Fatalf("empty append-log GET status=%d body=%q", resp.StatusCode, data)
	}
	requireTenantFileBytesDelta(t, readBytesBefore, 0, "read", "s3")
	requireTenantFileOpsDelta(t, readOpsBefore, 1, "read", "s3")
}

func TestAppendLogDeniedBeforeBackendAccess(t *testing.T) {
	s := &Server{}
	r := httptest.NewRequest(http.MethodPost, "/v1/fs/denied?append-log", bytes.NewReader([]byte("x")))
	r.Header.Set("X-Dat9-Expected-Revision", "0")
	r.Header.Set("X-Dat9-Expected-Size", "0")
	r = r.WithContext(withScope(r.Context(), &TenantScope{
		IsScoped: true,
		FSScopes: []FSScope{{
			Prefix: "/allowed",
			Ops:    map[FSOp]bool{FSOpWrite: true},
		}},
	}))
	w := httptest.NewRecorder()
	s.handleAppendLog(w, r, "/denied")
	if w.Code != http.StatusForbidden {
		t.Fatalf("denied append-log status=%d body=%s, want 403 before backend access", w.Code, w.Body.String())
	}
}

func TestAppendLogRejectsRootPath(t *testing.T) {
	s := newAppendLogServer(t)
	r := httptest.NewRequest(http.MethodPost, "/v1/fs/?append-log", bytes.NewReader([]byte("x")))
	r.Header.Set("X-Dat9-Expected-Revision", "0")
	r.Header.Set("X-Dat9-Expected-Size", "0")
	r = r.WithContext(withScope(r.Context(), &TenantScope{Backend: s.fallback}))
	w := httptest.NewRecorder()
	s.handleAppendLog(w, r, "/")
	if w.Code != http.StatusBadRequest {
		t.Fatalf("root append-log status=%d body=%s, want 400", w.Code, w.Body.String())
	}
}

func TestAppendLogRejectsInvalidPath(t *testing.T) {
	s := newAppendLogServer(t)
	r := httptest.NewRequest(http.MethodPost, "/v1/fs/ignored?append-log", bytes.NewReader([]byte("x")))
	r.Header.Set("X-Dat9-Expected-Revision", "0")
	r.Header.Set("X-Dat9-Expected-Size", "0")
	r = r.WithContext(withScope(r.Context(), &TenantScope{Backend: s.fallback}))
	w := httptest.NewRecorder()
	s.handleAppendLog(w, r, "/invalid\\path")
	if w.Code != http.StatusBadRequest {
		t.Fatalf("invalid append-log path status=%d body=%s, want 400", w.Code, w.Body.String())
	}
}

func TestAppendLogMultipartFinalizeRaceReturnsBadRequest(t *testing.T) {
	ctx := context.Background()
	assertUnsupported := func(t *testing.T, w *httptest.ResponseRecorder) {
		t.Helper()
		if w.Code != http.StatusBadRequest {
			t.Fatalf("status=%d body=%s, want 400", w.Code, w.Body.String())
		}
		var response struct {
			Code string `json:"code"`
		}
		if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {
			t.Fatalf("decode response: %v", err)
		}
		if response.Code != "append_log_unsupported" {
			t.Fatalf("response code=%q, want append_log_unsupported", response.Code)
		}
	}

	t.Run("v1", func(t *testing.T) {
		s := newAppendLogServer(t)
		local, ok := s.fallback.S3().(*s3client.LocalS3Client)
		if !ok {
			t.Fatalf("S3 client = %T, want local client", s.fallback.S3())
		}
		payload := []byte("v1 race")
		plan, err := s.fallback.InitiateUpload(ctx, "/v1-race", int64(len(payload)))
		if err != nil {
			t.Fatalf("initiate V1 upload: %v", err)
		}
		upload, err := s.fallback.GetUpload(ctx, plan.UploadID)
		if err != nil {
			t.Fatalf("get V1 upload: %v", err)
		}
		for _, part := range plan.Parts {
			if _, err := local.UploadPart(ctx, upload.S3UploadID, part.Number, bytes.NewReader(payload)); err != nil {
				t.Fatalf("upload V1 part: %v", err)
			}
		}
		if _, err := s.fallback.AppendLog(ctx, "/v1-race", bytes.NewReader([]byte("seed")), 4, 0, 0); err != nil {
			t.Fatalf("replace V1 target with append-log: %v", err)
		}
		r := httptest.NewRequest(http.MethodPost, "/v1/uploads/"+plan.UploadID+"/complete", http.NoBody)
		r = r.WithContext(withScope(r.Context(), &TenantScope{Backend: s.fallback}))
		w := httptest.NewRecorder()
		s.handleUploadComplete(w, r, plan.UploadID)
		assertUnsupported(t, w)
	})

	t.Run("v2", func(t *testing.T) {
		s := newAppendLogServer(t)
		local, ok := s.fallback.S3().(*s3client.LocalS3Client)
		if !ok {
			t.Fatalf("S3 client = %T, want local client", s.fallback.S3())
		}
		payload := []byte("v2 race")
		plan, err := s.fallback.InitiateUploadV2(ctx, "/v2-race", int64(len(payload)))
		if err != nil {
			t.Fatalf("initiate V2 upload: %v", err)
		}
		if _, err := s.fallback.PresignPart(ctx, plan.UploadID, 1, nil); err != nil {
			t.Fatalf("presign V2 part: %v", err)
		}
		upload, err := s.fallback.GetUpload(ctx, plan.UploadID)
		if err != nil {
			t.Fatalf("get V2 upload: %v", err)
		}
		etag, err := local.UploadPart(ctx, upload.S3UploadID, 1, bytes.NewReader(payload))
		if err != nil {
			t.Fatalf("upload V2 part: %v", err)
		}
		if _, err := s.fallback.AppendLog(ctx, "/v2-race", bytes.NewReader([]byte("seed")), 4, 0, 0); err != nil {
			t.Fatalf("replace V2 target with append-log: %v", err)
		}
		body, err := json.Marshal(map[string]any{"parts": []backend.CompletePart{{Number: 1, ETag: etag}}})
		if err != nil {
			t.Fatalf("marshal V2 completion: %v", err)
		}
		r := httptest.NewRequest(http.MethodPost, "/v2/uploads/"+plan.UploadID+"/complete", bytes.NewReader(body))
		r = r.WithContext(withScope(r.Context(), &TenantScope{Backend: s.fallback}))
		w := httptest.NewRecorder()
		s.handleV2UploadComplete(w, r, plan.UploadID)
		assertUnsupported(t, w)
	})
}

func TestAppendLogV2UploadInitiateIsUnsupported(t *testing.T) {
	ctx := context.Background()
	s := newAppendLogServer(t)
	if _, err := s.fallback.AppendLog(ctx, "/wal", bytes.NewReader([]byte("head")), 4, 0, 0); err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	body := []byte(`{"path":"/wal","total_size":4,"expected_revision":1}`)
	r := httptest.NewRequest(http.MethodPost, "/v2/uploads/initiate", bytes.NewReader(body))
	r = r.WithContext(withScope(r.Context(), &TenantScope{Backend: s.fallback}))
	w := httptest.NewRecorder()
	s.handleV2UploadInitiate(w, r)
	if w.Code != http.StatusBadRequest {
		t.Fatalf("V2 append-log initiate status=%d body=%s, want 400", w.Code, w.Body.String())
	}
	var response struct {
		Code string `json:"code"`
	}
	if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {
		t.Fatalf("decode V2 append-log response: %v", err)
	}
	if response.Code != "append_log_unsupported" {
		t.Fatalf("V2 append-log response code=%q, want append_log_unsupported", response.Code)
	}
}

func TestAppendLogHTTPValidationAndLimit(t *testing.T) {
	s := newAppendLogServer(t)
	scope := &TenantScope{Backend: s.fallback}
	tests := []struct {
		name    string
		body    []byte
		headers map[string]string
		chunked bool
		want    int
	}{
		{
			name: "chunked body rejected",
			body: []byte("x"),
			headers: map[string]string{
				"X-Dat9-Expected-Revision": "0",
				"X-Dat9-Expected-Size":     "0",
			},
			chunked: true,
			want:    http.StatusBadRequest,
		},
		{
			name: "negative expected size rejected",
			body: []byte("x"),
			headers: map[string]string{
				"X-Dat9-Expected-Revision": "0",
				"X-Dat9-Expected-Size":     "-1",
			},
			want: http.StatusBadRequest,
		},
		{
			name: "resulting size over limit rejected before stat",
			body: []byte("x"),
			headers: map[string]string{
				"X-Dat9-Expected-Revision": "0",
				"X-Dat9-Expected-Size":     "5000000000",
			},
			want: http.StatusRequestEntityTooLarge,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			r := httptest.NewRequest(http.MethodPost, "/v1/fs/validate?append-log", bytes.NewReader(tt.body))
			for name, value := range tt.headers {
				r.Header.Set(name, value)
			}
			if tt.chunked {
				r.ContentLength = -1
			}
			r = r.WithContext(withScope(r.Context(), scope))
			w := httptest.NewRecorder()
			s.handleAppendLog(w, r, "/validate")
			if w.Code != tt.want {
				t.Fatalf("append-log validation status=%d body=%s, want %d", w.Code, w.Body.String(), tt.want)
			}
		})
	}
}

func TestAppendLogHTTPRejectsMismatchedRequestBody(t *testing.T) {
	tests := []struct {
		name          string
		body          io.Reader
		contentLength int64
		maxUpload     int64
		wantStatus    int
		wantCode      string
	}{
		{
			name:          "short body",
			body:          bytes.NewReader([]byte("x")),
			contentLength: 2,
			wantStatus:    http.StatusBadRequest,
			wantCode:      "append_log_invalid_request",
		},
		{
			name:          "unexpected EOF",
			body:          appendLogErrorReader{err: io.ErrUnexpectedEOF},
			contentLength: 1,
			wantStatus:    http.StatusBadRequest,
			wantCode:      "append_log_invalid_request",
		},
		{
			name:          "max bytes exceeded while streaming",
			body:          bytes.NewReader([]byte("xx")),
			contentLength: 1,
			maxUpload:     1,
			wantStatus:    http.StatusRequestEntityTooLarge,
			wantCode:      "append_log_too_large",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			s := newAppendLogServer(t)
			if tt.maxUpload > 0 {
				s.maxUploadBytes = tt.maxUpload
			}
			r := httptest.NewRequest(http.MethodPost, "/v1/fs/wal?append-log", tt.body)
			r.ContentLength = tt.contentLength
			r.Header.Set("X-Dat9-Expected-Revision", "0")
			r.Header.Set("X-Dat9-Expected-Size", "0")
			r = r.WithContext(withScope(r.Context(), &TenantScope{Backend: s.fallback}))
			w := httptest.NewRecorder()
			s.handleAppendLog(w, r, "/wal")
			if w.Code != tt.wantStatus {
				t.Fatalf("status=%d body=%s, want %d", w.Code, w.Body.String(), tt.wantStatus)
			}
			var response struct {
				Code string `json:"code"`
			}
			if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {
				t.Fatalf("decode response: %v", err)
			}
			if response.Code != tt.wantCode {
				t.Fatalf("response code=%q, want %q", response.Code, tt.wantCode)
			}
			if _, err := s.fallback.StatNodeCtx(context.Background(), "/wal"); !errors.Is(err, datastore.ErrNotFound) {
				t.Fatalf("rejected append created /wal: %v", err)
			}
		})
	}
}

func TestAppendLogConditionalPUTRequiresExplicitLength(t *testing.T) {
	ctx := context.Background()
	s := newAppendLogServer(t)
	created, err := s.fallback.AppendLog(ctx, "/wal", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	r := httptest.NewRequest(http.MethodPut, "/v1/fs/wal", bytes.NewReader([]byte("reset")))
	r.ContentLength = -1
	r.Header.Set("X-Dat9-Expected-Revision", strconv.FormatInt(created.Revision, 10))
	r = r.WithContext(withScope(r.Context(), &TenantScope{Backend: s.fallback}))
	w := httptest.NewRecorder()
	s.handleWrite(w, r, "/wal")
	if w.Code != http.StatusBadRequest {
		t.Fatalf("unknown-length append-log PUT status=%d body=%s, want 400", w.Code, w.Body.String())
	}
}

func TestAppendLogConditionalPUTRejectsBodyLengthMismatch(t *testing.T) {
	ctx := context.Background()
	s := newAppendLogServer(t)
	created, err := s.fallback.AppendLog(ctx, "/wal", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	for _, tc := range []struct {
		name     string
		body     string
		declared string
	}{
		{name: "short", body: "short", declared: "6"},
		{name: "long", body: "longer", declared: "5"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			r := httptest.NewRequest(http.MethodPut, "/v1/fs/wal", bytes.NewReader([]byte(tc.body)))
			r.ContentLength = -1
			r.Header.Set("X-Dat9-Expected-Revision", strconv.FormatInt(created.Revision, 10))
			r.Header.Set("X-Dat9-Content-Length", tc.declared)
			r = r.WithContext(withScope(r.Context(), &TenantScope{Backend: s.fallback}))
			w := httptest.NewRecorder()
			s.handleWrite(w, r, "/wal")
			if w.Code != http.StatusBadRequest {
				t.Fatalf("body=%q declared=%s status=%d body=%s, want 400", tc.body, tc.declared, w.Code, w.Body.String())
			}
		})
	}
	nf, err := s.fallback.StatNodeCtx(ctx, "/wal")
	if err != nil {
		t.Fatalf("stat append-log: %v", err)
	}
	if nf.File == nil || nf.File.Revision != created.Revision || nf.File.SizeBytes != created.Size {
		t.Fatalf("metadata after rejected rewrite = %+v, want unchanged", nf.File)
	}
}

func TestAppendLogHTTPRejectsExistingZeroAppendAndOversizeBeforeMutation(t *testing.T) {
	ctx := context.Background()
	s := newAppendLogServer(t)
	created, err := s.fallback.AppendLog(ctx, "/wal", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	scope := &TenantScope{Backend: s.fallback}

	r := httptest.NewRequest(http.MethodPost, "/v1/fs/wal?append-log", bytes.NewReader(nil))
	r.Header.Set("X-Dat9-Expected-Revision", strconv.FormatInt(created.Revision, 10))
	r.Header.Set("X-Dat9-Expected-Size", strconv.FormatInt(created.Size, 10))
	r = r.WithContext(withScope(r.Context(), scope))
	w := httptest.NewRecorder()
	s.handleAppendLog(w, r, "/wal")
	if w.Code != http.StatusBadRequest {
		t.Fatalf("existing zero append status=%d body=%s, want 400", w.Code, w.Body.String())
	}
	nf, err := s.fallback.StatNodeCtx(ctx, "/wal")
	if err != nil {
		t.Fatalf("stat append-log: %v", err)
	}
	if nf.File == nil || nf.File.Revision != created.Revision || nf.File.SizeBytes != created.Size {
		t.Fatalf("metadata after zero append = %+v, want unchanged", nf.File)
	}

	s.maxUploadBytes = 1
	r = httptest.NewRequest(http.MethodPost, "/v1/fs/oversize?append-log", bytes.NewReader([]byte("xx")))
	r.Header.Set("X-Dat9-Expected-Revision", "0")
	r.Header.Set("X-Dat9-Expected-Size", "0")
	r = r.WithContext(withScope(r.Context(), scope))
	w = httptest.NewRecorder()
	s.handleAppendLog(w, r, "/oversize")
	if w.Code != http.StatusRequestEntityTooLarge {
		t.Fatalf("oversize append status=%d body=%s, want 413", w.Code, w.Body.String())
	}
	if _, err := s.fallback.StatNodeCtx(ctx, "/oversize"); !errors.Is(err, datastore.ErrNotFound) {
		t.Fatalf("oversize append created path: %v", err)
	}
}

func TestFSLayerSnapshotRejectsAppendLog(t *testing.T) {
	ctx := context.Background()
	s := newAppendLogServer(t)
	if _, err := s.fallback.AppendLog(ctx, "/wal", bytes.NewReader([]byte("seed")), 4, 0, 0); err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	snapshot := snapshotFSLayerBasePath(ctx, s.fallback, "/wal", false)
	if snapshot.SnapshotOK || !errors.Is(snapshot.SnapshotErr, backend.ErrAppendLogUnsupported) {
		t.Fatalf("fs layer append-log snapshot = %+v, want explicit unsupported", snapshot)
	}
}

func TestBatchWriteRejectsAppendLog(t *testing.T) {
	ctx := context.Background()
	s := newAppendLogServer(t)
	created, err := s.fallback.AppendLog(ctx, "/wal", bytes.NewReader([]byte("seed")), 4, 0, 0)
	if err != nil {
		t.Fatalf("create append-log: %v", err)
	}
	payload, err := json.Marshal(batchWriteRequest{Items: []batchWriteItem{{
		Path: "/wal",
		// Stay below the fixture's one-byte inline threshold so the layout
		// guard, rather than upload-size validation, rejects the overwrite.
		Data:             []byte{},
		ExpectedRevision: created.Revision,
	}}})
	if err != nil {
		t.Fatalf("marshal batch write: %v", err)
	}
	r := httptest.NewRequest(http.MethodPost, "/v1/fs:batch-write", bytes.NewReader(payload))
	r = r.WithContext(withScope(r.Context(), &TenantScope{Backend: s.fallback}))
	w := httptest.NewRecorder()
	s.handleBatchWrite(w, r)
	if w.Code != http.StatusOK {
		t.Fatalf("batch write status=%d body=%s, want 200 with per-item result", w.Code, w.Body.String())
	}
	var response batchWriteResponse
	if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {
		t.Fatalf("decode batch write response: %v", err)
	}
	if len(response.Results) != 1 || response.Results[0].Status != http.StatusBadRequest {
		t.Fatalf("batch write results = %+v, want append-log 400", response.Results)
	}
	if response.Results[0].Error != backend.ErrAppendLogUnsupported.Error() {
		t.Fatalf("batch write error = %q, want append-log rejection", response.Results[0].Error)
	}
	node, err := s.fallback.StatNodeCtx(ctx, "/wal")
	if err != nil {
		t.Fatalf("stat rejected batch-write target: %v", err)
	}
	if node.File == nil || node.File.ContentLayout != datastore.ContentLayoutAppendLog || node.File.Revision != created.Revision || node.File.SizeBytes != created.Size {
		t.Fatalf("rejected batch write changed append-log metadata: %+v", node.File)
	}
}