package testkit

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/api/auth"
	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	"github.com/tidbcloud/AgentTeam/internal/api/im/fanout"
	"github.com/tidbcloud/AgentTeam/internal/archive"
	"github.com/tidbcloud/AgentTeam/internal/core/secrets"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/coredb/testkit"
	"github.com/tidbcloud/AgentTeam/internal/mailer"
	"github.com/tidbcloud/AgentTeam/internal/runtimecontrol"
	"github.com/tidbcloud/AgentTeam/internal/schedule"
	"github.com/tidbcloud/AgentTeam/internal/service/agentruntime"
	"github.com/tidbcloud/AgentTeam/internal/service/serviceaccount"
	"github.com/tidbcloud/AgentTeam/internal/sms"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/organization"
	schedulestore "github.com/tidbcloud/AgentTeam/internal/store/schedule"
	"github.com/tidbcloud/AgentTeam/internal/store/taskboard"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/binding"
	calendarstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/calendar"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/invocation"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/plan"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/trigger"
	"github.com/tidbcloud/AgentTeam/internal/versioning"
	"github.com/tidbcloud/AgentTeam/internal/workflow/bindingmanage"
	"github.com/tidbcloud/AgentTeam/internal/workflow/calendar"
	"github.com/tidbcloud/AgentTeam/internal/workflow/invocationmanage"
	"github.com/tidbcloud/AgentTeam/internal/workflow/planmanage"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runmanage"
	"github.com/tidbcloud/AgentTeam/internal/workspace/owner"
)

func WorkspaceScopeMessagesPath(workspaceID string, scopeID string) string {
	return fmt.Sprintf(
		"/api/v1/workspaces/%s/scopes/%s/messages",
		url.PathEscape(workspaceID),
		url.PathEscape(scopeID),
	)
}

func WorkspaceScopeMessagesPathWithLimit(workspaceID string, scopeID string, limit int) string {
	return fmt.Sprintf("%s?limit=%d", WorkspaceScopeMessagesPath(workspaceID, scopeID), limit)
}

func WorkspaceMessageSearchPath(workspaceID string, rawQuery string) string {
	path := fmt.Sprintf("/api/v1/workspaces/%s/messages/search", url.PathEscape(workspaceID))
	if strings.TrimSpace(rawQuery) == "" {
		return path
	}
	return path + "?" + strings.TrimPrefix(rawQuery, "?")
}

func WorkspaceMessageDeliveryPath(workspaceID string, messageID string) string {
	return fmt.Sprintf(
		"/api/v1/workspaces/%s/messages/%s/delivery",
		url.PathEscape(workspaceID),
		url.PathEscape(messageID),
	)
}

func WorkspaceMessageRewindPath(workspaceID string, messageID string) string {
	return fmt.Sprintf(
		"/api/v1/workspaces/%s/messages/%s/rewind",
		url.PathEscape(workspaceID),
		url.PathEscape(messageID),
	)
}

func WorkspaceMessageReactionsPath(workspaceID string, scopeID string, messageID string) string {
	return fmt.Sprintf(
		"/api/v1/workspaces/%s/scopes/%s/messages/%s/reactions",
		url.PathEscape(workspaceID),
		url.PathEscape(scopeID),
		url.PathEscape(messageID),
	)
}

func WorkspaceMessageReactionPath(workspaceID string, scopeID string, messageID string, emoji string) string {
	return fmt.Sprintf(
		"%s/%s",
		WorkspaceMessageReactionsPath(workspaceID, scopeID, messageID),
		url.PathEscape(emoji),
	)
}

const (
	defaultAPITestRuntimeParallelism = 2
	apiTestRuntimeParallelismEnv     = "AGENTTEAM_API_TEST_RUNTIME_PARALLELISM"
)

var apiTestRuntimeSlots = make(chan struct{}, apiTestRuntimeParallelism())

func apiTestRuntimeParallelism() int {
	rawValue := strings.TrimSpace(os.Getenv(apiTestRuntimeParallelismEnv))
	if rawValue == "" {
		return defaultAPITestRuntimeParallelism
	}
	value, err := strconv.Atoi(rawValue)
	if err != nil || value < 1 {
		return defaultAPITestRuntimeParallelism
	}
	return value
}

func acquireAPITestRuntimeSlot(t *testing.T) {
	t.Helper()

	apiTestRuntimeSlots <- struct{}{}
	t.Cleanup(func() {
		<-apiTestRuntimeSlots
	})
}

func WorkspaceAllChannelID(t *testing.T, runtime Runtime, workspaceID string) string {
	t.Helper()

	var channelID string
	require.NoError(t, runtime.Pool.QueryRow(runtime.Ctx, `
SELECT scope_id
FROM scope_channel_profiles
WHERE workspace_id = $1
  AND is_all_channel = TRUE
LIMIT 1
`, workspaceID).Scan(&channelID))
	require.NotEmpty(t, channelID)
	return channelID
}

func SetupTaskHTTPActor(t *testing.T, client *http.Client, baseURL string, email string, name string) (dto.AuthTokensResponse, string, string) {
	t.Helper()

	var auth dto.AuthTokensResponse
	response := DoJSON(t, client, http.MethodPost, baseURL+"/api/v1/auth/register", map[string]any{
		"email":    email,
		"password": TestUserPassword,
		"name":     name,
	}, nil, &auth)
	require.Equal(t, http.StatusOK, response.StatusCode)

	var workspaces []dto.WorkspaceResponse
	response = DoJSON(t, client, http.MethodGet, baseURL+"/api/v1/workspaces", nil, map[string]string{
		"Authorization": fmt.Sprintf("Bearer %s", auth.AccessToken),
	}, &workspaces)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.NotEmpty(t, workspaces)
	workspaceID := workspaces[0].ID

	var channels []ChannelResponse
	response = DoJSON(t, client, http.MethodGet, baseURL+fmt.Sprintf("/api/v1/workspaces/%s/channels", workspaceID), nil, map[string]string{
		"Authorization":  fmt.Sprintf("Bearer %s", auth.AccessToken),
		"X-Workspace-Id": workspaceID,
	}, &channels)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.NotEmpty(t, channels)

	return auth, workspaceID, channels[0].ID
}

func AcceptTaskHTTPInvite(t *testing.T, runtime Runtime, client *http.Client, baseURL string, inviterUserID string, workspaceID string, email string, name string) dto.AuthTokensResponse {
	t.Helper()

	inviteToken := "tok_" + strings.ReplaceAll(name, "-", "_")
	_, err := runtime.Store.CreateWorkspaceInvite(
		runtime.Ctx,
		inviterUserID,
		workspaceID,
		email,
		secrets.HashToken(inviteToken),
		time.Now().UTC().Add(10*time.Minute),
	)
	require.NoError(t, err)

	var auth dto.AuthTokensResponse
	response := DoJSON(t, client, http.MethodPost, baseURL+"/api/v1/auth/register", map[string]any{
		"email":    email,
		"password": TestUserPassword,
		"name":     name,
	}, nil, &auth)
	require.Equal(t, http.StatusOK, response.StatusCode)

	response = DoJSON(t, client, http.MethodPost, baseURL+"/api/v1/auth/accept-invite", map[string]any{
		"token": inviteToken,
	}, map[string]string{
		"Authorization": fmt.Sprintf("Bearer %s", auth.AccessToken),
	}, nil)
	require.Equal(t, http.StatusOK, response.StatusCode)
	return auth
}

func TaskWorkspaceMemberIDForUser(t *testing.T, runtime Runtime, workspaceID string, userID string) string {
	t.Helper()

	var memberID string
	require.NoError(t, runtime.Pool.QueryRow(runtime.Ctx, `
SELECT id
FROM members
WHERE workspace_id = $1
  AND user_id = $2
LIMIT 1
`, workspaceID, userID).Scan(&memberID))
	require.NotEmpty(t, memberID)
	return memberID
}

func StartRuntime(t *testing.T) Runtime {
	t.Helper()

	return StartRuntimeWithEmailVerificationRequired(t, false)
}

// StartRuntimeWithGitHubEndpoints constructs both the App store and provider
// routes with the same GitHub web/API endpoints. Tests must choose endpoints
// before store construction because the App store performs exact installation
// verification during proposal promotion and approval; mutating only route
// dependencies afterward would split one flow across two providers.
func StartRuntimeWithGitHubEndpoints(t *testing.T, baseURL string, apiBaseURL string) Runtime {
	t.Helper()
	return startRuntimeWithOrganizationProviderAndWorkers(
		t,
		false,
		30,
		func(logger *slog.Logger) mailer.Mailer {
			return mailer.NewNoop(logger, "http://localhost:3001")
		},
		func(pool coredb.DB, dataStore *Store) OrganizationProvider {
			return organization.NewWithDBOptions(pool, NewOrganizationCoreBridge(pool, dataStore), organization.ServiceOptions{})
		},
		true,
		StoreOptions{GitHubAPIBaseURL: apiBaseURL},
		baseURL,
		apiBaseURL,
	)
}

func StartRuntimeWithEmailVerificationRequired(t *testing.T, emailVerificationRequired bool) Runtime {
	t.Helper()

	return StartRuntimeWithMailer(t, emailVerificationRequired, 30, func(logger *slog.Logger) mailer.Mailer {
		return mailer.NewNoop(logger, "http://localhost:3001")
	})
}

func StartRuntimeWithHotMessageRetentionDays(t *testing.T, hotMessageRetentionDays int) Runtime {
	t.Helper()

	return StartRuntimeWithMailer(t, false, hotMessageRetentionDays, func(logger *slog.Logger) mailer.Mailer {
		return mailer.NewNoop(logger, "http://localhost:3001")
	})
}

type TestControlledMailer struct {
	VerificationFailOnCall  int
	VerificationCallCount   int
	VerificationErr         error
	WorkspaceInviteErr      error
	OrganizationInviteErr   error
	OrganizationInviteEmail string
	OrganizationInviteName  string
	OrganizationInviteToken string
}

type TestMobileVerificationSender struct {
	LastMobile string
	LastCode   string
	SendErr    error
}

func (sender *TestMobileVerificationSender) Send(_ context.Context, mobile string, code string) (sms.Receipt, error) {
	sender.LastMobile = mobile
	sender.LastCode = code
	return sms.Receipt{Provider: sms.ProviderNoop}, sender.SendErr
}

func (mailer *TestControlledMailer) SendVerificationEmail(_ context.Context, _ string, _ string) error {
	mailer.VerificationCallCount++
	if mailer.VerificationFailOnCall > 0 && mailer.VerificationCallCount == mailer.VerificationFailOnCall {
		if mailer.VerificationErr != nil {
			return mailer.VerificationErr
		}
		return errors.New("verification email failed")
	}
	return nil
}

func (mailer *TestControlledMailer) SendPasswordResetEmail(_ context.Context, _ string, _ string) error {
	return nil
}

func (mailer *TestControlledMailer) SendWorkspaceInviteEmail(_ context.Context, _ string, _ string, _ string) error {
	return mailer.WorkspaceInviteErr
}

func (mailer *TestControlledMailer) SendOrganizationInviteEmail(_ context.Context, email string, organizationName string, token string) error {
	mailer.OrganizationInviteEmail = email
	mailer.OrganizationInviteName = organizationName
	mailer.OrganizationInviteToken = token
	return mailer.OrganizationInviteErr
}

func StartRuntimeWithMailer(t *testing.T, emailVerificationRequired bool, hotMessageRetentionDays int, makeMailer func(logger *slog.Logger) mailer.Mailer) Runtime {
	t.Helper()

	return StartRuntimeWithOrganizationProvider(t, emailVerificationRequired, hotMessageRetentionDays, makeMailer, func(pool coredb.DB, dataStore *Store) OrganizationProvider {
		return organization.NewWithDBOptions(pool, NewOrganizationCoreBridge(pool, dataStore), organization.ServiceOptions{})
	})
}

func StartRuntimeWithoutBackgroundWorkers(t *testing.T) Runtime {
	t.Helper()

	return StartRuntimeWithOrganizationProviderAndWorkers(t, false, 30, func(logger *slog.Logger) mailer.Mailer {
		return mailer.NewNoop(logger, "http://localhost:3001")
	}, func(pool coredb.DB, dataStore *Store) OrganizationProvider {
		return organization.NewWithDBOptions(pool, NewOrganizationCoreBridge(pool, dataStore), organization.ServiceOptions{})
	}, false)
}

func StartRuntimeWithOrganizationProvider(
	t *testing.T,
	emailVerificationRequired bool,
	hotMessageRetentionDays int,
	makeMailer func(logger *slog.Logger) mailer.Mailer,
	makeOrganizationProvider func(pool coredb.DB, dataStore *Store) OrganizationProvider,
) Runtime {
	return StartRuntimeWithOrganizationProviderAndWorkers(
		t,
		emailVerificationRequired,
		hotMessageRetentionDays,
		makeMailer,
		makeOrganizationProvider,
		true,
	)
}

func StartRuntimeWithOrganizationProviderAndWorkers(
	t *testing.T,
	emailVerificationRequired bool,
	hotMessageRetentionDays int,
	makeMailer func(logger *slog.Logger) mailer.Mailer,
	makeOrganizationProvider func(pool coredb.DB, dataStore *Store) OrganizationProvider,
	startWorkers bool,
) Runtime {
	return startRuntimeWithOrganizationProviderAndWorkers(
		t,
		emailVerificationRequired,
		hotMessageRetentionDays,
		makeMailer,
		makeOrganizationProvider,
		startWorkers,
		StoreOptions{},
		"https://github.com",
		"https://api.github.com",
	)
}

func startRuntimeWithOrganizationProviderAndWorkers(
	t *testing.T,
	emailVerificationRequired bool,
	hotMessageRetentionDays int,
	makeMailer func(logger *slog.Logger) mailer.Mailer,
	makeOrganizationProvider func(pool coredb.DB, dataStore *Store) OrganizationProvider,
	startWorkers bool,
	storeOptions StoreOptions,
	githubBaseURL string,
	githubAPIBaseURL string,
) Runtime {
	t.Helper()

	acquireAPITestRuntimeSlot(t)

	gin.SetMode(gin.TestMode)

	Ctx := context.Background()
	databaseDSN := testkit.StartInitializedTiDBWithSharedRuntime(t, testkit.TiDBTestRuntimeUnistore)
	sqlDB := testkit.OpenTiDB(t, databaseDSN)
	// Keep connection footprint small across the integration suite to avoid
	// exhausting the shared TiDB runtime.
	sqlDB.SetMaxOpenConns(4)
	sqlDB.SetMaxIdleConns(4)
	Pool := coredb.WrapSQLDB(sqlDB)

	archiveStorage, err := archive.NewFilesystem(t.TempDir())
	require.NoError(t, err)

	Logger := slog.New(slog.NewTextHandler(io.Discard, nil))
	storeOptions.AppCredentialStorageKey = []byte("0123456789abcdef0123456789abcdef")
	storeOptions.PublicBaseURL = "http://localhost:3001"
	dataStore := NewWithOptions(Pool, archiveStorage, hotMessageRetentionDays, storeOptions)
	organizationService := makeOrganizationProvider(Pool, dataStore)
	unreadRuntime, err := newAPITestUnreadRuntime(dataStore, Logger)
	require.NoError(t, err)
	taskUnreadBridge := newAPITestTaskUnreadBridge(unreadRuntime)
	taskStore := taskboard.NewWithOptions(Pool, taskboard.StoreOptions{TaskChangeSink: taskUnreadBridge})
	agentRuntime := agentruntime.New(agentruntime.Dependencies{
		Agents:           dataStore.Agent(),
		Messages:         dataStore.Message(),
		Recovery:         dataStore.Rewind(),
		WebAgents:        dataStore.App(),
		WebRecovery:      dataStore.Rewind(),
		IsRuntimeOffline: func(err error) bool { return errors.Is(err, runtimecontrol.ErrMachineOffline) },
	})
	runtime := Runtime{
		Ctx:                       Ctx,
		Pool:                      Pool,
		ArchiveStorage:            archiveStorage,
		Logger:                    Logger,
		Store:                     dataStore,
		Organizations:             organizationService,
		TaskBoard:                 taskStore,
		Mailer:                    makeMailer(Logger),
		MobileVerificationSender:  sms.NewNoop(Logger),
		AgentRuntime:              agentRuntime,
		UnreadRuntime:             unreadRuntime,
		HTTPClient:                &http.Client{Timeout: 10 * time.Second},
		EmailVerificationRequired: emailVerificationRequired,
		RegistrationPolicy:        auth.NewRegistrationPolicy(auth.RegistrationModeOpen, nil),
		VerificationTokenTTL:      24 * time.Hour,
		SessionTTL:                30 * 24 * time.Hour,
		HotMessageRetentionDays:   hotMessageRetentionDays,
		PublicBaseURL:             "http://localhost:3001",
		GitHubFlowStateKey:        []byte("0123456789abcdef0123456789abcdef"),
		GitHubBaseURL:             githubBaseURL,
		GitHubAPIBaseURL:          githubAPIBaseURL,
		DaemonConnectBaseURL:      "http://localhost:3001",
		LatestDaemonVersion:       versioning.NewStaticProvider(versioning.ServerVersion()),
		PingCAPAuthConfig: auth.PingCAPAuthConfig{
			Scopes: []string{"openid", "profile", "email", "phone"},
		},
	}
	dataStore.App().SetCommittedMessageFanout(runtimeDeliveryFanout(&runtime))
	serviceAccountDeletionObserver := newAPITestServiceAccountDeletionObserver(unreadRuntime)
	runtime.ServiceAccount = serviceaccount.New(Pool)
	runtime.ServiceAccount.SetLocalAccountDeletionObserver(serviceAccountDeletionObserver)
	runtime.PlatformServiceAccount = serviceaccount.NewPlatformService(Pool, serviceaccount.PlatformServiceOptions{
		ControlPlane: true, ManagedWorkspaces: owner.NewAllSet(), MessageWriter: dataStore.App(),
		LocalAccountDeletionObserver: serviceAccountDeletionObserver,
	})
	runtime.ApprovalAppProjector = dataStore.ApprovalAppProjector()
	require.NotNil(t, runtime.ApprovalAppProjector)
	runtime.SetTriggerService(trigger.New(Logger, Pool, dataStore.App(), trigger.ServiceOptions{}))
	workflowPlanStore := plan.New(Pool)
	workflowPlanService := planmanage.New(workflowPlanStore)
	runtime.WorkflowPlan = workflowPlanService
	runtime.WorkflowPlanAgentEdit = planmanage.NewAgentEditService(
		Pool,
		workflowPlanStore,
		dataStore.Message(),
		dataStore.App(),
		dataStore.AgentOps(),
	)
	workflowBindingStore := binding.New(Pool)
	runtime.WorkflowBinding = bindingmanage.New(
		workflowBindingStore,
		workflowPlanService,
		bindingmanage.WithAgentOwnerVerifier(dataStore.App()),
	)
	workflowInvocationStore := invocation.New(Pool)
	runtime.WorkflowInvocation = invocationmanage.New(
		workflowInvocationStore,
		invocationmanage.WithInputAuthority(dataStore.App()),
	)
	workflowRunStore := run.New(Pool)
	runtime.WorkflowRun = runmanage.New(workflowRunStore)
	runtime.WorkflowCalendar = calendar.New(calendarstore.New(Pool))
	scheduleStore := schedulestore.New(Pool)
	imRoutes := imRouteDependencies(&runtime)
	runtime.SetScheduleService(schedule.New(Logger, scheduleStore, schedule.ServiceOptions{
		Writer: fanout.NewScheduleEffectWriter(scheduleStore, fanout.Dependencies{
			HumanMessages:            imRoutes.HumanMessages,
			Hub:                      imRoutes.AgentNotifier,
			StreamWriter:             imRoutes.StreamWriter,
			ReplyThreadRootPreloader: imRoutes.ReplyThreadRootPreloader,
			ScopeFanoutMarker:        imRoutes.ScopeFanoutMarker,
			TaskScopeLifecycle:       dataStore.Message(),
			UnreadRuntime:            unreadRuntime,
			UnreadDeltaSink:          runtime.LiveEmitter,
			Logger:                   Logger,
		}),
	}))
	if startWorkers {
		workerCtx, stopWorkers := context.WithCancel(context.Background())
		unreadRuntime.Start(workerCtx)
		runtime.StartBackgroundWorkers(workerCtx)
		t.Cleanup(stopWorkers)
	}

	return runtime
}

func DoJSON(t *testing.T, client *http.Client, method string, url string, payload any, headers map[string]string, out any) *http.Response {
	t.Helper()

	var body io.Reader
	if payload != nil {
		encoded, err := json.Marshal(payload)
		require.NoError(t, err)
		body = bytes.NewReader(encoded)
	}

	request, err := http.NewRequestWithContext(context.Background(), method, url, body)
	require.NoError(t, err)
	if payload != nil {
		request.Header.Set("Content-Type", "application/json")
	}
	for key, value := range headers {
		request.Header.Set(key, value)
	}

	response, err := client.Do(request)
	require.NoError(t, err)
	t.Cleanup(func() {
		_ = response.Body.Close()
	})

	if out == nil {
		return response
	}

	data, err := io.ReadAll(response.Body)
	require.NoError(t, err)
	require.NoError(t, json.Unmarshal(data, out), "response body: %s", string(data))
	return response
}

func IsTimeout(err error) bool {
	var netErr net.Error
	return errors.As(err, &netErr) && netErr.Timeout()
}

func CollectAppMessageIDs(messages []message.AppMessage) []string {
	ids := make([]string, 0, len(messages))
	for _, message := range messages {
		ids = append(ids, message.ID)
	}
	return ids
}

func TaskFieldOptionKeys(options []taskboard.TaskFieldOption) []string {
	keys := make([]string, 0, len(options))
	for _, option := range options {
		keys = append(keys, option.Key)
	}
	return keys
}

func FindTaskFieldOptionByKey(t *testing.T, options []taskboard.TaskFieldOption, key string) taskboard.TaskFieldOption {
	t.Helper()

	for _, option := range options {
		if option.Key == key {
			return option
		}
	}
	t.Fatalf("missing task field option %q", key)
	return taskboard.TaskFieldOption{}
}