package server

import (
	"context"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/config"
	"github.com/tidbcloud/AgentTeam/internal/meshxruntime"
	"github.com/tidbcloud/AgentTeam/internal/observability"
	"github.com/tidbcloud/AgentTeam/internal/router"
)

func TestProfileIsTheOnlyMeshXTopologyOwner(t *testing.T) {
	require.True(t, profileComposesMeshX(router.ProfileAll, true))
	require.True(t, profileComposesMeshX(router.ProfileWorkspace, true))
	require.False(t, profileComposesMeshX(router.ProfileCentral, true))
	require.False(t, profileComposesMeshX(router.ProfileAll, false))
}

type profileMeshXRuntimeProbe struct {
	startCalls   int
	groupID      string
	requireGroup bool
	global       bool
	hasScope     bool
	startErr     error
}

func (probe *profileMeshXRuntimeProbe) StartWorkers(ownership meshxruntime.WorkerOwnership) error {
	probe.startCalls++
	probe.groupID = ownership.WorkspaceGroupID
	probe.requireGroup = ownership.RequireGroup
	probe.global = ownership.StartGlobalWorkers
	probe.hasScope = ownership.WorkspaceScope != nil
	return probe.startErr
}
func (*profileMeshXRuntimeProbe) Ready(context.Context) error          { return nil }
func (*profileMeshXRuntimeProbe) QuiesceWorkers(context.Context) error { return nil }
func (*profileMeshXRuntimeProbe) Shutdown(context.Context) error       { return nil }

func TestProfilePassesMeshXWorkerOwnership(t *testing.T) {
	tests := []struct {
		name         string
		profile      router.Profile
		server       config.ServerConfig
		wantGroup    string
		wantRequired bool
		wantGlobal   bool
	}{
		{
			name:      "non-designated workspace owner runs group workers only",
			profile:   router.ProfileWorkspace,
			server:    config.ServerConfig{Profile: config.ServerProfileWorkspace, WorkspaceGroupID: "workspace-a"},
			wantGroup: "workspace-a", wantRequired: true, wantGlobal: false,
		},
		{
			name:      "designated workspace owner also runs global maintenance",
			profile:   router.ProfileWorkspace,
			server:    config.ServerConfig{Profile: config.ServerProfileWorkspace, WorkspaceGroupID: "workspace-default"},
			wantGroup: "workspace-default", wantRequired: true, wantGlobal: true,
		},
		{
			name:      "all profile runs global maintenance",
			profile:   router.ProfileAll,
			server:    config.ServerConfig{Profile: config.ServerProfileAll},
			wantGroup: "workspace-default", wantRequired: false, wantGlobal: true,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			probe := &profileMeshXRuntimeProbe{}
			srv := &Server{
				config: config.Config{
					Server: tt.server,
					HTTP:   config.HTTPConfig{Address: "127.0.0.1:0", ShutdownTimeout: time.Second},
				},
				profile:       tt.profile,
				embeddedMeshX: probe,
				meshXWorkspaceScope: func(context.Context) ([]string, error) {
					return []string{"wsp_1"}, nil
				},
				httpServer: &http.Server{Addr: "127.0.0.1:0"},
				logger:     slog.Default(),
			}
			ctx, cancel := context.WithCancel(t.Context())
			cancel()
			require.NoError(t, srv.Run(ctx))
			require.Equal(t, 1, probe.startCalls)
			require.Equal(t, tt.wantGroup, probe.groupID)
			require.Equal(t, tt.wantRequired, probe.requireGroup)
			require.Equal(t, tt.wantGlobal, probe.global)
			require.True(t, probe.hasScope)
		})
	}
}

func TestProductionTaskRoutesReceiveRawTaskBoard(t *testing.T) {
	t.Parallel()

	source, err := os.ReadFile("wire.go")
	require.NoError(t, err)

	wire := string(source)
	taskDepsStart := strings.Index(wire, "Task: task.Dependencies{")
	require.NotEqual(t, -1, taskDepsStart)
	taskDepsEnd := strings.Index(wire[taskDepsStart:], "\n\t\t\t},")
	require.NotEqual(t, -1, taskDepsEnd)
	taskDeps := wire[taskDepsStart : taskDepsStart+taskDepsEnd]
	compactTaskDeps := strings.Join(strings.Fields(taskDeps), " ")

	require.Contains(t, compactTaskDeps, "TaskBoard: taskStore,")
	require.NotContains(t, compactTaskDeps, "TaskBoardRaw:")
	require.Contains(t, compactTaskDeps, "Store: task.NewTaskParityStoreAdapter(stores.App, stores.PrivateAgents),")
}

func TestProductionChannelRoutesReceiveMessageFanout(t *testing.T) {
	t.Parallel()

	source, err := os.ReadFile("wire.go")
	require.NoError(t, err)

	wire := string(source)
	channelDepsStart := strings.Index(wire, "Channel: channel.Dependencies{")
	require.NotEqual(t, -1, channelDepsStart)
	channelDepsEnd := strings.Index(wire[channelDepsStart:], "\n\t\t\t},")
	require.NotEqual(t, -1, channelDepsEnd)
	channelDeps := wire[channelDepsStart : channelDepsStart+channelDepsEnd]

	require.Contains(t, channelDeps, "MessageFanout:             deliveryFanout,")
}

func TestProductionServerStartsFirstClassWorkflowRunWorker(t *testing.T) {
	t.Parallel()

	source, err := os.ReadFile("server.go")
	require.NoError(t, err)

	serverSource := string(source)
	compactServerSource := strings.Join(strings.Fields(serverSource), " ")
	require.Contains(t, compactServerSource, "workflowRunWorker *orchestrator.Worker")
	require.Contains(t, compactServerSource, "workflowRunWorker: side.workflowRunWorker,")
	require.Contains(t, serverSource, "srv.workflowRunWorker.Start(ctx)")
	require.Contains(t, serverSource, "if srv.runsWorkspaceBackgroundLoops()")
}

// TestProductionServerRunsRecoveryThroughWorkflowRunWorkerTick proves that the
// production lifecycle starts the automatic recovery runner only through the
// first-class Workflow Run worker tick — there is no separate recovery
// goroutine to leak or race on shutdown.
func TestProductionServerRunsRecoveryThroughWorkflowRunWorkerTick(t *testing.T) {
	t.Parallel()

	source, err := os.ReadFile("server.go")
	require.NoError(t, err)

	serverSource := string(source)
	require.NotContains(t, serverSource, "Recovery")
	require.NotContains(t, serverSource, "recovery")
	require.Equal(t, 1, strings.Count(serverSource, "srv.workflowRunWorker.Start(ctx)"))
}

func TestWorkflowAppAvatarIsNoLongerPackaged(t *testing.T) {
	t.Parallel()

	_, err := os.Stat("../../web/public/builtin-apps/workflow.svg")
	require.Error(t, err)
	require.True(t, os.IsNotExist(err))
}

func TestProductionServerStartsWorkflowInvocationScheduleWorker(t *testing.T) {
	t.Parallel()

	source, err := os.ReadFile("server.go")
	require.NoError(t, err)

	serverSource := string(source)
	compactServerSource := strings.Join(strings.Fields(serverSource), " ")
	require.Contains(t, compactServerSource, "workflowInvocationSchedules *invocation.ScheduleWorker")
	require.Contains(t, compactServerSource, "workflowInvocationSchedules: side.workflowInvocationSchedules,")
	require.Contains(t, serverSource, "srv.workflowInvocationSchedules.Start(ctx)")
}

func TestProductionServerStartsApprovalProjectionReconciler(t *testing.T) {
	t.Parallel()

	source, err := os.ReadFile("server.go")
	require.NoError(t, err)

	serverSource := string(source)
	compactServerSource := strings.Join(strings.Fields(serverSource), " ")
	require.Contains(t, compactServerSource, "approvalProjectionReconciler *approvalapp.ProjectionReconciler")
	require.Contains(t, compactServerSource, "approvalProjectionReconciler: side.approvalProjectionReconciler,")
	require.Contains(t, serverSource, "srv.approvalProjectionReconciler.Run(ctx, 0)")
}

func TestProductionServerComposesAndStartsAgentOperationWorker(t *testing.T) {
	t.Parallel()

	wireSource, err := os.ReadFile("wire.go")
	require.NoError(t, err)
	wire := string(wireSource)
	compactWire := strings.Join(strings.Fields(wire), " ")

	require.Contains(t, compactWire, "operationExecutor := webagent.NewOperationExecutor( operationAgentService, webagent.WithOperationAgentInventory(stores.App), webagent.WithOperationOwnerIdentityFinder(stores.Identity), )")
	require.Contains(t, compactWire, "operationWorker, err := agentops.NewOperationWorker( stores.AgentOps, operationExecutor, approvalapp.NewAgentManagementTerminalObserver( stores.Approval, approvalAppProjector, ), 0, agentLogger.With(slog.String(\"component\", \"operation_worker\")), agentops.WithOperationWorkspaceGroupID(schedulerWorkspaceGroupIDForConfig(cfg.Server)), )")
	require.Equal(t, 1, strings.Count(wire, "agentops.NewOperationWorker("))
	require.Contains(t, compactWire, "operationWorker: operationWorker,")

	serverSource, err := os.ReadFile("server.go")
	require.NoError(t, err)
	server := string(serverSource)
	require.Contains(t, strings.Join(strings.Fields(server), " "), "operationWorker: side.operationWorker,")
	backgroundBlock := strings.Index(server, "if srv.runsWorkspaceBackgroundLoops()")
	runLocation := strings.Index(server, "srv.operationWorker.Run(ctx)")
	require.NotEqual(t, -1, backgroundBlock)
	require.Greater(t, runLocation, backgroundBlock)
}

func TestProductionServerComposesAndStartsRoleManagementWorkerOnce(t *testing.T) {
	t.Parallel()

	source, err := os.ReadFile("server.go")
	require.NoError(t, err)

	server := string(source)
	compactServer := strings.Join(strings.Fields(server), " ")
	require.Equal(t, 1, strings.Count(
		compactServer, "roleManagementWorker *execution.Worker",
	))
	require.Equal(t, 1, strings.Count(
		compactServer, "roleManagementWorker: side.roleManagementWorker,",
	))
	require.Equal(t, 1, strings.Count(
		server, "srv.roleManagementWorker.Run(ctx)",
	))

	backgroundStart := strings.Index(server, "if srv.runsWorkspaceBackgroundLoops()")
	backgroundEndOffset := strings.Index(
		server[backgroundStart:],
		"\n\t}\n\n\terrCh := make(chan error, 1)",
	)
	require.Equal(
		t,
		1,
		strings.Count(server, "if srv.runsWorkspaceBackgroundLoops() {"),
	)
	require.NotContains(
		t,
		server,
		"if srv.runsWorkspaceBackgroundLoops() ||",
	)
	workerGuard := `if srv.roleManagementWorker != nil {
			logging.Go(srv.logger, func() { srv.roleManagementWorker.Run(ctx) })
		}`
	require.Equal(t, 1, strings.Count(server, workerGuard))
	runLocation := strings.Index(server, workerGuard)
	require.NotEqual(t, -1, backgroundStart)
	require.NotEqual(t, -1, backgroundEndOffset)
	require.Greater(t, runLocation, backgroundStart)
	require.Less(t, runLocation, backgroundStart+backgroundEndOffset)
}

func TestRouterProfileForConfig(t *testing.T) {
	t.Parallel()

	for _, tt := range []struct {
		name    string
		profile string
		want    router.Profile
	}{
		{name: "all", profile: config.ServerProfileAll, want: router.ProfileAll},
		{name: "central", profile: config.ServerProfileCentral, want: router.ProfileCentral},
		{name: "workspace", profile: config.ServerProfileWorkspace, want: router.ProfileWorkspace},
	} {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()

			got, err := routerProfileForConfig(tt.profile)
			require.NoError(t, err)
			require.Equal(t, tt.want, got)
		})
	}
}

func TestServerBackgroundLoopsSkipCentralProfile(t *testing.T) {
	t.Parallel()

	require.False(t, (&Server{profile: router.ProfileCentral}).runsWorkspaceBackgroundLoops())
	require.True(t, (&Server{profile: router.ProfileWorkspace}).runsWorkspaceBackgroundLoops())
	require.True(t, (&Server{profile: router.ProfileAll}).runsWorkspaceBackgroundLoops())
	require.True(t, (&Server{}).runsWorkspaceBackgroundLoops())
}

func TestOptionalEmbeddedMeshXRuntimeKeepsDisabledLifecycleNil(t *testing.T) {
	t.Parallel()

	require.Nil(t, optionalEmbeddedMeshXRuntime(nil))
}

type blockingEmbeddedMeshX struct {
	readyCalled    chan struct{}
	shutdownCalled bool
}

func (runtime *blockingEmbeddedMeshX) StartWorkers(meshxruntime.WorkerOwnership) error {
	return nil
}

func (runtime *blockingEmbeddedMeshX) Ready(ctx context.Context) error {
	close(runtime.readyCalled)
	<-ctx.Done()
	return ctx.Err()
}

func (runtime *blockingEmbeddedMeshX) QuiesceWorkers(context.Context) error { return nil }

func (runtime *blockingEmbeddedMeshX) Shutdown(context.Context) error {
	runtime.shutdownCalled = true
	return nil
}

func TestRunStartupSchemaTimeoutFailsBeforeListener(t *testing.T) {
	runtime := &blockingEmbeddedMeshX{readyCalled: make(chan struct{})}
	srv := &Server{
		config: config.Config{HTTP: config.HTTPConfig{
			Address:              "127.0.0.1:0",
			ShutdownTimeout:      time.Second,
			StartupSchemaTimeout: time.Millisecond,
		}},
		logger:        slog.New(slog.NewTextHandler(io.Discard, nil)),
		embeddedMeshX: runtime,
		httpServer:    &http.Server{Addr: "127.0.0.1:0"},
	}

	err := srv.Run(context.Background())
	require.ErrorIs(t, err, context.DeadlineExceeded)
	require.True(t, runtime.shutdownCalled)
	select {
	case <-runtime.readyCalled:
	default:
		t.Fatal("startup schema gate was not called")
	}
}

func TestLLMProxyTransporterReturnsNilWithoutProxy(t *testing.T) {
	t.Parallel()

	transporter, err := llmProxyTransporter(config.Config{})
	require.NoError(t, err)
	require.Nil(t, transporter)
}

func TestGracefulShutdownGivesObservabilityOwnTimeout(t *testing.T) {
	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
	traceRequests := make(chan []byte, 1)
	collector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(r.Body)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		traceRequests <- body
		w.WriteHeader(http.StatusOK)
	}))
	defer collector.Close()

	provider, err := observability.NewBase(
		config.ObservabilityConfig{
			MetricsAddress:   "",
			OTLPEndpoint:     collector.URL,
			TraceSampleRatio: 1,
		},
		logger,
		observability.WithEnvironment("test"),
	)
	require.NoError(t, err)
	_, span := provider.Tracer().Start(context.Background(), "shutdown-test-span")
	span.End()

	requestStarted := make(chan struct{})
	releaseRequest := make(chan struct{})
	httpServer := &http.Server{
		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			close(requestStarted)
			<-releaseRequest
			w.WriteHeader(http.StatusNoContent)
		}),
		ReadHeaderTimeout: 5 * time.Second,
	}
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	require.NoError(t, err)
	defer listener.Close()
	serverErrCh := make(chan error, 1)
	go func() {
		serverErrCh <- httpServer.Serve(listener)
	}()

	client := &http.Client{Timeout: time.Second}
	requestDone := make(chan error, 1)
	go func() {
		resp, err := client.Get("http://" + listener.Addr().String())
		if resp != nil {
			_, _ = io.Copy(io.Discard, resp.Body)
			_ = resp.Body.Close()
		}
		requestDone <- err
	}()
	<-requestStarted

	srv := &Server{
		config:        config.Config{HTTP: config.HTTPConfig{ShutdownTimeout: 100 * time.Millisecond}},
		logger:        logger,
		httpServer:    httpServer,
		observability: provider,
	}
	err = srv.gracefulShutdown()
	require.ErrorIs(t, err, context.DeadlineExceeded)
	select {
	case body := <-traceRequests:
		require.NotEmpty(t, body)
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for trace flush after HTTP shutdown timeout")
	}

	close(releaseRequest)
	_ = httpServer.Close()
	<-serverErrCh
	<-requestDone
}