package control

import (
	"cloud.google.com/go/storage"
	"context"
	"encoding/json"
	"os"
	"strconv"
	"sync/atomic"
	"testing"
	"time"
)

const fixtureManifestURI = "gs://gcp-digital-bi-pingcap-nexuscrm-ingestion/manifests/source=nexuscrm/run_date=2026-09-03/run_id=fixture-run/manifest.json"

func manifestFixture() (manifestArtifact, time.Time) {
	start := time.Date(2026, 9, 3, 2, 17, 0, 0, time.UTC)
	body, _ := json.Marshal(map[string]any{"manifest_schema_version": 3, "source_name": "nexuscrm", "target_project": "gcp-digital-bi", "run_id": "fixture-run", "manifest_uri": fixtureManifestURI, "product_status": "exported", "created_at": start.Add(-time.Minute), "tables": []map[string]string{{"table_name": "Order"}}})
	return manifestArtifact{Body: body, Generation: 123, Updated: start.Add(-time.Minute)}, start
}

func TestManifestEvidenceKeepsAcceptanceBoundaries(t *testing.T) {
	artifact, start := manifestFixture()
	e := evaluateManifest(fixtureManifestURI, "fixture-run", "Order", start, start.Add(time.Minute), start.Add(time.Hour), artifact)
	if e.Status != "consistent" || e.Generation != "123" || len(e.SHA256) != 64 {
		t.Fatal("missing bounded manifest evidence", e)
	}
	artifact.Updated = start.Add(time.Second)
	if e = evaluateManifest(fixtureManifestURI, "fixture-run", "Order", start, start.Add(time.Minute), start, artifact); e.Status != "current_pointer_only" {
		t.Fatal("rewritten artifact attributed to old load")
	}
	if e = evaluateManifest(fixtureManifestURI, "fixture-run", "Order", time.Time{}, start, start, artifact); e.Status != "current_pointer_only" {
		t.Fatal("missing owner start treated as immutable lineage")
	}
}

func TestManifestRejectsMissingAmbiguousAndDifferentSource(t *testing.T) {
	artifact, start := manifestFixture()
	for _, change := range []map[string]any{
		{"manifest_schema_version": 99}, {"source_name": "other"}, {"target_project": "other"}, {"run_id": "another-run"}, {"product_status": "running"},
		{"tables": []map[string]string{{"table_name": "Account"}}},
		{"tables": []map[string]string{{"table_name": "Order"}, {"table_name": "Order"}}},
	} {
		var doc map[string]any
		_ = json.Unmarshal(artifact.Body, &doc)
		for k, v := range change {
			doc[k] = v
		}
		body, _ := json.Marshal(doc)
		bad := artifact
		bad.Body = body
		if e := evaluateManifest(fixtureManifestURI, "fixture-run", "Order", start, start, start, bad); e.Status != "mismatch" {
			t.Fatal("invalid manifest admitted", change, e)
		}
	}
	for _, uri := range []string{"", "gs://unapproved-bucket/manifest.json", fixtureManifestURI + "?token=private"} {
		if e := evaluateManifest(uri, "fixture-run", "Order", start, start, start, artifact); e.Status != "unbound" || e.URI != "" {
			t.Fatal("unapproved URI retained or followed")
		}
	}
	if e := evaluateManifest(fixtureManifestURI, "fixture-run", "Order", start, start, start, manifestArtifact{}); e.Status != "unavailable" {
		t.Fatal("unreadable object fabricated as good")
	}
}

func TestManifestRepollDoesNotDuplicateEvidenceOrPassVerification(t *testing.T) {
	o := testOperations(t)
	artifact, start := manifestFixture()
	e := evaluateManifest(fixtureManifestURI, "fixture-run", "Order", start, start, start, artifact)
	b := testBatch("succeeded", start)
	b.Runs[0].Manifest = &e
	if err := o.CommitBatch(context.Background(), b); err != nil {
		t.Fatal(err)
	}
	e.ObservedAt = start.Add(time.Hour)
	b.State.ObservedAt = e.ObservedAt
	if err := o.CommitBatch(context.Background(), b); err != nil {
		t.Fatal(err)
	}
	if countTable(t, o, "observations") != 1 {
		t.Fatal("manifest poll time created new evidence")
	}
	runs, err := o.Runs(context.Background(), RunFilter{})
	if err != nil || runs[0].Verification != "not_checked" {
		t.Fatal("artifact consistency became business acceptance")
	}
	e.Generation = "124"
	if err := o.CommitBatch(context.Background(), b); err != nil {
		t.Fatal(err)
	}
	if countTable(t, o, "observations") != 2 {
		t.Fatal("new object version not preserved")
	}
}

func TestManifestReadsAreBoundedAndCancellationDoesNotAdvance(t *testing.T) {
	requests := map[string]string{}
	for _, key := range []string{"a", "b", "c", "d", "e", "f", "g", "h"} {
		requests[key] = key
	}
	var active, peak atomic.Int32
	values, err := fetchManifestArtifacts(context.Background(), requests, func(ctx context.Context, uri, id string) (manifestArtifact, error) {
		n := active.Add(1)
		defer active.Add(-1)
		for {
			old := peak.Load()
			if n <= old || peak.CompareAndSwap(old, n) {
				break
			}
		}
		time.Sleep(10 * time.Millisecond)
		return manifestArtifact{Generation: 1}, nil
	})
	if err != nil || len(values) != 8 || peak.Load() > 4 || peak.Load() < 2 {
		t.Fatal("artifact reads are unbounded or serialized", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	if _, err := fetchManifestArtifacts(ctx, requests, func(ctx context.Context, uri, id string) (manifestArtifact, error) {
		return manifestArtifact{}, ctx.Err()
	}); err == nil {
		t.Fatal("cancelled scan would advance the checkpoint with missing evidence")
	}
}

func TestExplicitLiveManifestRead(t *testing.T) {
	uri, runID := os.Getenv("FINANCE_PORTAL_TEST_MANIFEST_URI"), os.Getenv("FINANCE_PORTAL_TEST_MANIFEST_RUN")
	if uri == "" || runID == "" {
		t.Skip("requires explicitly bound live read-only artifact")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	gcs, err := storage.NewClient(ctx)
	if err != nil {
		t.Fatal("GCS read client unavailable")
	}
	defer gcs.Close()
	s := &ExternalLiveSource{gcs: gcs}
	start := time.Now()
	artifact, err := s.readCRMManifest(ctx, uri, runID)
	t.Logf("generation=%d bytes=%d failure=%s elapsed=%s", artifact.Generation, len(artifact.Body), artifact.Failure, time.Since(start))
	if err != nil {
		t.Fatal("live artifact read failed")
	}
}

func TestCRMBackfillPacingIsBoundedAndStopsOnStalledCursor(t *testing.T) {
	for _, stalled := range []bool{false, true} {
		t.Run(strconv.FormatBool(stalled), func(t *testing.T) {
			o := testOperations(t)
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			var count atomic.Int32
			calls := make(chan int, 10)
			o.StartCollectors(ctx, []HistorySource{historySource{"crm_load", func(context.Context, CollectionState) (RunBatch, error) {
				n := int(count.Add(1))
				calls <- n
				cursor := strconv.Itoa(n)
				if stalled {
					cursor = "same"
				}
				return RunBatch{State: CollectionState{Status: "partial", Cursor: cursor}}, nil
			}}}, time.Hour)
			expected := 5
			if stalled {
				expected = 2
			}
			for i := 0; i < expected; i++ {
				select {
				case <-calls:
				case <-time.After(time.Second):
					t.Fatal("backfill did not advance promptly")
				}
			}
			select {
			case <-calls:
				t.Fatal("backfill ignored page budget or stalled cursor")
			case <-time.After(30 * time.Millisecond):
			}
		})
	}
}