package host

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

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

func TestPlan9LegacyJournalNeedsExactRootSignedStop(t *testing.T) {
	root := ed25519.NewKeyFromSeed(bytes.Repeat([]byte{1}, ed25519.SeedSize))
	signer := ed25519.NewKeyFromSeed(bytes.Repeat([]byte{2}, ed25519.SeedSize))
	store := release.NewDirStore(t.TempDir())
	enrollment := strings.Repeat("a", 32)
	roster := hostactors.Roster{V: 1, DeploymentBucket: "bucket", Revision: 3, RequestLimitNS: int64(hostbox.RequestLimit), Regions: map[string]hostactors.Region{"ie": {}}, Boxes: map[string]hostactors.Box{
		"box": {CensusPublicKey: signer.Public().(ed25519.PublicKey), Enrollment: enrollment, Namespace: "platform", BoxID: "box", Label: "box", Region: "ie", Pool: "default", Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: "ie", InstanceID: "i-box"}, EvidenceSHA256: strings.Repeat("c", 64), State: "active"},
	}, Consoles: map[string]hostactors.Console{}, Holds: map[string]hostactors.Reservation{}, Activations: map[string]hostactors.Reservation{}}
	raw, err := release.SignRootEnvelope(release.RootSigner(root), roster)
	if err != nil {
		t.Fatal(err)
	}
	if _, err = store.Put(t.Context(), hostactors.Key, raw, ""); err != nil {
		t.Fatal(err)
	}
	newBatch := func() func(context.Context, hostbox.LaunchInventory, hostbox.StopInventory, hostbox.StopKernelJournal) error {
		return legacyJournalProof(store, hostactors.Directory{Store: store, PublicKey: root.Public().(ed25519.PublicKey), Bucket: "bucket"})
	}
	verify := newBatch()
	j, err := hostbox.OpenLaunchJournal(filepath.Join(t.TempDir(), "legacy.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer j.Close()
	in := hostbox.LaunchInventory{Target: hostguard.FenceTarget{Host: "app.example", Incarnation: "inc", Stamp: 1, Session: 1, Box: "box", Enrollment: enrollment, BootID: "boot", ControllerID: "controller", Op: "operation"}, ControllerPID: 1, ProcessStart: "1", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "session", Device: 1, Inode: 2}}}
	if err = j.Create(t.Context(), in); err != nil {
		t.Fatal(err)
	}
	journal, err := j.ReadStopKernelJournal(t.Context(), in.Target)
	if err != nil {
		t.Fatal(err)
	}
	inventory := hostbox.StopInventory{Complete: true, Host: in.Target.Host, Incarnation: "inc", InstallationStamp: 1, Box: "box", Enrollment: enrollment, DirectoryRevision: "1"}
	if err = verify(t.Context(), in, inventory, journal); err == nil {
		t.Fatal("unproven legacy accepted")
	}
	target := in.Target
	receipt := StopReceipt{FenceID: "fence", ClaimGen: 1, ObservedBootID: "boot", DirectoryRevision: "1", Binding: "1", CgroupIdentities: []string{"session"}, V: 1, ID: "stop1", Kind: "stopped", Host: target.Host, Incarnation: target.Incarnation, InstallationStamp: target.Stamp, WriterSession: target.Session, Box: target.Box, Enrollment: target.Enrollment, BootID: target.BootID, ControllerID: target.ControllerID, OperationID: target.Op, RequestLimitNS: int64(hostbox.RequestLimit), EvidenceAt: time.Now().UTC(), InventorySHA256: digestStop(inventory), LocalJournalSHA256: digestStop(journal)}
	raw, _ = json.Marshal(receipt)
	key := "control/hosts/app.example/evidence/operation/box/stop1.json"
	if _, err = store.Put(t.Context(), key, raw, ""); err != nil {
		t.Fatal(err)
	}
	verify = newBatch()
	if err = verify(t.Context(), in, inventory, journal); err == nil {
		t.Fatal("unsigned stop accepted")
	}
	if err = hostevidence.Publish(t.Context(), store, signer, key, raw); err != nil {
		t.Fatal(err)
	}
	if err = verify(t.Context(), in, inventory, journal); err == nil {
		t.Fatal("batch silently refreshed unsigned evidence")
	}
	// Two distinct writer histories under one evidence namespace are fetched
	// and indexed once, rather than rescanning every receipt for each row.
	second := in
	second.Target.Session = 2
	secondJournal := journal
	secondJournal.Target = second.Target
	secondReceipt := receipt
	secondReceipt.ID = "stop2"
	secondReceipt.WriterSession = 2
	secondReceipt.LocalJournalSHA256 = digestStop(secondJournal)
	secondRaw, _ := json.Marshal(secondReceipt)
	secondKey := "control/hosts/app.example/evidence/operation/box/stop2.json"
	if err = hostevidence.Publish(t.Context(), store, signer, secondKey, secondRaw); err != nil {
		t.Fatal(err)
	}
	counted := &legacyProofCountStore{Store: store, t: t, gets: map[string]int{}, lists: map[string]int{}}
	directory := hostactors.Directory{Store: counted, PublicKey: root.Public().(ed25519.PublicKey), Bucket: "bucket"}
	verify = legacyJournalProof(counted, directory)
	if err = verify(t.Context(), second, inventory, secondJournal); err != nil {
		t.Fatal(err)
	}
	if err = verify(t.Context(), in, inventory, journal); err != nil {
		t.Fatal(err)
	}
	if counted.gets[hostactors.Key] != 1 || counted.lists["control/hosts/app.example/evidence/operation/box/"] != 1 {
		t.Fatalf("batch repeated roster/list: %v %v", counted.gets, counted.lists)
	}
	for _, key := range []string{key, key + ".sig", secondKey, secondKey + ".sig"} {
		if counted.gets[key] != 1 {
			t.Fatalf("receipt fetched %d times: %s", counted.gets[key], key)
		}
	}
	cancelled, cancel := context.WithCancel(t.Context())
	cancel()
	if err = verify(cancelled, in, inventory, journal); !errors.Is(err, context.Canceled) {
		t.Fatalf("cached proof ignored cancellation: %v", err)
	}
	if counted.gets[hostactors.Key] != 1 {
		t.Fatal("canceled cache lookup reread roster")
	}
	// In-flight cancellation is sticky in this immutable batch, while a new
	// startup gets a fresh closure and can retry the failed read.
	during, stopRead := context.WithCancel(t.Context())
	counted.onGet = func(ctx context.Context, key string) error {
		if key == hostactors.Key {
			stopRead()
			return ctx.Err()
		}
		return nil
	}
	interrupted := legacyJournalProof(counted, directory)
	if err = interrupted(during, in, inventory, journal); !errors.Is(err, context.Canceled) {
		t.Fatalf("in-flight cancellation: %v", err)
	}
	before := counted.gets[hostactors.Key]
	counted.onGet = nil
	if err = interrupted(t.Context(), in, inventory, journal); !errors.Is(err, context.Canceled) || counted.gets[hostactors.Key] != before {
		t.Fatal("failed batch reread roster", err)
	}
	if err = legacyJournalProof(counted, directory)(t.Context(), in, inventory, journal); err != nil {
		t.Fatal("fresh batch cannot retry", err)
	}
	changed := journal
	changed.Completed = hostbox.ProducerCompletion{Count: 1, Digest: strings.Repeat("d", 64)}
	if err = verify(t.Context(), in, inventory, changed); err == nil {
		t.Fatal("forged completed counter accepted")
	}
	inventory.DirectoryRevision = "2"
	if err = verify(t.Context(), in, inventory, journal); err == nil {
		t.Fatal("changed inventory accepted")
	}
}

type legacyProofCountStore struct {
	release.Store
	t           *testing.T
	gets, lists map[string]int
	onGet       func(context.Context, string) error
}

func (s *legacyProofCountStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.t.Helper()
	if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) > 30*time.Second {
		s.t.Fatal("unbounded migration object read")
	}
	s.gets[key]++
	if s.onGet != nil {
		if err := s.onGet(ctx, key); err != nil {
			return nil, "", err
		}
	}
	return s.Store.Get(ctx, key)
}
func (s *legacyProofCountStore) List(ctx context.Context, prefix string) ([]string, error) {
	s.t.Helper()
	if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) > 30*time.Second {
		s.t.Fatal("unbounded migration namespace read")
	}
	s.lists[prefix]++
	return s.Store.List(ctx, prefix)
}