package meshxassets

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

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/api/daemon"
)

type fakeNotifier struct {
	mu    sync.Mutex
	err   error
	calls []notifyCall
}

type notifyCall struct {
	WorkspaceID string
	AgentID     string
	Event       daemon.AssetsAccessChangedEvent
}

func (n *fakeNotifier) NotifyAssetsAccessChangedByAgent(_ context.Context, workspaceID string, agentID string, event daemon.AssetsAccessChangedEvent) error {
	n.mu.Lock()
	defer n.mu.Unlock()
	n.calls = append(n.calls, notifyCall{WorkspaceID: workspaceID, AgentID: agentID, Event: event})
	return n.err
}

func (n *fakeNotifier) snapshot() []notifyCall {
	n.mu.Lock()
	defer n.mu.Unlock()
	return append([]notifyCall(nil), n.calls...)
}

func TestWorkerRunOnceDeliversAndAcksMeshXEvents(t *testing.T) {
	now := time.Date(2026, 6, 22, 1, 0, 0, 0, time.UTC)
	notifier := &fakeNotifier{}
	var leaseReq leaseRequest
	var deliveredReq deliveredRequest
	var sawServiceClaims bool

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.NotEmpty(t, r.Header.Get("Authorization"))
		require.NotEmpty(t, r.Header.Get("X-AgentTeam-Verified-JWT"))
		require.Equal(t, r.Header.Get("Authorization"), r.Header.Get("X-AgentTeam-Verified-JWT"))
		claims := decodeBearerClaims(t, r.Header.Get("X-AgentTeam-Verified-JWT"))
		require.Equal(t, true, claims["service_actor"])
		require.Contains(t, claims["automation_permissions"], "automation:service")
		sawServiceClaims = true

		switch r.URL.Path {
		case "/api/v1/internal/agent-assets/access-events:lease":
			require.NoError(t, json.NewDecoder(r.Body).Decode(&leaseReq))
			require.Equal(t, []string{"wsp_1", "wsp_2"}, leaseReq.WorkspaceIDs)
			writeJSON(t, w, leaseResponse{Events: []accessEvent{{
				EventID:     "evt_1",
				WorkspaceID: "wsp_1",
				AgentIDs:    []string{"agt_1", "agt_2"},
				AssetTypes:  []string{"skill"},
				Version:     42,
				OccurredAt:  now.Format(time.RFC3339Nano),
				Source:      "outbox",
			}}})
		case "/api/v1/internal/agent-assets/access-events/evt_1/delivered":
			require.NoError(t, json.NewDecoder(r.Body).Decode(&deliveredReq))
			writeJSON(t, w, ackResponse{Updated: true})
		default:
			http.NotFound(w, r)
		}
	}))
	defer server.Close()

	worker, err := NewWorker(Options{
		BaseURL:  server.URL,
		Notifier: notifier,
		WorkspaceIDs: func(context.Context) ([]string, error) {
			return []string{" wsp_1 ", "", "wsp_2", "wsp_1"}, nil
		},
		Now: func() time.Time { return now },
	})
	require.NoError(t, err)

	result, err := worker.RunOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, RunResult{Leased: 1, Delivered: 1}, result)
	require.True(t, sawServiceClaims)
	require.Equal(t, "agentteam-asset-access-worker", leaseReq.LeaseOwner)
	require.Equal(t, int64(defaultLeaseDuration/time.Second), leaseReq.LeaseSeconds)
	require.Equal(t, defaultLimit, leaseReq.Limit)
	require.Equal(t, "agentteam-asset-access-worker", deliveredReq.LeaseOwner)

	calls := notifier.snapshot()
	require.Len(t, calls, 2)
	require.Equal(t, "wsp_1", calls[0].WorkspaceID)
	require.Equal(t, "agt_1", calls[0].AgentID)
	require.Equal(t, "evt_1", calls[0].Event.EventID)
	require.Equal(t, []string{"agt_1", "agt_2"}, calls[0].Event.AgentIDs)
	require.Equal(t, []string{"skill"}, calls[0].Event.AssetTypes)
	require.Equal(t, int64(42), calls[0].Event.Version)
	require.Equal(t, now, calls[0].Event.OccurredAt)
}

func TestWorkerRunOnceMarksEventFailedWhenDaemonNotifyFails(t *testing.T) {
	now := time.Date(2026, 6, 22, 2, 0, 0, 0, time.UTC)
	notifier := &fakeNotifier{err: errors.New("machine offline")}
	var failedReq failedRequest

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/v1/internal/agent-assets/access-events:lease":
			writeJSON(t, w, leaseResponse{Events: []accessEvent{{
				EventID:     "evt_retry",
				WorkspaceID: "wsp_1",
				AgentIDs:    []string{"agt_1"},
				AssetTypes:  []string{"memory", "skill"},
				OccurredAt:  now.Format(time.RFC3339Nano),
				Attempts:    1,
			}}})
		case "/api/v1/internal/agent-assets/access-events/evt_retry/failed":
			require.NoError(t, json.NewDecoder(r.Body).Decode(&failedReq))
			writeJSON(t, w, ackResponse{Updated: true})
		default:
			http.NotFound(w, r)
		}
	}))
	defer server.Close()

	worker, err := NewWorker(Options{
		BaseURL:  server.URL,
		Notifier: notifier,
		Now:      func() time.Time { return now },
	})
	require.NoError(t, err)

	result, err := worker.RunOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, RunResult{Leased: 1, Retried: 1}, result)
	require.Equal(t, "agentteam-asset-access-worker", failedReq.LeaseOwner)
	require.Contains(t, failedReq.LastError, "machine offline")
	nextAttempt, err := time.Parse(time.RFC3339Nano, failedReq.NextAttemptAt)
	require.NoError(t, err)
	require.True(t, nextAttempt.After(now))
}

func writeJSON(t *testing.T, w http.ResponseWriter, payload any) {
	t.Helper()
	w.Header().Set("Content-Type", "application/json")
	require.NoError(t, json.NewEncoder(w).Encode(payload))
}

func decodeBearerClaims(t *testing.T, header string) map[string]any {
	t.Helper()
	fields := strings.Fields(header)
	require.Len(t, fields, 2)
	require.Equal(t, "Bearer", fields[0])
	parts := strings.Split(fields[1], ".")
	require.GreaterOrEqual(t, len(parts), 2)
	payload, err := base64.RawURLEncoding.DecodeString(parts[1])
	require.NoError(t, err)
	var claims map[string]any
	require.NoError(t, json.Unmarshal(payload, &claims))
	return claims
}