package control

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"google.golang.org/api/option"
	runapi "google.golang.org/api/run/v2"
)

func jobFixture(job, state string) *runapi.GoogleCloudRunV2Execution {
	return &runapi.GoogleCloudRunV2Execution{Name: job + "/executions/fixture", Job: job, StartTime: time.Now().UTC().Add(-time.Hour).Format(time.RFC3339), CompletionTime: time.Now().UTC().Format(time.RFC3339), Conditions: []*runapi.GoogleCloudRunV2Condition{{Type: "Completed", State: state}}}
}

func TestCRMJobExecutionStateDoesNotImplyOutput(t *testing.T) {
	job, task, _ := crmJobBinding("crm_load_job")
	for state, expected := range map[string]string{"CONDITION_SUCCEEDED": "succeeded", "CONDITION_FAILED": "failed", "CONDITION_RECONCILING": "running", "NEW_STATE": "unknown"} {
		entry := jobFixture(job, state)
		entry.Job = "pingcap-nexuscrm-load"
		run, err := crmExecutionRun(entry, job, task, "crm_load_job")
		if err != nil || run.Status != expected || run.Verification != "not_checked" || len(run.OutputRefs) != 0 || run.Manifest != nil {
			t.Fatal("execution state confused with produced data", state, err)
		}
	}
	entry := jobFixture(job, "CONDITION_SUCCEEDED")
	entry.FailedCount = 1
	if _, err := crmExecutionRun(entry, job, task, "crm_load_job"); err == nil {
		t.Fatal("conflicting completion accepted")
	}
	entry = jobFixture(job, "CONDITION_FAILED")
	entry.Conditions[0].ExecutionReason = "CANCELLED"
	if r, err := crmExecutionRun(entry, job, task, "crm_load_job"); err != nil || r.Status != "cancelled" {
		t.Fatal("cancelled classified as successful/failed")
	}
	entry.Name = "projects/other/locations/us-west2/jobs/other/executions/fixture"
	if _, err := crmExecutionRun(entry, job, task, "crm_load_job"); err == nil {
		t.Fatal("another job admitted")
	}
}

func TestCRMJobsReadOnlyScopedAndReplaySafe(t *testing.T) {
	job, _, _ := crmJobBinding("crm_extract")
	entry := jobFixture(job, "CONDITION_FAILED")
	entry.Conditions[0].Message = "private provider payload"
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method != "GET" || r.URL.Path != "/v2/"+job+"/executions" || r.URL.Query().Get("pageSize") != "100" {
			t.Errorf("unexpected operation: %s %s", r.Method, r.URL.Path)
		}
		fields := r.URL.Query().Get("fields")
		if fields == "" || strings.Contains(fields, "template") || strings.Contains(fields, "message") {
			t.Error("unnecessary secret-bearing fields requested")
		}
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(runapi.GoogleCloudRunV2ListExecutionsResponse{Executions: []*runapi.GoogleCloudRunV2Execution{entry}})
	}))
	defer server.Close()
	client, err := runapi.NewService(context.Background(), option.WithEndpoint(server.URL+"/"), option.WithoutAuthentication())
	if err != nil {
		t.Fatal(err)
	}
	s := &ExternalLiveSource{config: LiveSourceConfig{CRMControlDataset: "gcp-digital-bi.nexuscrm_control"}, runJobs: client}
	b, err := s.crmJobHistory(context.Background(), CollectionState{}, "crm_extract")
	if err != nil || len(b.Runs) != 1 {
		t.Fatal(err)
	}
	encoded, _ := json.Marshal(b)
	if strings.Contains(string(encoded), "private provider") {
		t.Fatal("provider message leaked")
	}
	b.State.Provider = "crm_extract"
	b.State.ObservedAt = time.Now().UTC()
	o := testOperations(t)
	for i := 0; i < 2; i++ {
		if err := o.CommitBatch(context.Background(), b); err != nil {
			t.Fatal(err)
		}
	}
	if countTable(t, o, "runs") != 1 || countTable(t, o, "alerts") != 1 {
		t.Fatal("poll duplicated run/alert")
	}
	s.config.CRMControlDataset = "other.control"
	if b, err := s.crmJobHistory(context.Background(), CollectionState{}, "crm_extract"); err != nil || b.State.Status != "unconfigured" {
		t.Fatal("job guessed for other CRM environment")
	}
}

func TestCRMJobAndTableAuditRemainSeparateInCatalog(t *testing.T) {
	o := testOperations(t)
	now := time.Now().UTC()
	b := testBatch("succeeded", now)
	b.Runs[0].Provider = "crm_load"
	b.Runs[0].Task = "crm_load"
	b.State.Provider = "crm_load"
	if err := o.CommitBatch(context.Background(), b); err != nil {
		t.Fatal(err)
	}
	b.Runs[0].Provider = "crm_load_job"
	b.Runs[0].Status = "failed"
	b.State.Provider = "crm_load_job"
	if err := o.CommitBatch(context.Background(), b); err != nil {
		t.Fatal(err)
	}
	items, err := o.TaskViews(context.Background(), now)
	if err != nil {
		t.Fatal(err)
	}
	for _, item := range items {
		if item.Key == "crm_load" {
			if item.LatestRun == nil || item.LatestRun.Status != "succeeded" || item.LatestJobRun == nil || item.LatestJobRun.Status != "failed" {
				t.Fatal("job state replaced table publication state")
			}
		}
	}
}