package runtimeapi

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	permissionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/effectivepermissionsv1"
	querysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
	accesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1"
)

// Counts real store reads, rather than replacing artifact validation. Reopening
// the same immutable source for every capability made full-size startup exceed
// the deployment readiness deadline.
type observedServingStore struct {
	catalogsnapshot.Store
	reads map[string]int
}

func (s *observedServingStore) Get(ctx context.Context, kind catalogsnapshot.ArtifactKind, digest catalogsnapshot.Digest) (catalogsnapshot.Artifact, error) {
	s.reads[servingArtifactKey(kind, digest)]++
	return s.Store.Get(ctx, kind, digest)
}

func TestServingStartupArtifactReads(t *testing.T) {
	fixture := newServingSetFixture(t, 7, []servingCapabilitySetEntry{
		{testSFCRMAppCapabilityID, accesssemantic.CapabilityName, catalogsnapshot.ProjectionRecordAccessV1},
		{"io.example.query.v1", querysemantic.CapabilityName, catalogsnapshot.ProjectionRecordQueryV1},
		{"io.example.permissions.v1", permissionsemantic.CapabilityName, catalogsnapshot.ProjectionEffectivePermissionsV1},
	})
	root := t.TempDir()
	directory, err := catalogsnapshot.NewDirectoryStore(root)
	if err != nil {
		t.Fatal(err)
	}
	for _, artifact := range fixture.state.store.(*servingMemoryStore).artifacts {
		if err := directory.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	store := &observedServingStore{Store: directory, reads: map[string]int{}}
	fixture.state.store = store
	if err := fixture.state.ReconcileAll(t.Context(), fixture.selections); err != nil {
		t.Fatal(err)
	}
	for key, count := range store.reads {
		if count != 1 {
			t.Errorf("immutable input %s read %d times in one preparation, want once", key, count)
		}
	}
	// Reconciliation owns live authority. A later attempt must read and verify
	// its inputs again; successful startup cannot conceal subsequent corruption.
	store.reads = map[string]int{}
	if err := fixture.state.ReconcileAll(t.Context(), fixture.selections); err != nil {
		t.Fatal(err)
	}
	if len(store.reads) == 0 {
		t.Fatal("later preparation reused process-lifetime authority")
	}
	for _, count := range store.reads {
		if count != 1 {
			t.Fatal("later preparation repeated immutable reads")
		}
	}
	// Damage a real immutable file after readiness, then repair it. Every new
	// attempt must revalidate storage and invalidate readiness on failure.
	for key := range store.reads {
		kind, digest, _ := strings.Cut(key, "\x00")
		if kind != string(catalogsnapshot.ArtifactCatalogSnapshot) {
			continue
		}
		paths, err := filepath.Glob(filepath.Join(root, "*", strings.TrimPrefix(digest, "sha256:")+".json"))
		if err != nil || len(paths) != 1 {
			t.Fatalf("snapshot file: %v %v", paths, err)
		}
		raw, err := os.ReadFile(paths[0])
		if err != nil {
			t.Fatal(err)
		}
		// Published artifacts are read-only. Fault injection changes permissions
		// only inside this test's disposable directory.
		if err := os.Chmod(paths[0], 0600); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(paths[0], []byte("{}"), 0600); err != nil {
			t.Fatal(err)
		}
		if err := fixture.state.ReconcileAll(t.Context(), fixture.selections); err == nil || fixture.state.Ready() {
			t.Fatal("previous successful preparation concealed corrupt storage")
		}
		if err := os.WriteFile(paths[0], raw, 0600); err != nil {
			t.Fatal(err)
		}
		if err := fixture.state.ReconcileAll(t.Context(), fixture.selections); err != nil || !fixture.state.Ready() {
			t.Fatalf("repaired storage did not recover: %v", err)
		}
	}
}

func TestServingArtifactProofRequiresExactSourceAndChecksCancellation(t *testing.T) {
	first, next := newServingFixture(t, 7), newServingFixture(t, 8)
	a := newServingArtifactSet(first.store)
	receipt, err := a.receipt(t.Context(), first.selection.ExpectedCompatibilityReceiptDigest)
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := a.snapshot(t.Context(), receipt.CatalogSnapshotDigest())
	if err != nil {
		t.Fatal(err)
	}
	view, err := a.view(t.Context(), receipt.RuntimeMetadataViewDigest())
	if err != nil {
		t.Fatal(err)
	}
	if err := a.verifySource(view, snapshot); err != nil {
		t.Fatal(err)
	}
	other := newServingArtifactSet(next.store)
	otherReceipt, err := other.receipt(t.Context(), next.selection.ExpectedCompatibilityReceiptDigest)
	if err != nil {
		t.Fatal(err)
	}
	otherSnapshot, err := other.snapshot(t.Context(), otherReceipt.CatalogSnapshotDigest())
	if err != nil {
		t.Fatal(err)
	}
	if err := a.verifySource(view, otherSnapshot); err == nil {
		t.Fatal("source proof reused for a different revision")
	}
	ctx, cancel := context.WithCancel(t.Context())
	cancel()
	if _, err := a.snapshot(ctx, snapshot.Digest()); err == nil {
		t.Fatal("shared snapshot ignored cancellation")
	}
	if _, err := a.view(ctx, view.Digest()); err == nil {
		t.Fatal("shared view ignored cancellation")
	}
	if _, err := a.receipt(ctx, receipt.Digest()); err == nil {
		t.Fatal("shared receipt ignored cancellation")
	}
}

// Opt-in profiling workload, not a timing assertion or a substitute for the
// exact deployed-candidate walking slice. The ordinary regression above is
// unconditional and independent of host speed.
func TestServingStartupProfile(t *testing.T) {
	if os.Getenv("NEXUSCRM_SERVING_STARTUP_PROFILE") != "1" {
		t.Skip("explicit startup profiling workload")
	}
	var objects []*models.ObjectMetadata
	for i := 0; i < 200; i++ {
		name := fmt.Sprintf("Widget%03d", i)
		object := &models.ObjectMetadata{ID: "object-" + name, APIName: name, Label: name, PluralLabel: name + "s", SharingModel: constants.SharingModelPublicReadWrite}
		for j := 0; j < 20; j++ {
			field := fmt.Sprintf("Field%03d", j)
			object.Fields = append(object.Fields, models.FieldMetadata{ID: name + "-" + field, APIName: field, PhysicalName: field, Label: field, Type: constants.FieldTypeString})
		}
		objects = append(objects, object)
	}
	entries := []servingCapabilitySetEntry{
		{testSFCRMAppCapabilityID, accesssemantic.CapabilityName, catalogsnapshot.ProjectionRecordAccessV1},
		{"io.example.query.v1", querysemantic.CapabilityName, catalogsnapshot.ProjectionRecordQueryV1},
		{"io.example.permissions.v1", permissionsemantic.CapabilityName, catalogsnapshot.ProjectionEffectivePermissionsV1},
	}
	fixture := newServingSetPublicationFixtureWithObjects(t, 7, 1, 0, entries, nil, objects)
	directory, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	for _, artifact := range fixture.state.store.(*servingMemoryStore).artifacts {
		if err := directory.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	store := &observedServingStore{Store: directory, reads: map[string]int{}}
	fixture.state.store = store
	start := time.Now()
	if err := fixture.state.ReconcileAll(t.Context(), fixture.selections); err != nil {
		t.Fatal(err)
	}
	t.Logf("startup: objects=%d fields/object=20 capabilities=%d elapsed=%s reads=%v", len(objects), len(entries), time.Since(start), store.reads)
}