package jobs

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"strings"
	"sync/atomic"
	"testing"

	"evobgp/internal/nodedispatch"
	"evobgp/internal/store"

	"github.com/google/uuid"
)

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

func (f deployRoundTrip) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }

func TestDeployApplyConfirmation(t *testing.T) {
	for _, tc := range []struct {
		name, response, confirmation, status string
		dispatch                             bool
		stale                                bool
		wantApplied                          bool
	}{
		{name: "dispatch disabled", confirmation: "unconfirmed", status: StatusSucceeded},
		{name: "bare success", dispatch: true, response: `{"ok":true}`, confirmation: "unconfirmed", status: StatusSucceeded},
		{name: "matching ACK", dispatch: true, response: "matching", confirmation: "agent_reported", status: StatusSucceeded, wantApplied: true},
		{name: "mismatched ACK", dispatch: true, response: `{"ok":true,"applied_revision_id":"other"}`, confirmation: "unconfirmed", status: StatusFailed},
		{name: "stale desired", dispatch: true, response: "matching", confirmation: "unconfirmed", status: StatusFailed, stale: true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", "")
			if tc.dispatch {
				// A remote-only job must not touch the control plane BIRD tree.
				t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", t.TempDir())
				t.Setenv("EVOBGP_NODE_DISPATCH_ENABLED", "1")
			} else {
				t.Setenv("EVOBGP_NODE_DISPATCH_ENABLED", "0")
			}
			m := store.NewMemory()
			m.SeedDemo()
			tenant, _, module, previous, speakerID := m.DemoIDs()
			if err := m.SetLastAppliedRevision(tenant, speakerID, previous); err != nil {
				t.Fatal(err)
			}
			revision := uuid.NewString()
			if err := m.CreateRenderRevision(context.Background(), revision, tenant, module, &previous, "new-hash", map[string]string{"bird.conf": "# new\n"}, nil); err != nil {
				t.Fatal(err)
			}
			newer := uuid.NewString()
			if err := m.CreateRenderRevision(context.Background(), newer, tenant, module, &revision, "newer-hash", map[string]string{"bird.conf": "# newer\n"}, nil); err != nil {
				t.Fatal(err)
			}
			meta := store.SpeakerMetaJSON(store.SpeakerMeta{AgentDomain: "agent.example.test", AgentSecret: "secret"})
			if _, err := m.UpdateSpeaker(tenant, speakerID, &store.SpeakerPatch{MetaJSON: &meta}); err != nil {
				t.Fatal(err)
			}
			client := &http.Client{Transport: deployRoundTrip(func(r *http.Request) (*http.Response, error) {
				if r.Method != http.MethodPost || r.URL.Path != "/v1/agent/sync" || r.Header.Get("Authorization") != "Bearer secret" {
					t.Errorf("unexpected sync request: %s %s", r.Method, r.URL)
				}
				var body map[string]string
				if err := json.NewDecoder(r.Body).Decode(&body); err != nil || len(body) != 1 || body["revision_id"] != revision {
					t.Errorf("legacy sync body: %#v, %v", body, err)
				}
				if tc.stale {
					if err := m.PublishRevisionForSpeaker(speakerID, newer); err != nil {
						t.Error(err)
					}
				}
				response := tc.response
				if response == "matching" {
					response = `{"ok":true,"applied_revision_id":"` + revision + `"}`
				}
				return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(response)), Header: make(http.Header)}, nil
			})}
			j := &Job{TenantID: tenant, Kind: KindDeployApply, Status: StatusQueued, Meta: map[string]any{"revision_id": revision, "speaker_id": speakerID}}
			(&Worker{Store: m, NodeHTTPClient: client}).Process(j)
			if j.Status != tc.status {
				t.Fatalf("job status = %s, want %s; error=%v", j.Status, tc.status, j.Error)
			}
			sp, err := m.GetSpeaker(tenant, speakerID)
			if err != nil {
				t.Fatal(err)
			}
			want := previous
			if tc.wantApplied {
				want = revision
			}
			if sp.LastAppliedRevisionID == nil || *sp.LastAppliedRevisionID != want {
				t.Fatalf("applied = %v, want %s", sp.LastAppliedRevisionID, want)
			}
			published, _, err := m.LatestPublishedRevision(speakerID)
			if err != nil {
				t.Fatal(err)
			}
			wantPublished := revision
			if tc.stale {
				wantPublished = newer
			}
			if published != wantPublished {
				t.Fatalf("published = %s, want %s", published, wantPublished)
			}
			results := j.Meta["node_dispatch"].(map[string]any)["results"].([]nodedispatch.Result)
			if len(results) != 1 || results[0].Confirmation != tc.confirmation {
				t.Fatalf("results = %+v", results)
			}
		})
	}
}

func TestDeployApplyPartialResult(t *testing.T) {
	t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", "")
	t.Setenv("EVOBGP_NODE_DISPATCH_ENABLED", "1")
	m := store.NewMemory()
	m.SeedDemo()
	tenant, _, module, previous, first := m.DemoIDs()
	revision := uuid.NewString()
	if err := m.CreateRenderRevision(context.Background(), revision, tenant, module, &previous, "new-hash", map[string]string{"bird.conf": "# new\n"}, nil); err != nil {
		t.Fatal(err)
	}
	goodMeta := store.SpeakerMetaJSON(store.SpeakerMeta{AgentDomain: "good.example.test", AgentSecret: "secret"})
	if _, err := m.UpdateSpeaker(tenant, first, &store.SpeakerPatch{MetaJSON: &goodMeta}); err != nil {
		t.Fatal(err)
	}
	badMeta := store.SpeakerMetaJSON(store.SpeakerMeta{AgentDomain: "bad.example.test", AgentSecret: "secret"})
	second, err := m.CreateSpeaker(tenant, &store.Speaker{Role: "replica", MetaJSON: badMeta})
	if err != nil {
		t.Fatal(err)
	}
	client := &http.Client{Transport: deployRoundTrip(func(r *http.Request) (*http.Response, error) {
		response := `{"ok":true,"applied_revision_id":"` + revision + `"}`
		if strings.HasPrefix(r.URL.Host, "bad.") {
			response = `{"ok":false,"error":"apply failed"}`
		}
		return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(response)), Header: make(http.Header)}, nil
	})}
	j := &Job{TenantID: tenant, Kind: KindDeployApply, Status: StatusQueued, Meta: map[string]any{"revision_id": revision}}
	(&Worker{Store: m, NodeHTTPClient: client}).Process(j)
	if j.Status != StatusFailed {
		t.Fatalf("job status = %s, want failed", j.Status)
	}
	summary := j.Meta["apply_summary"].(map[string]any)
	if summary["applied_count"] != 1 || summary["failed_count"] != 1 || summary["published_count"] != 2 {
		t.Fatalf("partial summary = %#v", summary)
	}
	firstSp, _ := m.GetSpeaker(tenant, first)
	secondSp, _ := m.GetSpeaker(tenant, second.ID)
	if firstSp.LastAppliedRevisionID == nil || *firstSp.LastAppliedRevisionID != revision || secondSp.LastAppliedRevisionID != nil {
		t.Fatalf("partial applied state: first=%v second=%v", firstSp.LastAppliedRevisionID, secondSp.LastAppliedRevisionID)
	}
	results := j.Meta["node_dispatch"].(map[string]any)["results"].([]nodedispatch.Result)
	got := map[string]bool{}
	for _, res := range results {
		got[res.SpeakerID] = true
	}
	if len(results) != 2 || !got[first] || !got[second.ID] {
		t.Fatalf("partial results=%+v", results)
	}
}

func TestDeployApplyCancellationStopsFurtherPublication(t *testing.T) {
	t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", "")
	t.Setenv("EVOBGP_NODE_DISPATCH_ENABLED", "1")
	m := store.NewMemory()
	m.SeedDemo()
	tenant, _, module, previous, seeded := m.DemoIDs()
	if err := m.DeleteSpeaker(tenant, seeded); err != nil {
		t.Fatal(err)
	}
	revision := uuid.NewString()
	if err := m.CreateRenderRevision(context.Background(), revision, tenant, module, &previous, "new-hash", map[string]string{"bird.conf": "# new\n"}, nil); err != nil {
		t.Fatal(err)
	}
	meta := store.SpeakerMetaJSON(store.SpeakerMeta{AgentDomain: "agent.example.test", AgentSecret: "secret"})
	speakerCount := speakerWakeLimit + 2
	for range speakerCount {
		if _, err := m.CreateSpeaker(tenant, &store.Speaker{Role: "replica", MetaJSON: meta}); err != nil {
			t.Fatal(err)
		}
	}
	speakers := m.ListSpeakersForTenant(tenant)
	if len(speakers) != speakerCount {
		t.Fatalf("speakers=%d, want %d", len(speakers), speakerCount)
	}
	var j *Job
	var calls atomic.Int32
	client := &http.Client{Transport: deployRoundTrip(func(r *http.Request) (*http.Response, error) {
		// Cancel only once the in-flight wake limit is busy, so later speakers stay unpublished.
		if calls.Add(1) == int32(speakerWakeLimit) {
			j.RequestCancel()
		}
		<-r.Context().Done()
		return nil, r.Context().Err()
	})}
	j = &Job{TenantID: tenant, Kind: KindDeployApply, Status: StatusQueued, Meta: map[string]any{"revision_id": revision}}
	(&Worker{Store: m, NodeHTTPClient: client}).Process(j)
	if j.Status != StatusCancelled || int(calls.Load()) != speakerWakeLimit {
		t.Fatalf("cancelled job status=%s dispatch calls=%d", j.Status, calls.Load())
	}
	results := j.Meta["node_dispatch"].(map[string]any)["results"].([]nodedispatch.Result)
	gotResult := map[string]bool{}
	for _, res := range results {
		gotResult[res.SpeakerID] = res.Status == "cancelled"
	}
	if len(results) != speakerWakeLimit {
		t.Fatalf("cancelled results=%+v", results)
	}
	for i, sp := range speakers {
		got, _, err := m.LatestPublishedRevision(sp.ID)
		if i < speakerWakeLimit {
			if err != nil || got != revision || !gotResult[sp.ID] {
				t.Fatalf("speaker %s published=%s err=%v result=%v", sp.ID, got, err, gotResult[sp.ID])
			}
			if sp.LastAppliedRevisionID != nil {
				t.Fatalf("cancelled speaker %s marked applied", sp.ID)
			}
			continue
		}
		if err == nil {
			t.Fatalf("speaker %s published %s after cancellation", sp.ID, got)
		}
	}
	if j.Meta["apply_summary"].(map[string]any)["published_count"] != speakerWakeLimit {
		t.Fatalf("cancelled summary = %#v", j.Meta["apply_summary"])
	}
}