package console

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"net/http"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	lsfile "github.com/benbjohnson/litestream/file"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/svcrole"
)

func enableDoorPlan(t *testing.T, fx *dbFx) *leasetest.VersionedBucket {
	t.Helper()
	rec, err := provision.GetRecord(t.Context(), fx.rel, "bab", "titogo", dbTestHost)
	if err != nil {
		t.Fatal(err)
	}
	inc, err := hostclaim.Incarnation("bab", "titogo", dbTestHost, rec.CreatedAt)
	if err != nil {
		t.Fatal(err)
	}
	c := hostclaim.Claim{V: 2, Host: dbTestHost, Gen: 1, Holder: &hostclaim.Holder{Family: "instance", Key: provision.RecordKey("bab", "titogo", dbTestHost), Account: "bab", App: "titogo", CreatedAt: rec.CreatedAt, Incarnation: inc, Location: "instances/" + dbTestHost}, Op: hostclaim.Op{Kind: hostclaim.KindServing}}
	raw, _ := json.Marshal(c)
	key, _ := hostclaim.Key(dbTestHost)
	if _, err := fx.rel.Put(t.Context(), key, raw, ""); err != nil {
		t.Fatal(err)
	}
	m := leasetest.NewVersionedBucket()
	m.AddBucket("data-test")
	fx.srv.cfg.HomeRegion = "ie"
	fx.srv.cfg.HostClaim = true
	fx.srv.cfg.DBVersions = m
	secret, err := fx.st.SigningKey()
	if err != nil {
		t.Fatal(err)
	}
	fx.srv.dbHandleKey = dbHandleSecret(secret)
	return m
}

func TestDoorPlanHTTPCompatibilityAndRestore(t *testing.T) {
	for _, mode := range []string{"legacy", "origin", "published", "unfinished"} {
		t.Run(mode, func(t *testing.T) {
			enabled := mode != "legacy"
			fx := dbFixture(t)
			fx.replicate(t, dbTestHost, 3)
			if enabled {
				model := enableDoorPlan(t, fx)
				location := "instances/" + dbTestHost
				if mode == "published" || mode == "unfinished" {
					obs, err := hostclaim.Read(t.Context(), fx.rel, dbTestHost)
					if err != nil {
						t.Fatal(err)
					}
					c := cloneDoorClaim(obs.Claim)
					c.Holder.Generation = "live"
					marker := genselect.Marker{Op: "live", Host: dbTestHost, Incarnation: c.Holder.Incarnation, Box: "box", InstalledAt: time.Unix(1, 0).UTC()}
					raw, err := json.Marshal(marker)
					if err != nil {
						t.Fatal(err)
					}
					model.Put("data-test", location+"/generation.json", raw)
					if mode == "unfinished" {
						c.Holder.Generation = "next"
						c.Holder.Admissions = []hostclaim.AdmissionRecord{{Generation: "next", Predecessor: "live", Region: "ie", PredecessorRegion: "ie", Mode: "cold"}}
					}
					writeDoorClaim(t, fx.srv, dbRequest{host: dbTestHost}, c)
					location += "/g-live"
				}
				root := filepath.Join(fx.root, "instances", dbTestHost, "ltx")
				err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
					if err != nil {
						return err
					}
					if d.IsDir() {
						return nil
					}
					rel, err := filepath.Rel(root, p)
					if err != nil {
						return err
					}
					parts := strings.Split(filepath.ToSlash(rel), "/")
					if len(parts) != 2 {
						return fmt.Errorf("unexpected file: %s", rel)
					}
					level, err := strconv.Atoi(parts[0])
					if err != nil {
						return err
					}
					b, err := os.ReadFile(p)
					if err != nil {
						return err
					}
					model.Put("data-test", dbLevelPrefix(location, level)+parts[1], b)
					return nil
				})
				if err != nil {
					t.Fatal(err)
				}
			}
			do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
				r, err := http.NewRequestWithContext(ctx, "GET", fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil)
				if err != nil {
					return nil, err
				}
				r.Header = h.Clone()
				r.Header.Set("Authorization", "Bearer "+fx.member)
				return http.DefaultClient.Do(r)
			}
			got, err := dbdoor.RestoreTo(t.Context(), dbdoor.NewClient(do), t.TempDir())
			if err != nil {
				t.Fatal(err)
			}
			defer os.RemoveAll(got.Dir)
			db, err := sql.Open("sqlite", "file:"+got.Path+"?mode=ro")
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			var rows int
			if err := db.QueryRow("SELECT count(*) FROM t").Scan(&rows); err != nil {
				t.Fatal(err)
			}
			if rows != 3*dbdoortest.Rows {
				t.Fatal(rows)
			}
			// The old CLI ignores additive fields and sends no handle.
			res, err := do(t.Context(), dbdoor.ListTail(0), http.Header{})
			if err != nil {
				t.Fatal(err)
			}
			var list dbdoor.ListResponse
			if err := json.NewDecoder(res.Body).Decode(&list); err != nil {
				t.Fatal(err)
			}
			res.Body.Close()
			if len(list.Files) == 0 {
				t.Fatal("empty L0")
			}
			f := list.Files[0]
			res, err = do(t.Context(), dbdoor.GetTail(f.Level, f.MinTXID, f.MaxTXID), http.Header{})
			if err != nil {
				t.Fatal(err)
			}
			io.Copy(io.Discard, res.Body)
			res.Body.Close()
			want := http.StatusOK
			if enabled {
				want = http.StatusBadRequest
			}
			if res.StatusCode != want {
				t.Fatalf("old CLI: %d want %d", res.StatusCode, want)
			}
			if !enabled && list.Protocol != 0 {
				t.Fatal("flag-off wire changed")
			}
			if enabled && fx.opens.Load() != 0 {
				t.Fatal("pinned door opened unversioned replica")
			}
		})
	}
}

func TestDoorPlanHTTPAuthorizationAndMissingClaim(t *testing.T) {
	fx := dbFixture(t)
	model := enableDoorPlan(t, fx)
	model.FailOn = func(op, key string) error { t.Fatalf("unauthorized data access %s %s", op, key); return nil }
	stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "door-plan-stranger")
	status, h, body := rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), stranger, "")
	assertNotFound(t, "stranger", canonicalNotFound(t), status, h, body)
	admin := serviceToken(t, fx.st, OpsAccountID, "door-plan-admin", "titogo", svcrole.Admin)
	status, _, _ = rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), admin, "")
	if status != http.StatusForbidden {
		t.Fatal(status)
	}
	// A second existing record with no claim must not auto-adopt here.
	other := "other.go.tito.io"
	putDBRecord(t, fx.consoleFixture, provision.Record{Host: other})
	status, h, body = rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", other, 0)), fx.member, "")
	if status != 409 || h.Get("Retry-After") != "1" || !strings.Contains(body, "host not adopted") {
		t.Fatal(status, h, body)
	}
}

func TestDoorPlanHTTPTruncationResumeAndFlagOff(t *testing.T) {
	fx := dbFixture(t)
	model := enableDoorPlan(t, fx)
	key := dbVersionKey("instances/"+dbTestHost, 0, 1, 1)
	payload := append(doorLTXHeader(t, 1, 1), []byte("original")...)
	model.Put("data-test", key, payload)
	var opens atomic.Int32
	fx.srv.cfg.DBVersions = doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, id string) (io.ReadCloser, error) {
		rc, err := model.OpenVersion(ctx, b, k, id)
		if err != nil {
			return nil, err
		}
		if opens.Add(1) == 2 {
			return &dbRangeReader{Reader: io.LimitReader(rc, 3), Closer: rc}, nil
		}
		return rc, nil
	}}
	do := func(tail string, h http.Header) *http.Response {
		t.Helper()
		r, err := http.NewRequest("GET", fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil)
		if err != nil {
			t.Fatal(err)
		}
		r.Header = h.Clone()
		r.Header.Set("Authorization", "Bearer "+fx.member)
		res, err := http.DefaultClient.Do(r)
		if err != nil {
			t.Fatal(err)
		}
		return res
	}
	res := do(dbdoor.ListTail(0), http.Header{})
	var list dbdoor.ListResponse
	if err := json.NewDecoder(res.Body).Decode(&list); err != nil {
		t.Fatal(err)
	}
	res.Body.Close()
	if len(list.Files) != 1 {
		t.Fatal(list)
	}
	f := list.Files[0]
	h := http.Header{}
	h.Set(dbdoor.HandleHeader, f.Handle)
	res = do(dbdoor.GetTail(0, 1, 1), h)
	first, err := io.ReadAll(res.Body)
	res.Body.Close()
	if !errors.Is(err, io.ErrUnexpectedEOF) || string(first) != string(payload[:3]) || res.Header.Get(dbdoor.HandleHeader) != f.Handle {
		t.Fatal(string(first), err, res.Header)
	}
	model.Put("data-test", key, []byte("replacement"))
	h.Set("Range", "bytes=3-")
	res = do(dbdoor.GetTail(0, 1, 1), h)
	rest, err := io.ReadAll(res.Body)
	res.Body.Close()
	if err != nil || res.StatusCode != 206 || string(append(first, rest...)) != string(payload) {
		t.Fatal(string(rest), err, res.StatusCode)
	}
	events, err := fx.st.Events(OpsAccountID, "titogo", 10)
	if err != nil {
		t.Fatal(err)
	}
	ranged := false
	for _, e := range events {
		if e.Action == dbAuditAction && e.Target == dbTestHost && strings.Contains(e.Detail, "byte 3") {
			ranged = true
		}
	}
	if !ranged {
		t.Fatal("missing pinned range audit", events)
	}
	// A flag-off console ignores the handle header exactly as today (R1):
	// it serves the coordinate and never echoes the handle, so a v2 client
	// rejects the response (Task 1's acknowledgement check).
	// Keep startup configuration immutable while either server is running.
	legacy := dbFixture(t)
	legacy.replicate(t, dbTestHost, 1)
	lc := lsfile.NewReplicaClient(filepath.Join(legacy.root, "instances", dbTestHost))
	lf := listFiles(t, lc, 0)[0]
	req, err := http.NewRequest("GET", legacy.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(lf.MinTXID), uint64(lf.MaxTXID))), nil)
	if err != nil {
		t.Fatal(err)
	}
	req.Header = h.Clone()
	req.Header.Del("Range") // a whole-file GET, so success is 200, not 206
	req.Header.Set("Authorization", "Bearer "+legacy.member)
	res, err = http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	io.Copy(io.Discard, res.Body)
	res.Body.Close()
	if res.StatusCode != 200 {
		t.Fatal("flag-off changed its answer to a request carrying a handle header", res.StatusCode)
	}
	if res.Header.Get(dbdoor.HandleHeader) != "" {
		t.Fatal("flag-off console echoed a handle it cannot enforce")
	}
}