package control

import (
	"context"
	"github.com/gin-gonic/gin"
	"io"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"testing"
	"time"
)

func testOperations(t *testing.T) *Operations {
	t.Helper()
	o, err := OpenOperations(filepath.Join(t.TempDir(), "operations.db"), "test")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { o.Close() })
	return o
}

func TestFinanceHistoryDistinguishesMissingConfigFromConnectionFailure(t *testing.T) {
	source := &ExternalLiveSource{}
	batch, err := source.financeHistory(context.Background(), CollectionState{})
	if err != nil || batch.State.Status != "unconfigured" || len(batch.Runs) != 0 {
		t.Fatal("missing configuration must not fabricate a task or collection execution failure")
	}
	source.config.FinanceDatabaseURL = "configured-but-unavailable"
	if _, err := source.financeHistory(context.Background(), CollectionState{}); err == nil {
		t.Fatal("a configured but unavailable connection must still fail")
	}
}
func testBatch(status string, at time.Time) RunBatch {
	return RunBatch{Runs: []ProviderRun{{Provider: "finance", Task: "salesforce_org", ExternalID: "run-1", Status: status, Verification: "not_checked", StartedAt: at.Add(-time.Hour), EventAt: at}}, State: CollectionState{Provider: "finance", Status: "partial", ObservedAt: at, Cursor: "page-1", CoveredFrom: at.AddDate(0, 0, -30)}}
}
func countTable(t *testing.T, o *Operations, table string) int {
	t.Helper()
	var n int
	if err := o.db.QueryRow("SELECT COUNT(*) FROM " + table).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

func TestOperationsAtomicReplayAndLateEvents(t *testing.T) {
	o := testOperations(t)
	ctx := context.Background()
	now := time.Now().UTC()
	b := testBatch("failed", now)
	for i := 0; i < 2; i++ {
		b.State.ObservedAt = now.Add(time.Duration(i) * time.Minute)
		if err := o.CommitBatch(ctx, b); err != nil {
			t.Fatal(err)
		}
	}
	if countTable(t, o, "runs") != 1 || countTable(t, o, "observations") != 1 || countTable(t, o, "alerts") != 1 || countTable(t, o, "deliveries") != 1 {
		t.Fatal("duplicate polling created duplicate facts or alerts")
	}
	late := testBatch("running", now.Add(-time.Minute))
	late.Runs[0].StartedAt = b.Runs[0].StartedAt
	if err := o.CommitBatch(ctx, late); err != nil {
		t.Fatal(err)
	}
	runs, err := o.Runs(ctx, RunFilter{})
	if err != nil || runs[0].Status != "failed" {
		t.Fatal("late running overwrote terminal state")
	}
	invalid := b
	invalid.State.Cursor = "must-not-commit"
	invalid.Runs = append([]ProviderRun{b.Runs[0]}, ProviderRun{})
	if err := o.CommitBatch(ctx, invalid); err == nil {
		t.Fatal("invalid batch accepted")
	}
	if o.Checkpoint(ctx, "finance").Cursor == "must-not-commit" {
		t.Fatal("cursor advanced without facts")
	}
}

func TestOperationsEnvironmentAndReopen(t *testing.T) {
	path := filepath.Join(t.TempDir(), "ops.db")
	o, err := OpenOperations(path, "test")
	if err != nil {
		t.Fatal(err)
	}
	if err = o.CommitBatch(context.Background(), testBatch("succeeded", time.Now())); err != nil {
		t.Fatal(err)
	}
	o.Close()
	o, err = OpenOperations(path, "test")
	if err != nil {
		t.Fatal(err)
	}
	if countTable(t, o, "runs") != 1 {
		t.Fatal("lost durable run")
	}
	o.Close()
	if other, err := OpenOperations(path, "production"); err == nil {
		other.Close()
		t.Fatal("environment isolation missing")
	}
}

func TestAcknowledgementDoesNotResolveAndRequiresBoundActor(t *testing.T) {
	o := testOperations(t)
	if err := o.CommitBatch(context.Background(), testBatch("failed", time.Now())); err != nil {
		t.Fatal(err)
	}
	var id string
	o.db.QueryRow("SELECT id FROM alerts").Scan(&id)
	r := gin.New()
	o.Register(r, []AlertOperator{{Actor: "operator@example.test", Token: "test-only-token"}})
	call := func(token string) int {
		w := httptest.NewRecorder()
		req := httptest.NewRequest("POST", "/api/v1/alerts/"+id+"/acknowledge", nil)
		req.Header.Set("Authorization", "Bearer "+token)
		r.ServeHTTP(w, req)
		return w.Code
	}
	if call("wrong") != 403 || call("test-only-token") != 200 || call("test-only-token") != 409 {
		t.Fatal("bad acknowledgement admission")
	}
	var status, actor string
	o.db.QueryRow("SELECT status,actor FROM alerts WHERE id=?", id).Scan(&status, &actor)
	if status != "acknowledged" || actor != "operator@example.test" {
		t.Fatal("actor or status incorrect")
	}
	runs, _ := o.Runs(context.Background(), RunFilter{})
	if runs[0].Status != "failed" {
		t.Fatal("ack changed business outcome")
	}
}

func TestRealStoreIsEmptyAndCannotAcceptDemo(t *testing.T) {
	path := filepath.Join(t.TempDir(), "state.json")
	s, err := NewStoreForMode(path, "observe")
	if err != nil {
		t.Fatal(err)
	}
	if len(s.ListCloses()) != 0 {
		t.Fatal("real store seeded demo")
	}
	if _, err = NewStoreForMode(path, "demo"); err != nil {
		t.Fatal(err)
	}
	if _, err = NewStoreForMode(path, "observe"); err == nil {
		t.Fatal("real store accepted seeded state")
	}
}

func TestSuccessfulRunnerDoesNotImplyAcceptance(t *testing.T) {
	s, err := NewStoreForMode(filepath.Join(t.TempDir(), "state.json"), "demo")
	if err != nil {
		t.Fatal(err)
	}
	c := s.ListCloses()[0]
	a, err := s.BeginAttempt(c.ID, "run_monthly_cloud_billing", "run", "test")
	if err != nil {
		t.Fatal(err)
	}
	if err = s.FinishAttempt(a.ID, RunnerResult{ExternalRunID: "external"}, nil); err != nil {
		t.Fatal(err)
	}
	c, _ = s.GetClose(c.ID)
	step := c.Steps["run_monthly_cloud_billing"]
	if step.Execution != ExecutionSucceeded || step.Verification != VerificationNotChecked || step.Action.Enabled || c.Steps["run_cloud_breakdown"].Action.Enabled {
		t.Fatal("unverified success unlocked execution")
	}
}

func TestCollectorRunsWithoutPageRequest(t *testing.T) {
	o := testOperations(t)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	o.StartCollectors(ctx, []HistorySource{historySource{"finance", func(context.Context, CollectionState) (RunBatch, error) {
		return testBatch("succeeded", time.Now()), nil
	}}}, time.Hour)
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		runs, _ := o.Runs(ctx, RunFilter{})
		if len(runs) > 0 {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatal("background collection requires page request")
}

func TestNotificationRetriesBoundedAndDoesNotResolve(t *testing.T) {
	o := testOperations(t)
	now := time.Now().UTC()
	ctx := context.Background()
	if err := o.CommitBatch(ctx, testBatch("failed", now)); err != nil {
		t.Fatal(err)
	}
	calls := 0
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		calls++
		if r.Header.Get("Idempotency-Key") == "" {
			t.Error("missing delivery identity")
		}
		w.WriteHeader(503)
	}))
	defer server.Close()
	for i := 0; i < 4; i++ {
		_ = o.deliverOne(ctx, server.Client(), server.URL, now.Add(time.Duration(i)*10*time.Minute))
	}
	if calls != 3 {
		t.Fatalf("expected bounded retry, got %d", calls)
	}
	var status string
	o.db.QueryRow("SELECT status FROM alerts").Scan(&status)
	if status != "open" {
		t.Fatal("delivery changed alert state")
	}
}

func TestStablePaginationAndTriggerWithoutRun(t *testing.T) {
	o := testOperations(t)
	ctx := context.Background()
	now := time.Now().UTC()
	b := testBatch("succeeded", now)
	for _, id := range []string{"two", "three"} {
		r := b.Runs[0]
		r.ExternalID = id
		b.Runs = append(b.Runs, r)
	}
	b.Triggers = []TriggerObservation{{ID: "trigger-no-run", ScheduleID: "schedule", At: now, Status: "failed", ObservedAt: now}}
	if err := o.CommitBatch(ctx, b); err != nil {
		t.Fatal(err)
	}
	first, err := o.Runs(ctx, RunFilter{Limit: 2})
	if err != nil || len(first) != 2 {
		t.Fatal("first page failed")
	}
	second, err := o.Runs(ctx, RunFilter{Limit: 2, BeforeTime: stamp(first[1].StartedAt), BeforeID: first[1].ID})
	if err != nil || len(second) != 1 || second[0].ID == first[0].ID || second[0].ID == first[1].ID {
		t.Fatal("keyset pagination repeated or lost a run")
	}
	if countTable(t, o, "triggers") != 1 || countTable(t, o, "alerts") != 1 {
		t.Fatal("submission without run lost")
	}
}

func TestV7PrerequisiteDAGHasNoCycles(t *testing.T) {
	visiting := map[string]bool{}
	visited := map[string]bool{}
	var walk func(string)
	walk = func(key string) {
		if visiting[key] {
			t.Fatalf("DAG cycle at %s", key)
		}
		if visited[key] {
			return
		}
		d, ok := definitionByKey(key)
		if !ok {
			t.Fatalf("unknown dependency %s", key)
		}
		visiting[key] = true
		for _, dep := range d.DependsOn {
			walk(dep)
		}
		visiting[key] = false
		visited[key] = true
	}
	for _, step := range MonthlyCloudClose {
		walk(step.Key)
	}
	for key, target := range map[string]string{"prepare_finance_org": "salesforce_org", "prepare_finance_account": "salesforce_account", "prepare_finance_order": "salesforce_order"} {
		d, ok := definitionByKey(key)
		if !ok || len(d.Scope) != 1 || d.Scope[0] != target {
			t.Fatalf("wrong prerequisite scope %s", key)
		}
	}
}

type testTransport func(*http.Request) (*http.Response, error)

func (f testTransport) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
func TestFivetranHistoryIsReadOnlyAndUsesStableSyncID(t *testing.T) {
	s := &ExternalLiveSource{config: LiveSourceConfig{FivetranAPIKey: "test", FivetranAPISecret: "test", FivetranConnectorID: "connector"}}
	s.httpClient = &http.Client{Transport: testTransport(func(r *http.Request) (*http.Response, error) {
		if r.Method != "GET" || !strings.HasSuffix(r.URL.Path, "/sync-history") || r.Header.Get("Accept") != "application/json;version=2" {
			t.Error("wrong sync history request")
		}
		start, e1 := time.Parse(time.RFC3339, r.URL.Query().Get("start_time"))
		end, e2 := time.Parse(time.RFC3339, r.URL.Query().Get("end_time"))
		if e1 != nil || e2 != nil || end.Sub(start) > 7*24*time.Hour {
			t.Error("invalid history window")
		}
		return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"code":"Success","data":{"items":[{"sync_id":"sync-123","start":"2026-09-01T00:00:00Z","end":"2026-09-01T01:00:00Z","status":"COMPLETED"}],"next_cursor":"page-two"}}`)), Header: http.Header{}}, nil
	})}
	b, err := s.fivetranHistory(context.Background(), CollectionState{})
	if err != nil {
		t.Fatal(err)
	}
	if len(b.Runs) != 1 || b.Runs[0].ExternalID != "sync-123" || b.Runs[0].Status != "succeeded" || b.Runs[0].Verification != "not_checked" || b.State.Cursor == "" {
		t.Fatal("history contract violated")
	}
}

func TestDBTHistoryRejectsWrongJobAndNeverDispatches(t *testing.T) {
	t.Setenv("FINANCE_PORTAL_DBT_ACCOUNT_ID", "123")
	t.Setenv("FINANCE_PORTAL_DBT_JOB_ID", "456")
	t.Setenv("DBT_CLOUD_API_TOKEN", "test")
	s := &ExternalLiveSource{}
	s.httpClient = &http.Client{Transport: testTransport(func(r *http.Request) (*http.Response, error) {
		if r.Method != "GET" || r.URL.Query().Get("job_definition_id") != "456" || !strings.HasSuffix(r.URL.Path, "/runs/") {
			t.Error("dbt scope or method incorrect")
		}
		return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"status":{"is_success":true},"data":[{"id":1,"job_definition_id":999,"created_at":"2026-09-07T00:00:00Z","status":10}]}`)), Header: http.Header{}}, nil
	})}
	if _, err := s.dbtHistory(context.Background(), CollectionState{}); err == nil {
		t.Fatal("accepted run from wrong job")
	}
}

func TestDirectAPIRequiresContractsBeyondWriteSwitch(t *testing.T) {
	s, err := NewStore(filepath.Join(t.TempDir(), "state.json"))
	if err != nil {
		t.Fatal(err)
	}
	r := gin.New()
	NewAPI(s, NewEngine(s, DemoRunner{}), "direct", "test-token", true).Register(r)
	w := httptest.NewRecorder()
	req := httptest.NewRequest("POST", "/api/v1/closes", strings.NewReader(`{"bill_month":"2026-08","mode":"actual"}`))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer test-token")
	r.ServeHTTP(w, req)
	if w.Code != 403 || len(s.ListCloses()) != 0 {
		t.Fatal("write switch bypassed unbound contracts")
	}
}