package host

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostevidence"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/release"
)

// receiptWorld is one box's stop path over a signed roster: the production
// roster identity, the box's evidence key bound as its census key by the
// real BindCensusKey writer, a stored claim, the real hostguard fence and
// ClaimStopCoordinator, and receipts published into the deployment store
// the unchanged v0.28.4 console verifiers read. Only the kernel's
// observation and the producer drain are fixtures: they are not the path
// under test. The receipt's numeral and binding are.
type receiptWorld struct {
	t       *testing.T
	roster  *bindingRosterFixture
	key     ed25519.PrivateKey
	host    string
	serving bool
	data    string
	locks   string
	kernel  *receiptKernel
	ids     int
	// installed is the actor whose binding stamps the installation.
	installed hostbox.Actor
}

type receiptKernel struct {
	fail      error
	recovered []hostguard.FenceTarget
}

func receiptKernelEvidence() StopKernelEvidence {
	return StopKernelEvidence{ObservedBootID: "boot-1", InventorySHA256: strings.Repeat("b", 64), LocalJournalSHA256: strings.Repeat("c", 64), CgroupIdentities: []string{"cg-1"}, UnitInvocationIDs: []string{"unit-1"}}
}

func (k *receiptKernel) VerifyStopped(context.Context, hostguard.FenceTarget, StopInventory, DrainAcknowledgement) (StopKernelEvidence, error) {
	if k.fail != nil {
		return StopKernelEvidence{}, k.fail
	}
	return receiptKernelEvidence(), nil
}

func (k *receiptKernel) RecoverStopped(_ context.Context, t hostguard.FenceTarget, _ StopInventory) (StopKernelEvidence, DrainAcknowledgement, error) {
	k.recovered = append(k.recovered, t)
	return receiptKernelEvidence(), DrainClaimAcknowledgement(t.FenceID, t.ControllerID, "start-dead", nil), nil
}

func (k *receiptKernel) VerifyNoProcess(context.Context, hostguard.FenceTarget, StopInventory) (StopKernelEvidence, error) {
	e := receiptKernelEvidence()
	e.UnitInvocationIDs = []string{}
	return e, nil
}

const receiptWorldHost = "shop.bab.oncarlos.com"

// newReceiptWorld creates the box, binds its census key, creates the claim
// and holds it at writer session 5: an operation's stop (op purge-1) or, with
// serving, an ordinary serving writer's.
func newReceiptWorld(t *testing.T, serving bool) *receiptWorld {
	t.Helper()
	ctx := t.Context()
	w := &receiptWorld{t: t, roster: newBindingRosterFixture(t, "box-1", "ie"), key: ed25519.NewKeyFromSeed(bytes.Repeat([]byte{7}, ed25519.SeedSize)), host: receiptWorldHost, serving: serving, kernel: &receiptKernel{}}
	w.roster.bindCensusKey(w.key)
	var err error
	if w.installed, err = w.roster.identity().Verify(ctx); err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC().Truncate(time.Second)
	in := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/jam/instances/" + w.host + ".json", Account: "bab", App: "jam", Host: w.host, Region: "ie", RecordRegion: "ie", CreatedAt: now}
	body := []byte(`{"schema":1,"host":"` + w.host + `","account":"bab","app":"jam","region":"ie","created_at":"` + now.Format(time.RFC3339) + `"}`)
	absent, err := hostclaim.Read(ctx, w.roster.store, w.host)
	if err != nil {
		t.Fatal(err)
	}
	o, err := hostclaim.Update(ctx, w.roster.store, absent, hostclaim.BeginCreate(in, body, "create", "console", now))
	if err != nil {
		t.Fatal(err)
	}
	if o, err = hostclaim.Update(ctx, w.roster.store, o, hostclaim.CompleteCreate("create", "box-1")); err != nil {
		t.Fatal(err)
	}
	if _, err = hostclaim.Update(ctx, w.roster.store, o, func(c *hostclaim.Claim) error {
		h := c.Holder
		h.Session, h.WriterBox, h.Generation = 5, "box-1", "g5"
		h.Admissions = []hostclaim.AdmissionRecord{{Region: "ie", Session: 5, Box: "box-1", Generation: "g5", Mode: "fresh"}}
		if !serving {
			c.Op = hostclaim.Op{Kind: hostclaim.KindMoving, ID: "purge-1", Phase: "stopping", Exec: "box-1"}
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	w.data, w.locks = t.TempDir(), t.TempDir()
	if err := os.Mkdir(filepath.Join(w.data, "hostguard"), 0770); err != nil {
		t.Fatal(err)
	}
	if err := w.guard().Activate(ctx, w.host, "box-1", nil); err != nil {
		t.Fatal(err)
	}
	return w
}

func (w *receiptWorld) claim() hostclaim.Observed {
	w.t.Helper()
	o, err := hostclaim.Read(w.t.Context(), w.roster.store, w.host)
	if err != nil {
		w.t.Fatal(err)
	}
	return o
}

func (w *receiptWorld) guard() hostguard.Guard {
	return hostguard.Guard{DataDir: w.data, LockDir: w.locks, Claims: hostguard.StoreReader{Store: w.roster.store}}
}

// coordinator is root's stop coordinator as one process builds it; a new
// call is a restarted root over the same persisted fence records.
func (w *receiptWorld) coordinator() ClaimStopCoordinator {
	inventory := func(_ context.Context, host string) (StopInventory, error) {
		h := w.claim().Claim.Holder
		return StopInventory{Host: host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, Box: "box-1", Enrollment: w.installed.Enrollment, DirectoryRevision: w.installed.Binding, Complete: true, Entries: []StopInventoryEntry{{Kind: "path", ArtifactID: "artifact-1", ManagedRoot: "data", RelativePath: "instances/" + host}}}, nil
	}
	return ClaimStopCoordinator{VerifiedRosterRevision: (&claimCleanupEnrollment{actor: w.installed, directory: w.roster.dir}).rosterRevision, EvidenceSigner: w.key, RequestLimit: hostbox.RequestLimit, Guard: w.guard(), Claims: hostguard.StoreReader{Store: w.roster.store}, CarlosUID: 1001, Evidence: w.roster.store,
		Peers: stopPeerFunc(func(context.Context) (StopPeer, error) {
			return StopPeer{UID: 1001, PID: 42, ProcessStart: "start-42", BootID: "boot-1", ControllerID: "controller-1"}, nil
		}),
		Inventory: exactEvidenceInventory{stopInventoryFunc(inventory)},
		Drainer: stopDrainFunc(func(_ context.Context, _ StopRequest, fence string) (DrainAcknowledgement, error) {
			return DrainClaimAcknowledgement(fence, "controller-1", "start-42", nil), nil
		}),
		Kernel: w.kernel, Now: time.Now,
		NewID: func() string { w.ids++; return fmt.Sprintf("id-%d", w.ids) }}
}

// request is the agent's stop request as its producers build it from a
// fresh actor: the installation's binding and the roster counter it read.
func (w *receiptWorld) request(actor hostbox.Actor) StopRequest {
	w.t.Helper()
	o := w.claim()
	h := o.Claim.Holder
	r := StopRequest{Region: actor.Region, Host: w.host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session, Box: actor.Box, Enrollment: actor.Enrollment,
		BootID: "boot-1", ControllerID: "controller-1", ExpectedClaimETag: o.ETag, ExpectedDirectoryRevision: w.installed.Binding, RosterRevision: actor.Revision}
	if !w.serving {
		r.OperationID, r.Phase, r.ObligationID = "purge-1", "stopping", "pre-teardown-stop"
	}
	return r
}

func (w *receiptWorld) fresh() hostbox.Actor {
	w.t.Helper()
	a, err := w.roster.identity().Verify(w.t.Context())
	if err != nil {
		w.t.Fatal(err)
	}
	return a
}

func decodeStopReceipt(t *testing.T, raw []byte) StopReceipt {
	t.Helper()
	var r StopReceipt
	if err := json.Unmarshal(raw, &r); err != nil {
		t.Fatal(err)
	}
	return r
}

// purgeVerify runs the unchanged console's StopBeforeTeardown twice, as the
// purge's first phase does: the first collects and verifies the receipt and
// cites it in the claim (ErrPending), the second verifies the citation.
func (w *receiptWorld) purgeVerify() error {
	authority := hostops.ProductionAuthority{Store: w.roster.store, Directory: w.roster.dir, Now: time.Now}
	err := authority.StopBeforeTeardown(w.t.Context(), w.claim())
	if !errors.Is(err, hostops.ErrPending) {
		return err
	}
	return authority.StopBeforeTeardown(w.t.Context(), w.claim())
}

// wakeResetVerify is the unchanged console's WakeResetTransport.VerifyReceipt
// for this serving session's stop, named at the claim the stop was asked at.
func (w *receiptWorld) wakeResetVerify(stopped hostclaim.Observed) error {
	h := stopped.Claim.Holder
	now := time.Now().UTC()
	r := hostops.WakeResetRequest{Admission: hostops.WakeResetAdmission{Account: "bab", App: "jam", AppID: "app-1", Registration: "registration-1", Epoch: 1, Owner: hostops.RegistrationOwner{Console: "console", Enrollment: strings.Repeat("d", 32), Controller: "console-controller"}},
		V: 1, ID: strings.Repeat("f", 64), Bucket: bindingFixtureBucket, Console: "console", ConsoleEnrollment: strings.Repeat("d", 32), Namespace: "platform", Host: w.host,
		ClaimETag: stopped.ETag, ClaimGen: stopped.Claim.Gen, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Episode: "episode-1",
		Attempt: hostclaim.WakeRetryAttempt{ID: "attempt-1", Box: "box-1", Enrollment: w.installed.Enrollment, BootID: "boot-1", ControllerID: "controller-1"}, IssuedAt: now, Expires: now.Add(time.Minute)}
	return hostops.WakeResetTransport{Store: w.roster.store, Directory: w.roster.dir, Now: time.Now}.VerifyReceipt(w.t.Context(), r)
}

// L7a (roster binding task 4): the normal and no-process receipt
// constructors take directory_revision from the stop target's roster
// numeral, never the binding, and carry the installation's binding in
// binding. The numeral is the counter of the roster the requester read; a
// retry after roster writes resumes the persisted fence and returns the
// original receipt with its original numeral.
func TestStopReceiptCarriesNumeralAndBinding(t *testing.T) {
	ctx := t.Context()
	for _, serving := range []bool{false, true} {
		t.Run(fmt.Sprintf("serving=%v", serving), func(t *testing.T) {
			w := newReceiptWorld(t, serving)
			actor := w.fresh()
			if !hostactors.ValidBinding(actor.Binding) || actor.Binding != w.installed.Binding {
				t.Fatalf("actor binding %q", actor.Binding)
			}
			req := w.request(actor)
			result, err := w.coordinator().FenceAndStop(ctx, req)
			if err != nil {
				t.Fatal(err)
			}
			r := decodeStopReceipt(t, result.Bytes)
			if r.DirectoryRevision != strconv.FormatUint(actor.Revision, 10) || r.Binding != actor.Binding || result.Target.RosterRevision != actor.Revision || result.Target.DirectoryRevision != actor.Binding {
				t.Fatalf("receipt numeral %q binding %q target %+v; want numeral %d binding %s", r.DirectoryRevision, r.Binding, result.Target, actor.Revision, actor.Binding)
			}
			if _, err := validateTransportStop(req, result); err != nil {
				t.Fatal("agent refused the receipt:", err)
			}
			if err := verifyRetirementReceipt(ctx, hostguard.FenceReceipt{Target: result.Target, Bytes: result.Bytes, EvidenceAt: result.EvidenceAt}); err != nil {
				t.Fatal("root's retirement check refused the receipt:", err)
			}

			// Two console creates' holds land; the agent retries with the newer
			// numeral. Same fence, same receipt, the original numeral.
			w.roster.holdChurn()
			w.roster.holdChurn()
			later := w.fresh()
			if later.Revision != actor.Revision+4 || later.Binding != actor.Binding {
				t.Fatalf("hold churn moved revision %d -> %d, binding %v", actor.Revision, later.Revision, later.Binding == actor.Binding)
			}
			retry := w.request(later)
			again, err := w.coordinator().FenceAndStop(ctx, retry)
			if err != nil {
				t.Fatal("retry after hold churn:", err)
			}
			if !bytes.Equal(again.Bytes, result.Bytes) || again.Target != result.Target {
				t.Fatal("retry after hold churn did not return the original receipt")
			}
			if _, err := validateTransportStop(retry, again); err != nil {
				t.Fatal("agent refused the original numeral on retry:", err)
			}
		})
	}

	// The no-process witness (writer session zero): no fence, the witness's
	// own target is the one source.
	w := newReceiptWorld(t, true)
	o := w.claim()
	if _, err := hostclaim.Update(ctx, w.roster.store, o, func(c *hostclaim.Claim) error {
		c.Holder.Session, c.Holder.WriterBox, c.Holder.Generation, c.Holder.Admissions = 0, "", "", nil
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	actor := w.fresh()
	req := w.request(actor)
	result, err := w.coordinator().FenceAndStop(ctx, req)
	if err != nil {
		t.Fatal("no-process witness:", err)
	}
	r := decodeStopReceipt(t, result.Bytes)
	if r.Kind != StopReceiptNoProcess || r.DirectoryRevision != strconv.FormatUint(actor.Revision, 10) || r.Binding != actor.Binding || result.Target.RosterRevision != actor.Revision {
		t.Fatalf("no-process receipt numeral %q binding %q kind %q", r.DirectoryRevision, r.Binding, r.Kind)
	}
	if _, err := validateTransportStop(req, result); err != nil {
		t.Fatal("agent refused the no-process receipt:", err)
	}
	if err := validateServingLegacyStop(servingLegacyStop{Observed: o, Result: result}); err == nil {
		// o is the claim before the session reset; the stored legacy stop
		// names the claim the witness read.
		t.Fatal("legacy stop validated against the wrong claim")
	}
	if err := validateServingLegacyStop(servingLegacyStop{Observed: w.claim(), Result: result}); err != nil {
		t.Fatal("retained no-process stop:", err)
	}
}

// N1 (roster binding task 4): the fence persists the requester's roster
// numeral before any stop effect. Root crashes after BeginFence and before
// CompleteFence; holds land; a restarted root's RecoverPending recovers the
// stop from the fence alone. The recovered receipt's directory_revision is
// the persisted numeral and its binding the installation's stamp, and the
// branch's unchanged console verifiers accept it: the purge's
// StopBeforeTeardown for an operation stop, WakeResetTransport.VerifyReceipt
// for a serving stop. A fence persisted without a numeral is refused.
func TestRecoveredReceiptKeepsRosterNumeral(t *testing.T) {
	ctx := t.Context()
	for _, serving := range []bool{false, true} {
		t.Run(fmt.Sprintf("serving=%v", serving), func(t *testing.T) {
			w := newReceiptWorld(t, serving)
			actor := w.fresh()
			req := w.request(actor)
			stopped := w.claim()
			crash := errors.New("root crashed after the fence")
			w.kernel.fail = crash
			if _, err := w.coordinator().FenceAndStop(ctx, req); !errors.Is(err, crash) {
				t.Fatal(err)
			}
			pending, err := w.guard().PendingStopFences(ctx)
			if err != nil || len(pending) != 1 || pending[0].RosterRevision != actor.Revision || pending[0].DirectoryRevision != actor.Binding {
				t.Fatalf("persisted fence %+v %v", pending, err)
			}

			w.roster.holdChurn()
			w.roster.activationChurn()
			w.kernel.fail = nil
			if err := (ClaimStopHTTP{Coordinator: w.coordinator()}).RecoverPending(ctx); err != nil {
				t.Fatal("restarted root's recovery:", err)
			}
			if len(w.kernel.recovered) != 1 {
				t.Fatalf("recovered %d", len(w.kernel.recovered))
			}
			durable, err := w.guard().FenceReceiptFor(ctx, pending[0])
			if err != nil {
				t.Fatal(err)
			}
			r := decodeStopReceipt(t, durable.Bytes)
			if r.DirectoryRevision != strconv.FormatUint(actor.Revision, 10) || r.Binding != actor.Binding {
				t.Fatalf("recovered receipt numeral %q binding %q; want %d and %s", r.DirectoryRevision, r.Binding, actor.Revision, actor.Binding)
			}

			// The agent's next stop request carries the newer numeral and is
			// a recovery of the dead controller's tuple: root publishes the
			// recovered receipt, unchanged.
			later := w.fresh()
			if later.Revision != actor.Revision+4 || later.Binding != actor.Binding {
				t.Fatalf("churn moved revision %d -> %d", actor.Revision, later.Revision)
			}
			retry := w.request(later)
			retry.Recover = true
			result, err := w.coordinator().FenceAndStop(ctx, retry)
			if err != nil {
				t.Fatal("publication of the recovered receipt:", err)
			}
			if !bytes.Equal(result.Bytes, durable.Bytes) || result.Target != pending[0] {
				t.Fatal("published another receipt than the recovered one")
			}
			if _, err := validateTransportStop(retry, result); err != nil {
				t.Fatal("agent refused the recovered receipt:", err)
			}
			if serving {
				if err := w.wakeResetVerify(stopped); err != nil {
					t.Fatal("unchanged wake-reset verifier refused the recovered receipt:", err)
				}
			} else if err := w.purgeVerify(); err != nil {
				t.Fatal("unchanged purge verifier refused the recovered receipt:", err)
			}
		})
	}

	// A fence persisted before roster binding has no numeral: recovery
	// refuses it before any kill and journals nothing.
	w := newReceiptWorld(t, false)
	actor := w.fresh()
	req := w.request(actor)
	legacy := hostguard.FenceTarget{FenceID: "legacy", ClaimGen: w.claim().Claim.Gen, Region: req.Region, Host: req.Host, Incarnation: req.Incarnation, Stamp: req.InstallationStamp, Session: req.WriterSession, Op: req.OperationID, Phase: req.Phase, ObligationID: req.ObligationID,
		Box: req.Box, Enrollment: req.Enrollment, BootID: req.BootID, ControllerID: req.ControllerID, ClaimETag: req.ExpectedClaimETag, DirectoryRevision: req.ExpectedDirectoryRevision}
	if _, err := w.guard().BeginFence(ctx, legacy); err != nil {
		t.Fatal(err)
	}
	if err := (ClaimStopHTTP{Coordinator: w.coordinator()}).RecoverPending(ctx); !errors.Is(err, ErrClaimStopReceipt) || len(w.kernel.recovered) != 0 {
		t.Fatalf("legacy fence recovery: %v, kills %d", err, len(w.kernel.recovered))
	}
	if _, err := w.guard().FenceReceiptFor(ctx, legacy); !errors.Is(err, hostguard.ErrNoFenceReceipt) {
		t.Fatalf("legacy fence journaled a receipt: %v", err)
	}
}

// publishTampered publishes body, re-signed with the box's evidence key,
// under a fresh receipt id: a receipt only a wrong producer would write.
func (w *receiptWorld) publishTampered(result StopResult, edit func(map[string]any)) {
	w.t.Helper()
	var fields map[string]any
	if err := json.Unmarshal(result.Bytes, &fields); err != nil {
		w.t.Fatal(err)
	}
	edit(fields)
	fields["id"] = "tampered"
	body, err := json.Marshal(fields)
	if err != nil {
		w.t.Fatal(err)
	}
	r := decodeStopReceipt(w.t, body)
	if err := hostevidence.Publish(w.t.Context(), w.roster.store, w.key, stopEvidenceKey(r.Host, r.OperationID, r.Box, r.ID), body); err != nil {
		w.t.Fatal(err)
	}
}

// stopUnpublished collects a genuine receipt through the real coordinator
// without publishing it into the deployment store.
func (w *receiptWorld) stopUnpublished(req StopRequest) StopResult {
	w.t.Helper()
	c := w.coordinator()
	c.Evidence = release.NewDirStore(w.t.TempDir())
	result, err := c.FenceAndStop(w.t.Context(), req)
	if err != nil {
		w.t.Fatal(err)
	}
	return result
}

// L7b (roster binding R1, mixed version): a task 4 receipt, signed with the
// box's evidence key under the operation's evidence/ key, passes the
// branch's unchanged StopBeforeTeardown and verifyEvidence after further
// roster writes. The same receipt with the binding in directory_revision is
// refused.
func TestPurgeAcceptsC1ReceiptOnUnchangedVerifier(t *testing.T) {
	w := newReceiptWorld(t, false)
	if _, err := w.coordinator().FenceAndStop(t.Context(), w.request(w.fresh())); err != nil {
		t.Fatal(err)
	}
	w.roster.holdChurn()
	if err := w.purgeVerify(); err != nil {
		t.Fatal("unchanged purge verifier refused a task 4 receipt:", err)
	}

	w = newReceiptWorld(t, false)
	result := w.stopUnpublished(w.request(w.fresh()))
	w.publishTampered(result, func(f map[string]any) { f["directory_revision"] = f["binding"]; delete(f, "binding") })
	if err := w.purgeVerify(); !errors.Is(err, hostops.ErrIdentity) {
		t.Fatalf("purge verifier with the binding in directory_revision: %v", err)
	}
}

// L7c (roster binding R1, mixed version): the same through the unchanged
// WakeResetTransport.VerifyReceipt under serving-evidence/, and a
// quarantined actor is still refused before any receipt is read.
func TestWakeResetAcceptsC1ReceiptOnUnchangedVerifier(t *testing.T) {
	w := newReceiptWorld(t, true)
	stopped := w.claim()
	if _, err := w.coordinator().FenceAndStop(t.Context(), w.request(w.fresh())); err != nil {
		t.Fatal(err)
	}
	w.roster.holdChurn()
	if err := w.wakeResetVerify(stopped); err != nil {
		t.Fatal("unchanged wake-reset verifier refused a task 4 receipt:", err)
	}
	w.roster.quarantine("loss_1")
	if err := w.wakeResetVerify(stopped); !errors.Is(err, hostops.ErrIdentity) {
		t.Fatalf("wake reset for a quarantined actor: %v", err)
	}

	w = newReceiptWorld(t, true)
	stopped = w.claim()
	result := w.stopUnpublished(w.request(w.fresh()))
	w.publishTampered(result, func(f map[string]any) { f["directory_revision"] = f["binding"]; delete(f, "binding") })
	if err := w.wakeResetVerify(stopped); !errors.Is(err, hostops.ErrWakeResetUncertain) {
		t.Fatalf("wake-reset verifier with the binding in directory_revision: %v", err)
	}
}