package console

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

	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/mail"
	"github.com/carlosframework/platform/internal/release"
)

func TestClaimMutationStaleConsoleCannotMutateRecreatedApp(t *testing.T) {
	t.Run("bound", func(t *testing.T) { claimStaleIncarnationCase(t, false) })
	t.Run("unbound", func(t *testing.T) { claimStaleIncarnationCase(t, true) })
}

func claimStaleIncarnationCase(t *testing.T, unbound bool) {
	w, _ := newClaimRestoreWorld(t)
	for _, srv := range []*Server{w.a, w.b} {
		if err := srv.store.AddMember(crAcctID, crOwner, RoleOwner, time.Now()); err != nil {
			t.Fatal(err)
		}
	}
	if err := w.a.restoreApp(w.ctx, w.account(t, w.a), crApp, time.Now()); err != nil {
		t.Fatal(err)
	}
	w.materialize(t, w.b)
	// B retains its queued callbacks while A purges and recreates the name.
	w.b.cfg.AppsDomain = "oncarlos.com"
	rec := &claimRecorder{}
	priorTransport := mailPushClient.Transport
	mailPushClient.Transport = claimOKTransport{rec: rec}
	t.Cleanup(func() { mailPushClient.Transport = priorTransport })
	if err := w.b.store.SetMailEventHook(w.ctx, crSqid, crApp, "/mail/events", crOwner, time.Now()); err != nil {
		t.Fatal(err)
	}
	enqueue := func(notice string) {
		t.Helper()
		if err := w.b.store.EnqueueMailEventDelivery(w.ctx, crSqid, crApp, notice, []byte(`{}`), time.Now().Add(-time.Minute)); err != nil {
			t.Fatal(err)
		}
	}
	enqueue("before-recreation")
	if err := w.b.mailPushPass(w.ctx, time.Now); err != nil {
		t.Fatal(err)
	}
	if events := rec.since(0); len(events) != 1 {
		t.Fatalf("valid worker did not send before recreation: %v", events)
	}
	enqueue("delayed-before-recreation")
	oldHooks, err := w.b.store.activeMailEventHooks(w.ctx)
	if err != nil || len(oldHooks) != 1 {
		t.Fatalf("old hooks: %v %v", oldHooks, err)
	}
	oldHook := oldHooks[0]
	oldHook.claimAppID = crAppID
	oldRows, err := w.b.store.dueMailEventDeliveriesFor(w.ctx, crSqid, crApp, time.Now(), mailPushPerApp)
	if err != nil || len(oldRows) != 1 {
		t.Fatalf("old rows: %v %v", oldRows, err)
	}

	// B goes offline holding A's row and credentials. Stage the durable result
	// of A's completed purge; the independent purge suite drives its chores.
	staleToken := mintUserToken(t, w.b.store, crOwner, time.Now())
	stale := httptest.NewServer(w.b)
	defer stale.Close()
	body := `{"set":{"REV5":{"kind":"env","value":"stale"}}}`
	for _, token := range []string{staleToken} {
		status, _, out := rawReq(t, http.MethodPut, stale.URL+"/api/cli/apps/bab/kass/config", token, body)
		if status != http.StatusOK {
			t.Fatalf("old credential was not a valid mutation fixture: %d %s", status, out)
		}
	}
	keys, err := w.bucket.List(w.ctx, "apps/"+crSqid+"/"+crApp+"/")
	if err != nil {
		t.Fatal(err)
	}
	for _, key := range keys {
		if !strings.HasSuffix(key, "/admission.json") {
			if err := w.bucket.Delete(w.ctx, key); err != nil {
				t.Fatal(err)
			}
		}
	}
	a := w.admission(t).Value
	a.State = "closed"
	a.Op = nil
	a.Registrations = map[string]hostops.Registration{}
	a.Effects = map[string]hostops.Effect{}
	a.LastPurge = &hostops.PurgeMark{ID: "finished-purge", AppID: crAppID}
	seedClaimAdmission(t, w.bucket, crSqid, crApp, a)
	err = w.a.store.withClaimSQL(gateCtx(t), []claimSQLScope{{AccountID: crAcctID, App: crApp, AppID: crAppID, Mode: claimModeEffect}}, allowAllClaimSQL, func(st *Store) error {
		_, err := st.db.Exec(`DELETE FROM apps WHERE account_id=? AND app=?`, crAcctID, crApp)
		return err
	})
	if err != nil {
		t.Fatal(err)
	}
	// The production creation path reserves and binds B's new incarnation.
	created, err := w.a.createClaimAppRow(w.ctx, w.account(t, w.a), crApp, func(st *Store) (bool, error) { return st.CreateAppIn(crAcctID, crApp, crOwner, time.Now()) })
	if err != nil || !created {
		t.Fatalf("recreate: %v %v", created, err)
	}
	current, err := hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, crApp)
	if err != nil {
		t.Fatal(err)
	}
	if current.Value.AppID == crAppID {
		t.Fatal("recreation reused incarnation")
	}
	if local, err := w.b.store.ClaimAppID(crAcctID, crApp); err != nil || local != crAppID {
		t.Fatalf("stale row %q %v", local, err)
	}

	if unbound {
		err = w.b.store.withClaimSQL(gateCtx(t), []claimSQLScope{{AccountID: crAcctID, App: crApp, AppID: crAppID, Mode: claimModeEffect}}, allowAllClaimSQL, func(st *Store) error {
			_, err := st.db.Exec(`UPDATE apps SET app_id='' WHERE account_id=? AND app=?`, crAcctID, crApp)
			return err
		})
		if err != nil {
			t.Fatal(err)
		}
		if err = w.b.bindExistingClaimAppAdmission(w.ctx, w.account(t, w.b), crApp); !errors.Is(err, hostops.ErrPending) {
			t.Fatalf("unbound stale row adopted successor: %v", err)
		}
		if err = w.b.SweepClaimAppIdentities(w.ctx); err != nil {
			t.Fatal(err)
		}
		if id, err := w.b.store.ClaimAppID(crAcctID, crApp); id != "" || !errors.Is(err, hostops.ErrIdentity) {
			t.Fatalf("sweep bound untrusted identity: %q %v", id, err)
		}
	}
	// A queued in-memory turn cannot bind to the same console's new row.
	w.a.cfg.AppsDomain = "oncarlos.com"
	if err := w.a.store.SetMailEventHook(w.ctx, crSqid, crApp, "/mail/events", crOwner, time.Now()); err != nil {
		t.Fatal(err)
	}
	signingKey, err := w.a.store.MailEventsSigningKey()
	if err != nil {
		t.Fatal(err)
	}
	delayedMark := rec.mark()
	w.a.deliverAppMailEvents(w.ctx, oldHook, oldRows, signingKey, time.Now)
	if events := rec.since(delayedMark); len(events) != 0 {
		t.Fatalf("delayed turn sent to successor: %v", events)
	}
	current, err = hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, crApp)
	if err != nil {
		t.Fatal(err)
	}
	mark := rec.mark()
	if err := w.b.mailPushPass(w.ctx, time.Now); err != nil {
		t.Fatal(err)
	}
	if events := rec.since(mark); len(events) != 0 {
		t.Fatalf("stale worker sent after recreation: %v", events)
	}
	pending, err := w.b.store.dueMailEventDeliveriesFor(w.ctx, crSqid, crApp, time.Now(), mailPushPerApp)
	if err != nil || len(pending) != 1 {
		t.Fatalf("stale worker settled old queue: %v %v", pending, err)
	}
	for _, token := range []string{staleToken} {
		status, _, out := rawReq(t, http.MethodPut, stale.URL+"/api/cli/apps/bab/kass/config", token, body)
		if status != http.StatusBadGateway {
			t.Fatalf("stale console mutated successor: %d %s", status, out)
		}
		after, err := hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, crApp)
		if err != nil {
			t.Fatal(err)
		}
		if after.ETag != current.ETag {
			t.Fatal("stale request acquired successor admission")
		}
		keys, err = w.bucket.List(w.ctx, "apps/"+crSqid+"/"+crApp+"/config/")
		if err != nil || len(keys) != 0 {
			t.Fatalf("stale config effect: %v %v", keys, err)
		}
	}
	called := false
	err = release.WithAppMutation(w.ctx, w.b.cfg.Releases, crSqid, crApp, "stale", func(context.Context) error { called = true; return nil })
	if !errors.Is(err, hostops.ErrIdentity) || called {
		t.Fatalf("stale runner: called=%v err=%v", called, err)
	}
	_, err = mail.Declare(w.ctx, w.b.claimMailStore(), crSqid, crApp, "EMAIL", "sender@example.com", crOwner, time.Now())
	if !errors.Is(err, hostops.ErrIdentity) {
		t.Fatalf("stale mail admitted: %v", err)
	}
	after, err := hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, crApp)
	if err != nil || after.ETag != current.ETag {
		t.Fatalf("stale mail changed successor admission: %v", err)
	}
	gate, _, err := w.bucket.Get(w.ctx, hostops.MailGateKey)
	if err != nil {
		t.Fatal(err)
	}
	var held struct {
		State   string            `json:"state"`
		Members []json.RawMessage `json:"members"`
	}
	if err = json.Unmarshal(gate, &held); err != nil || held.State != "free" || len(held.Members) != 0 {
		t.Fatalf("stale mail left gate membership: %s %v", gate, err)
	}
	fresh := httptest.NewServer(w.a)
	defer fresh.Close()
	token := mintUserToken(t, w.a.store, crOwner, time.Now())
	status, _, out := rawReq(t, http.MethodPut, fresh.URL+"/api/cli/apps/bab/kass/config", token, body)
	if status != http.StatusOK {
		t.Fatalf("current console refused: %d %s", status, out)
	}
}

func TestClaimAdmissionSweepContinuesPastStaleLocalIdentity(t *testing.T) {
	w, _ := newClaimRestoreWorld(t)
	a := w.admission(t).Value
	a.AppID = "successor"
	a.LastPurge = &hostops.PurgeMark{ID: "finished", AppID: crAppID}
	seedClaimAdmission(t, w.bucket, crSqid, crApp, a)
	op := hostops.AppOperation{ID: "joined", Kind: hostops.OpAppMutation, AppID: "other-id", Intent: map[string]string{"action": "test"}}
	pending := hostops.AppAdmission{V: 1, AppID: "other-id", Revision: 1, Epoch: 1, State: "closed", Op: &op, Registrations: map[string]hostops.Registration{}, Effects: map[string]hostops.Effect{"joined": {ID: "joined", OperationID: "joined", AppID: "other-id", State: "complete", Resource: hostops.OpAppMutation, Action: "test", Owner: w.b.claimOwner}}}
	seedClaimAdmission(t, w.bucket, crSqid, "other", pending)
	if err := w.b.SweepClaimAdmissions(w.ctx); err != nil {
		t.Fatal(err)
	}
	other, err := hostops.ReadAdmission(w.ctx, w.bucket, crSqid, "other", "other-id")
	if err != nil {
		t.Fatal(err)
	}
	if other.Value.Op != nil || other.Value.State != "open" {
		t.Fatal("stale local row blocked another app's recovery")
	}
	if id, err := w.b.store.ClaimAppID(crAcctID, crApp); err != nil || id != crAppID {
		t.Fatalf("sweep rebound stale identity: %q %v", id, err)
	}
}

func TestClaimAdmissionSweepContinuesPastMalformedIdentity(t *testing.T) {
	w, _ := newClaimRestoreWorld(t)
	a := hostops.AppAdmission{V: 1, AppID: "other-id", Revision: 1, Epoch: 1, State: "closed", Op: &hostops.AppOperation{ID: "joined", Kind: hostops.OpAppMutation, AppID: "other-id", Intent: map[string]string{"action": "test"}}, Registrations: map[string]hostops.Registration{}, Effects: map[string]hostops.Effect{"joined": {ID: "joined", OperationID: "joined", AppID: "other-id", State: "complete", Owner: w.b.claimOwner}}}
	seedClaimAdmission(t, w.bucket, crSqid, "other", a)
	key := "apps/" + crSqid + "/" + crApp + "/admission.json"
	_, tag, err := w.bucket.Get(w.ctx, key)
	if err != nil {
		t.Fatal(err)
	}
	if _, err = w.bucket.Put(w.ctx, key, []byte(`{broken`), tag); err != nil {
		t.Fatal(err)
	}
	if err = w.b.SweepClaimAdmissions(w.ctx); err == nil {
		t.Fatal("malformed identity was not reported")
	}
	other, err := hostops.ReadAdmission(w.ctx, w.bucket, crSqid, "other", "other-id")
	if err != nil {
		t.Fatal(err)
	}
	if other.Value.Op != nil || other.Value.State != "open" {
		t.Fatal("identity error stopped independent recovery")
	}
}

func TestClaimMailPreflightsAllLocalIdentitiesBeforeAnyMember(t *testing.T) {
	w, _ := newClaimRestoreWorld(t)
	const other = "aaa"
	for _, srv := range []*Server{w.a, w.b} {
		if _, err := srv.createClaimAppRow(w.ctx, w.account(t, srv), other, func(st *Store) (bool, error) { return st.CreateAppIn(crAcctID, other, crOwner, time.Now()) }); err != nil {
			t.Fatal(err)
		}
	}
	a := w.admission(t).Value
	a.AppID = "successor"
	a.LastPurge = &hostops.PurgeMark{ID: "finished", AppID: crAppID}
	seedClaimAdmission(t, w.bucket, crSqid, crApp, a)
	before, err := hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, other)
	if err != nil {
		t.Fatal(err)
	}
	_, err = mail.PutDomain(w.ctx, w.b.claimMailStore(), crSqid, &mail.DomainRecord{Schema: 1, Domain: "shared.example.com", Apps: []string{other, crApp}}, "")
	if !errors.Is(err, hostops.ErrIdentity) {
		t.Fatalf("stale shared-domain member: %v", err)
	}
	after, err := hostops.ReadCanonicalAdmission(w.ctx, w.bucket, crSqid, other)
	if err != nil || after.ETag != before.ETag {
		t.Fatalf("earlier member was admitted before complete identity preflight: %v", err)
	}
	gate, _, err := w.bucket.Get(w.ctx, hostops.MailGateKey)
	if err != nil {
		t.Fatal(err)
	}
	var held struct {
		State   string            `json:"state"`
		Members []json.RawMessage `json:"members"`
	}
	if err = json.Unmarshal(gate, &held); err != nil || held.State != "free" || len(held.Members) != 0 {
		t.Fatalf("refusal recorded members: %s %v", gate, err)
	}
}