package sitesd

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"tangled.org/core/sitesd/cloudflare"
	"tangled.org/core/sitesd/config"
	"tangled.org/core/sitesd/models"
	"tangled.org/core/sitesd/reposync"
	"tangled.org/core/sitesd/worker"
)

const (
	testOwnerDid = "did:plc:mockowner"
	testRepoDid  = "did:plc:repo1"
	testKnot     = "knot.invalid"
	waitFor      = 5 * time.Second
)

var errBoom = errors.New("boom")

func TestServeRequiresR2OutsideDev(t *testing.T) {
	for _, tc := range []struct {
		name, account, bucket, access, secret string
	}{
		{name: "account", bucket: "bucket", access: "access", secret: "secret"},
		{name: "bucket", account: "account", access: "access", secret: "secret"},
		{name: "access", account: "account", bucket: "bucket", secret: "secret"},
		{name: "secret", account: "account", bucket: "bucket", access: "access"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			t.Setenv("CLOUDFLARE_ACCOUNT_ID", tc.account)
			t.Setenv("CLOUDFLARE_R2_BUCKET", tc.bucket)
			t.Setenv("CLOUDFLARE_R2_ACCESS_KEY_ID", tc.access)
			t.Setenv("CLOUDFLARE_R2_SECRET_ACCESS_KEY", tc.secret)
			opts := Options{Knots: testKnot, Jetstream: "wss://jetstream.invalid/subscribe", KeyHex: strings.Repeat("0", 63) + "1"}
			err := Serve(context.Background(), slog.New(slog.NewTextHandler(io.Discard, nil)), opts)
			if err == nil || !strings.Contains(err.Error(), "cloudflare R2 requires") {
				t.Fatalf("Serve with missing %s: %v", tc.name, err)
			}
		})
	}
}

type serverHooks struct {
	getCtx func(context.Context, string) (*worker.DeployContext, error)
	deploy func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error
	record func(context.Context, *worker.DeployContext, string, string, string, string) error
}

func newTestServer(hooks serverHooks) *server {
	cf, err := cloudflare.New(&config.Config{Cloudflare: config.Cloudflare{
		AccountId: "acct",
		R2:        config.R2Config{Bucket: "bucket"},
	}})
	if err != nil {
		panic(err)
	}

	return &server{
		cfg:       &config.Config{},
		cf:        cf,
		locks:     reposync.New(),
		logger:    slog.New(slog.NewTextHandler(io.Discard, nil)),
		deployCtx: context.Background(),
		getCtx:    hooks.getCtx,
		deploy:    hooks.deploy,
		record:    hooks.record,
	}
}

func internalDeployRequest(body string) *http.Request {
	return httptest.NewRequest(http.MethodPost, "/deploy", strings.NewReader(body))
}

func ctxFor(overrides ...func(*worker.DeployContext)) *worker.DeployContext {
	dc := &worker.DeployContext{
		RepoDid:  testRepoDid,
		OwnerDid: testOwnerDid,
		Name:     "myblog",
		Rkey:     "abc1",
		Knot:     testKnot,
		Branch:   "main",
		Dir:      "/",
	}
	for _, o := range overrides {
		o(dc)
	}
	return dc
}

func decodeStatus(t *testing.T, w *httptest.ResponseRecorder) string {
	t.Helper()
	var out struct {
		Status string `json:"status"`
	}
	if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil {
		t.Fatalf("decoding body %q: %v", w.Body.String(), err)
	}
	return out.Status
}

func await(t *testing.T, ch <-chan struct{}, what string) {
	t.Helper()
	select {
	case <-ch:
	case <-time.After(waitFor):
		t.Fatalf("timed out waiting for %s", what)
	}
}

func awaitQuiet(t *testing.T, ch <-chan struct{}, d time.Duration, why string) {
	t.Helper()
	select {
	case <-ch:
		t.Fatal(why)
	case <-time.After(d):
	}
}

func requireAccepted(t *testing.T, w *httptest.ResponseRecorder) {
	t.Helper()
	if w.Code != http.StatusAccepted {
		t.Fatalf("status = %d, want 202 (body %q)", w.Code, w.Body.String())
	}
	if got := decodeStatus(t, w); got != "accepted" {
		t.Errorf("status body = %q, want accepted", got)
	}
}

func requireRefused(t *testing.T, srv *server, w *httptest.ResponseRecorder, deployed, recorded <-chan struct{}, want int) {
	t.Helper()
	if w.Code != want {
		t.Fatalf("status = %d, want %d (body %q)", w.Code, want, w.Body.String())
	}
	for what, ch := range map[string]<-chan struct{}{"deploy": deployed, "history write": recorded} {
		select {
		case <-ch:
			t.Errorf("a refused request still did the %s", what)
		default:
		}
	}
	release, ok := srv.locks.TryAcquire()
	if !ok {
		t.Error("a refused request kept a deploy slot")
		return
	}
	release()
}

func fillSlots(t *testing.T, r *reposync.Registry) []func() {
	t.Helper()
	var held []func()
	for len(held) < 64 {
		release, ok := r.TryAcquire()
		if !ok {
			return held
		}
		held = append(held, release)
	}
	t.Fatal("the deploy pool never filled")
	return nil
}

func TestInternalDeploy_RequiresRepoDid(t *testing.T) {
	srv := newTestServer(serverHooks{})
	w := httptest.NewRecorder()
	srv.handleInternalDeploy(w, internalDeployRequest(`{}`))
	if w.Code != http.StatusBadRequest {
		t.Fatalf("status = %d, want 400", w.Code)
	}
}

func TestInternalDeploy_AcceptsBeforeTheDeployRuns(t *testing.T) {
	var repo models.Repo
	var recordCall *worker.DeployContext
	var recordArgs [4]string
	var ctxCalls, recordCalls int
	started := make(chan struct{})
	release := make(chan struct{})
	recorded := make(chan struct{})
	srv := newTestServer(serverHooks{
		getCtx: func(_ context.Context, repoDid string) (*worker.DeployContext, error) {
			ctxCalls++
			if repoDid != testRepoDid {
				t.Errorf("getCtx repoDid = %q, want %q", repoDid, testRepoDid)
			}
			return ctxFor(), nil
		},
		deploy: func(_ context.Context, _ *cloudflare.Client, _ *config.Config, r *models.Repo, branch, dir string) error {
			repo = *r
			if branch != "main" || dir != "/" {
				t.Errorf("deploy branch/dir = %q/%q, want main//", branch, dir)
			}
			close(started)
			<-release
			return nil
		},
		record: func(_ context.Context, dc *worker.DeployContext, sha, status, trigger, errMsg string) error {
			recordCalls++
			recordCall = dc
			recordArgs = [4]string{sha, status, trigger, errMsg}
			close(recorded)
			return nil
		},
	})

	w := httptest.NewRecorder()
	srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	requireAccepted(t, w)
	select {
	case <-recorded:
		t.Fatal("the ack waited for the deploy to finish")
	default:
	}

	await(t, started, "the accepted deploy to run")
	close(release)
	await(t, recorded, "the history write")

	if ctxCalls != 1 {
		t.Errorf("getDeployContext calls = %d, want the ack's one preflight", ctxCalls)
	}
	if recordCalls != 1 {
		t.Errorf("recordDeploy calls = %d, want 1", recordCalls)
	}
	if repo.Did != testOwnerDid || repo.Rkey != "abc1" || repo.Knot != testKnot || repo.RepoDid != testRepoDid {
		t.Errorf("deploy repo = %+v, want the owner's rkey/knot/repoDid", repo)
	}
	if recordCall == nil || recordCall.RepoDid != testRepoDid {
		t.Fatalf("record was not called with the repo: %+v", recordCall)
	}
	if recordArgs[0] != "" || recordArgs[1] != "success" || recordArgs[2] != "config_change" || recordArgs[3] != "" {
		t.Errorf("record args = %v, want empty sha, success, config_change, empty error", recordArgs)
	}
}

func TestInternalDeploy_Trigger(t *testing.T) {
	cases := []struct {
		name string
		body string
		want string
	}{
		{"omitted is config change", `{"repoDid":"did:plc:repo1"}`, "config_change"},
		{"manual", `{"repoDid":"did:plc:repo1","trigger":"manual"}`, "manual"},
		{"push", `{"repoDid":"did:plc:repo1","trigger":"push"}`, "push"},
		{"unknown is refused", `{"repoDid":"did:plc:repo1","trigger":"cron"}`, ""},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			var got string
			deployed := make(chan struct{})
			recorded := make(chan struct{})
			srv := newTestServer(serverHooks{
				getCtx: func(context.Context, string) (*worker.DeployContext, error) {
					return ctxFor(), nil
				},
				deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
					close(deployed)
					return nil
				},
				record: func(_ context.Context, _ *worker.DeployContext, _, _, trigger, _ string) error {
					got = trigger
					close(recorded)
					return nil
				},
			})
			w := httptest.NewRecorder()
			srv.handleInternalDeploy(w, internalDeployRequest(tc.body))
			if tc.want == "" {
				requireRefused(t, srv, w, deployed, recorded, http.StatusBadRequest)
				return
			}
			requireAccepted(t, w)
			await(t, recorded, "the history write")
			if got != tc.want {
				t.Errorf("recorded trigger = %q, want %q", got, tc.want)
			}
		})
	}
}

func TestInternalDeploy_FailureIsRecordedNotReturned(t *testing.T) {
	var recordArgs [4]string
	recorded := make(chan struct{})
	srv := newTestServer(serverHooks{
		getCtx: func(_ context.Context, _ string) (*worker.DeployContext, error) {
			return ctxFor(), nil
		},
		deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
			return errBoom
		},
		record: func(_ context.Context, _ *worker.DeployContext, sha, status, trigger, errMsg string) error {
			recordArgs = [4]string{sha, status, trigger, errMsg}
			close(recorded)
			return nil
		},
	})
	w := httptest.NewRecorder()
	srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	requireAccepted(t, w)
	await(t, recorded, "the history write")
	if recordArgs[1] != "failure" || recordArgs[3] != "boom" {
		t.Errorf("record args = %v, want failure/boom", recordArgs)
	}
}

func TestInternalDeploy_AcksWhileTheDeployIsBlocked(t *testing.T) {
	started := make(chan struct{})
	release := make(chan struct{})
	recorded := make(chan struct{})
	srv := newTestServer(serverHooks{
		getCtx: func(context.Context, string) (*worker.DeployContext, error) {
			return ctxFor(), nil
		},
		deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
			close(started)
			<-release
			return nil
		},
		record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
			close(recorded)
			return nil
		},
	})

	w := httptest.NewRecorder()
	srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	requireAccepted(t, w)
	select {
	case <-recorded:
		t.Fatal("the ack carried the deploy's outcome")
	default:
	}

	await(t, started, "the accepted deploy to run")
	close(release)
	await(t, recorded, "the history write")
}

// negative control: synchronous handler blocks until deploy returns
func TestInternalDeploy_SynchronousShapeStillBlocks(t *testing.T) {
	release := make(chan struct{})
	recorded := make(chan struct{})
	srv := newTestServer(serverHooks{
		getCtx: func(context.Context, string) (*worker.DeployContext, error) {
			return ctxFor(), nil
		},
		deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
			<-release
			return nil
		},
		record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
			close(recorded)
			return nil
		},
	})

	w := httptest.NewRecorder()
	served := make(chan struct{})
	go func() {
		defer close(served)
		synchronousDeployHandler(srv)(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	}()

	awaitQuiet(t, served, 100*time.Millisecond, "the synchronous handler answered while its deploy was blocked")
	close(release)
	await(t, served, "the synchronous handler's answer")
	if w.Code != http.StatusOK {
		t.Fatalf("status = %d, want the pre-fix 200 (body %q)", w.Code, w.Body.String())
	}
}

func synchronousDeployHandler(s *server) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		var req deployRequest
		if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&req); err != nil || req.RepoDid == "" {
			writeErr(w, http.StatusBadRequest, "InvalidRequest", "request body must be JSON with a repoDid")
			return
		}
		trigger := models.SiteDeployTriggerConfigChange
		if req.Trigger != "" {
			var ok bool
			if trigger, ok = models.ParseSiteDeployTrigger(req.Trigger); !ok {
				writeErr(w, http.StatusBadRequest, "InvalidRequest", "unknown trigger")
				return
			}
		}
		done, ok := s.locks.Begin()
		if !ok {
			writeErr(w, http.StatusServiceUnavailable, "ShuttingDown", "deploy executor is shutting down")
			return
		}
		release, err := s.locks.Acquire(r.Context())
		if err != nil {
			done()
			writeErr(w, http.StatusServiceUnavailable, "DeployUnavailable", err.Error())
			return
		}
		dc, err := s.getCtx(r.Context(), req.RepoDid)
		if err != nil || dc == nil {
			done()
			release()
			writeErr(w, http.StatusBadGateway, "DeployFailed", "no deploy context")
			return
		}
		s.runDeploy(acceptedDeploy{dc: dc, trigger: trigger, release: release, done: done})
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		json.NewEncoder(w).Encode(map[string]string{"status": "success"})
	}
}

func TestInternalDeploy_DrainWaitsForAcceptedDeploy(t *testing.T) {
	started := make(chan struct{})
	release := make(chan struct{})
	recorded := make(chan struct{})
	srv := newTestServer(serverHooks{
		getCtx: func(context.Context, string) (*worker.DeployContext, error) {
			return ctxFor(), nil
		},
		deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
			close(started)
			<-release
			return nil
		},
		record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
			close(recorded)
			return nil
		},
	})

	w := httptest.NewRecorder()
	srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	requireAccepted(t, w)
	await(t, started, "the accepted deploy to run")

	drained := make(chan struct{})
	go func() {
		defer close(drained)
		srv.locks.Drain()
	}()
	awaitQuiet(t, drained, 50*time.Millisecond, "Drain returned while an accepted deploy was still running")

	close(release)
	await(t, drained, "Drain after the accepted deploy")
	select {
	case <-recorded:
	default:
		t.Error("Drain returned before the accepted deploy recorded its outcome")
	}
}

func TestInternalDeploy_DrainRefusesNewWork(t *testing.T) {
	deployed := make(chan struct{})
	recorded := make(chan struct{})
	srv := newTestServer(serverHooks{
		getCtx: func(context.Context, string) (*worker.DeployContext, error) {
			return ctxFor(), nil
		},
		deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
			close(deployed)
			return nil
		},
		record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
			close(recorded)
			return nil
		},
	})
	srv.locks.Drain()

	w := httptest.NewRecorder()
	srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	requireRefused(t, srv, w, deployed, recorded, http.StatusServiceUnavailable)
}

func TestInternalDeploy_DrainDuringPreflightRefuses(t *testing.T) {
	preflight := make(chan struct{})
	finishPreflight := make(chan struct{})
	deployed := make(chan struct{})
	recorded := make(chan struct{})
	srv := newTestServer(serverHooks{
		getCtx: func(context.Context, string) (*worker.DeployContext, error) {
			close(preflight)
			<-finishPreflight
			return ctxFor(), nil
		},
		deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
			close(deployed)
			return nil
		},
		record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
			close(recorded)
			return nil
		},
	})

	w := httptest.NewRecorder()
	served := make(chan struct{})
	go func() {
		defer close(served)
		srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	}()
	await(t, preflight, "the preflight")

	srv.locks.Drain()
	close(finishPreflight)
	await(t, served, "the refused answer")

	requireRefused(t, srv, w, deployed, recorded, http.StatusServiceUnavailable)
}

func TestInternalDeploy_NoDeployContext(t *testing.T) {
	cases := []struct {
		name string
		ctx  func() (*worker.DeployContext, error)
		want int
	}{
		{"no site config or active claim", func() (*worker.DeployContext, error) { return nil, nil }, http.StatusUnprocessableEntity},
		{"the context lookup failed", func() (*worker.DeployContext, error) { return nil, errBoom }, http.StatusBadGateway},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			deployed := make(chan struct{})
			recorded := make(chan struct{})
			srv := newTestServer(serverHooks{
				getCtx: func(context.Context, string) (*worker.DeployContext, error) { return tc.ctx() },
				deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
					close(deployed)
					return nil
				},
				record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
					close(recorded)
					return nil
				},
			})
			w := httptest.NewRecorder()
			srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
			requireRefused(t, srv, w, deployed, recorded, tc.want)
		})
	}
}

func TestInternalDeploy_RefusesWhenSlotsFull(t *testing.T) {
	deployed := make(chan struct{})
	recorded := make(chan struct{})
	srv := newTestServer(serverHooks{
		getCtx: func(context.Context, string) (*worker.DeployContext, error) {
			return ctxFor(), nil
		},
		deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
			close(deployed)
			return nil
		},
		record: func(context.Context, *worker.DeployContext, string, string, string, string) error {
			close(recorded)
			return nil
		},
	})
	held := fillSlots(t, srv.locks)

	w := httptest.NewRecorder()
	srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	if w.Code != http.StatusTooManyRequests {
		t.Fatalf("status = %d, want 429 (body %q)", w.Code, w.Body.String())
	}
	select {
	case <-deployed:
		t.Error("a deploy ran without a slot")
	default:
	}
	select {
	case <-recorded:
		t.Error("a deploy without a slot still wrote history")
	default:
	}

	held[0]()
	w = httptest.NewRecorder()
	srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	requireAccepted(t, w)
	await(t, recorded, "the history write")
}

func TestInternalDeploy_CloudflareNotConfigured(t *testing.T) {
	var ctxCalls int
	srv := newTestServer(serverHooks{
		getCtx: func(context.Context, string) (*worker.DeployContext, error) {
			ctxCalls++
			return ctxFor(), nil
		},
	})
	cf, err := cloudflare.New(&config.Config{})
	if err != nil {
		t.Fatal(err)
	}
	if cf.Enabled() {
		t.Fatal("an empty cloudflare config must not read as enabled")
	}
	srv.cf = cf

	w := httptest.NewRecorder()
	srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 (body %q)", w.Code, w.Body.String())
	}
	if ctxCalls != 0 {
		t.Errorf("getDeployContext calls = %d, want none before the cloudflare check", ctxCalls)
	}
}

func TestInternalDeploy_RecordsAnAbandonedRepoLock(t *testing.T) {
	var recordArgs [4]string
	deployed := make(chan struct{})
	recorded := make(chan struct{})
	srv := newTestServer(serverHooks{
		getCtx: func(context.Context, string) (*worker.DeployContext, error) {
			return ctxFor(), nil
		},
		deploy: func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error {
			close(deployed)
			return nil
		},
		record: func(_ context.Context, _ *worker.DeployContext, sha, status, trigger, errMsg string) error {
			recordArgs = [4]string{sha, status, trigger, errMsg}
			close(recorded)
			return nil
		},
	})

	unlock, err := srv.locks.Lock(context.Background(), testRepoDid)
	if err != nil {
		t.Fatal(err)
	}
	defer unlock()
	deployCtx, cancel := context.WithCancel(context.Background())
	cancel()
	srv.deployCtx = deployCtx

	w := httptest.NewRecorder()
	srv.handleInternalDeploy(w, internalDeployRequest(`{"repoDid":"did:plc:repo1"}`))
	requireAccepted(t, w)
	await(t, recorded, "the abandoned deploy's history write")

	if recordArgs[1] != "failure" || !strings.Contains(recordArgs[3], "repo lock") {
		t.Errorf("record args = %v, want a failure naming the abandoned repo lock", recordArgs)
	}
	select {
	case <-deployed:
		t.Error("the deploy ran without the repo lock")
	default:
	}
}