//go:build failpoint

package server

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

	"github.com/c4pt0r/agfs/agfs-server/pkg/filesystem"

	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/metrics"
)

func TestAppendLogConditionalPUTTargetDisappears(t *testing.T) {
	writeCount := func(t *testing.T, result string) int64 {
		t.Helper()
		w := httptest.NewRecorder()
		metrics.WritePrometheus(w)
		prefix := `drive9_business_events_total{event="fs_write",result="` + result + `"} `
		for _, line := range strings.Split(w.Body.String(), "\n") {
			if value, ok := strings.CutPrefix(line, prefix); ok {
				count, err := strconv.ParseInt(value, 10, 64)
				if err != nil {
					t.Fatalf("parse write metric %q: %v", line, err)
				}
				return count
			}
		}
		return 0
	}
	for _, action := range []string{"delete", "rename"} {
		t.Run(action, func(t *testing.T) {
			s := newAppendLogServer(t)
			ctx := context.Background()
			const path = "/vanishing-wal"
			created, err := s.fallback.AppendLog(ctx, path, bytes.NewReader([]byte("seed")), 4, 0, 0)
			if err != nil {
				t.Fatalf("create append-log: %v", err)
			}
			hits := 0
			enableCallFailpoint(t, "github.com/mem9-ai/drive9/pkg/server/beforeAppendLogPUTRewrite", func(server *Server, gotPath string) {
				if server != s || gotPath != path {
					return
				}
				hits++
				if action == "delete" {
					err = s.fallback.RemoveCtx(ctx, path)
				} else {
					err = s.fallback.RenameCtx(ctx, path, "/moved-wal")
				}
				if err != nil {
					t.Fatalf("%s between stats: %v", action, err)
				}
			})
			beforeConflict, beforeError := writeCount(t, "conflict"), writeCount(t, "error")
			body := bytes.NewReader([]byte("replacement"))
			r := httptest.NewRequest(http.MethodPut, "/v1/fs"+path, body)
			r.Header.Set("X-Dat9-Expected-Revision", strconv.FormatInt(created.Revision, 10))
			r = r.WithContext(withMetrics(withScope(r.Context(), &TenantScope{Backend: s.fallback}), s.metrics))
			w := httptest.NewRecorder()
			s.handleWrite(w, r, path)
			if hits != 1 {
				t.Fatalf("race hook calls = %d, want 1", hits)
			}
			if w.Code != http.StatusConflict {
				t.Fatalf("status=%d body=%s, want 409", w.Code, w.Body.String())
			}
			var response struct {
				Code string `json:"code"`
			}
			if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil || response.Code != "append_log_conflict" {
				t.Fatalf("response=%s error=%v, want append_log_conflict", w.Body.String(), err)
			}
			if got := writeCount(t, "conflict") - beforeConflict; got != 1 {
				t.Errorf("conflict metric delta=%d, want 1", got)
			}
			if got := writeCount(t, "error") - beforeError; got != 0 {
				t.Errorf("error metric delta=%d, want 0", got)
			}
			if body.Len() != len("replacement") {
				t.Error("rejected rewrite consumed request body")
			}
			if _, err := s.fallback.StatNodeCtx(ctx, path); !errors.Is(err, datastore.ErrNotFound) {
				t.Fatalf("original path error=%v, want ErrNotFound", err)
			}
			if action == "rename" {
				moved, err := s.fallback.StatNodeCtx(ctx, "/moved-wal")
				if err != nil {
					t.Fatalf("stat renamed target: %v", err)
				}
				if moved.File == nil || moved.File.Revision != created.Revision || moved.File.SizeBytes != created.Size {
					t.Fatalf("renamed target metadata=%+v, want unchanged", moved.File)
				}
			}
		})
	}
}

func TestAppendLogGenericMutationLayoutRace(t *testing.T) {
	for _, operation := range []string{"put", "multipart_put", "patch", "append_plan"} {
		t.Run(operation, func(t *testing.T) {
			s := newAppendLogServer(t)
			ctx := context.Background()
			const path = "/racing-file"
			if _, err := s.fallback.WriteCtx(ctx, path, []byte("normal"), 0, filesystem.WriteFlagCreate); err != nil {
				t.Fatal(err)
			}
			hits := 0
			enableCallFailpoint(t, "github.com/mem9-ai/drive9/pkg/server/afterGenericWriteLayoutStat", func(server *Server, gotPath string) {
				if server != s || gotPath != path {
					return
				}
				hits++
				if err := s.fallback.RemoveCtx(ctx, path); err != nil {
					t.Fatal(err)
				}
				if _, err := s.fallback.AppendLog(ctx, path, bytes.NewReader([]byte("seed")), 4, 0, 0); err != nil {
					t.Fatal(err)
				}
			})
			method, body := http.MethodPut, "x"
			handler := s.handleWrite
			switch operation {
			case "multipart_put":
				body = "large"
			case "patch":
				method, body, handler = http.MethodPatch, `{"new_size":8,"dirty_parts":[1]}`, s.handlePatch
			case "append_plan":
				method, body, handler = http.MethodPost, `{"append_size":4}`, s.handleAppend
			}
			r := httptest.NewRequest(method, "/v1/fs"+path, strings.NewReader(body))
			r.Header.Set("X-Dat9-Part-Checksums", "AAAAAA==")
			r = r.WithContext(withScope(r.Context(), &TenantScope{Backend: s.fallback}))
			w := httptest.NewRecorder()
			handler(w, r, path)
			var response struct {
				Code string `json:"code"`
			}
			if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil || w.Code != http.StatusBadRequest || response.Code != "append_log_unsupported" || hits != 1 {
				t.Fatalf("hook calls=%d status=%d body=%s error=%v", hits, w.Code, w.Body.String(), err)
			}
			node, err := s.fallback.StatNodeCtx(ctx, path)
			if err != nil || node.File.ContentLayout != datastore.ContentLayoutAppendLog || node.File.Revision != 1 || node.File.SizeBytes != 4 {
				t.Fatalf("replacement append-log changed: node=%+v error=%v", node, err)
			}
		})
	}
}