package host

import (
	"context"
	"encoding/json"
	"strings"
	"sync"
	"time"

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

// One closure is one immutable migration batch. Its verified roster and each
// evidence namespace are loaded once; both successful reads and errors are
// sticky until a new root startup creates a fresh closure. No cache is global
// or survives migration. Only signatures, never copied counters, admit rows.
func legacyJournalProof(store release.Store, directory hostactors.Directory) func(context.Context, hostbox.LaunchInventory, hostbox.StopInventory, hostbox.StopKernelJournal) error {
	store = legacyProofReadStore{Store: store}
	directory.Store = legacyProofReadStore{Store: directory.Store}
	type proofSet struct {
		matches map[string]struct{}
		err     error
	}
	var mu sync.Mutex
	var roster hostactors.Observed
	var rosterErr error
	loadedRoster := false
	namespaces := map[string]proofSet{}
	return func(ctx context.Context, in hostbox.LaunchInventory, inventory hostbox.StopInventory, journal hostbox.StopKernelJournal) error {
		if err := ctx.Err(); err != nil {
			return err
		}
		mu.Lock()
		defer mu.Unlock()
		if err := ctx.Err(); err != nil {
			return err
		}
		if !loadedRoster {
			roster, rosterErr = directory.Read(ctx)
			loadedRoster = true
		}
		if rosterErr != nil {
			return rosterErr
		}
		actor, ok := roster.Roster.Boxes[in.Target.Box]
		if !ok || actor.Enrollment != in.Target.Enrollment || roster.Roster.RequestLimitNS != int64(hostbox.RequestLimit) {
			return hostbox.ErrLegacyJournalProof
		}
		// A serving writer has no operation; its stop is published under
		// serving-evidence, exactly where stopEvidenceKey puts it.
		prefix := strings.TrimSuffix(stopEvidenceKey(in.Target.Host, in.Target.Op, in.Target.Box, ""), ".json")
		cached, ok := namespaces[prefix]
		if !ok {
			cached = proofSet{matches: map[string]struct{}{}}
			keys, err := store.List(ctx, prefix)
			cached.err = err
			if err == nil {
				for _, key := range keys {
					if err = ctx.Err(); err != nil {
						cached.err = err
						break
					}
					if !strings.HasPrefix(key, prefix) || !strings.HasSuffix(key, ".json") {
						continue
					}
					raw, _, err := store.Get(ctx, key)
					if err != nil {
						cached.err = err
						break
					}
					signature, _, err := store.Get(ctx, key+".sig")
					if err != nil {
						if ctx.Err() != nil {
							cached.err = ctx.Err()
							break
						}
						continue
					}
					if hostevidence.Verify(actor.CensusPublicKey, key, raw, signature) != nil {
						continue
					}
					var proof StopReceipt
					if json.Unmarshal(raw, &proof) != nil || proof.V != 1 || proof.Kind != "stopped" || proof.ID == "" || key != prefix+proof.ID+".json" || proof.RequestLimitNS != int64(hostbox.RequestLimit) || proof.EvidenceAt.IsZero() || proof.FenceID == "" || proof.ClaimGen == 0 || proof.ObservedBootID == "" || len(proof.CgroupIdentities) == 0 {
						continue
					}
					// Cache only the exact match digest, not every decoded receipt and its
					// cgroup vectors. Each later row is one map lookup, not a receipt scan.
					cached.matches[legacyJournalProofMatch(proof)] = struct{}{}
				}
			}
			if cached.err == nil {
				cached.err = ctx.Err()
			}
			namespaces[prefix] = cached
		}
		if cached.err != nil {
			return cached.err
		}
		if err := ctx.Err(); err != nil {
			return err
		}
		target := in.Target
		wanted := StopReceipt{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, Region: target.Region, Binding: inventory.DirectoryRevision, InventorySHA256: digestStop(inventory), LocalJournalSHA256: digestStop(journal)}
		if _, ok := cached.matches[legacyJournalProofMatch(wanted)]; ok {
			return nil
		}
		return hostbox.ErrLegacyJournalProof
	}
}
func legacyJournalProofMatch(p StopReceipt) string {
	return digestStop([]any{p.Host, p.Incarnation, p.InstallationStamp, p.WriterSession, p.Box, p.Enrollment, p.BootID, p.ControllerID, p.OperationID, p.Region, p.Binding, p.InventorySHA256, p.LocalJournalSHA256})
}

// A shared batch can cover many rows, but no individual network request may
// consume more than one read allowance. Neither wrapper outlives its closure.
type legacyProofReadStore struct{ release.Store }

func (s legacyProofReadStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	body, etag, err := s.Store.Get(ctx, key)
	if err == nil {
		err = ctx.Err()
	}
	return body, etag, err
}
func (s legacyProofReadStore) List(ctx context.Context, prefix string) ([]string, error) {
	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	keys, err := s.Store.List(ctx, prefix)
	if err == nil {
		err = ctx.Err()
	}
	return keys, err
}