package automation

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	agsconfig "github.com/pingcap/agent-git-service/config"
	agsserver "github.com/pingcap/agent-git-service/server"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/automation/blob"
	"github.com/tidbcloud/AgentTeam/internal/service/automation/runtime"
	automationstore "github.com/tidbcloud/AgentTeam/internal/store/automation"
)

func TestAGSAutomationSourceStartsAndServerRunnerCompletesCheckout(t *testing.T) {
	ctx := t.Context()
	ags, token := newAGSContractServer(t)
	const (
		owner          = "testuser"
		repo           = "meshx-contract"
		definitionPath = ".agentteam/build.yml"
		workspaceID    = "ws-ags-contract"
		subject        = "user-ags-contract"
	)
	automation := `name: MeshX AGS Contract
on:
  manual: {}
jobs:
  verify:
    runs-on: ubuntu-latest
    steps:
      - uses: agentteam/checkout
        with:
          path: repo
      - run: test -f repo/src/contract.txt
`

	agsJSON(t, ags.Handler(), http.MethodPost, "/api/v3/user/repos", token, map[string]any{
		"name":      repo,
		"auto_init": true,
	}, http.StatusCreated)
	agsJSON(t, ags.Handler(), http.MethodPut, "/api/v3/repos/"+owner+"/"+repo+"/contents/"+definitionPath, token, map[string]any{
		"message": "add automation",
		"content": base64.StdEncoding.EncodeToString([]byte(automation)),
	}, http.StatusCreated)
	agsJSON(t, ags.Handler(), http.MethodPut, "/api/v3/repos/"+owner+"/"+repo+"/contents/src/contract.txt", token, map[string]any{
		"message": "add contract marker",
		"content": base64.StdEncoding.EncodeToString([]byte("ags archive reached meshx runner\n")),
	}, http.StatusCreated)

	store := newAGSContractAutomationStore(t)
	now := time.Date(2026, 6, 14, 9, 0, 0, 0, time.UTC)
	service := NewAutomationsService(store, func() time.Time { return now }, AGSHTTPAutomationSourceResolver{Handler: ags.Handler()})

	run, err := service.StartRepositoryAutomationRun(ctx, workspaceID, subject, AutomationRepositoryRunMutation{
		RepositoryOwner:     owner,
		RepositoryName:      repo,
		DefinitionPath:      definitionPath,
		GitRef:              "main",
		AuthorizationHeader: "token " + token,
	})
	require.NoError(t, err)
	require.Equal(t, automationstore.AutomationRunStatusRunning, run.Status)
	require.Equal(t, owner+"/"+repo, run.RepositoryFullName)
	require.NotEmpty(t, run.CommitSHA)

	blobStore, err := blob.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	runner, err := runtime.NewServerRunner(store, blobStore, runtime.ServerRunnerConfig{
		WorkDir:              automationCheckoutTempDir(t),
		PollInterval:         10 * time.Millisecond,
		JobTimeout:           time.Minute,
		CheckoutTimeout:      time.Minute,
		CheckoutMaxBytes:     1024 * 1024,
		CheckoutMaxFiles:     100,
		CheckoutMaterializer: AGSAutomationCheckoutMaterializer{Archiver: agsServerArchiveAdapter{server: ags}},
		Executor: agsContractExecutor{check: func(t *testing.T, req runtime.ExecuteJobRequest) {
			t.Helper()
			body, err := os.ReadFile(filepath.Join(req.WorkDir, "repo", "src", "contract.txt"))
			require.NoError(t, err)
			require.Equal(t, "ags archive reached meshx runner\n", string(body))
		}, t: t},
	}, func() time.Time { return now }, nil)
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, runner.Close()) })

	runnerCtx, cancel := context.WithCancel(ctx)
	done := make(chan struct{})
	go func() {
		defer close(done)
		runner.Run(runnerCtx)
	}()
	defer func() {
		cancel()
		<-done
	}()
	runner.SignalWorkspace(workspaceID)
	finished := waitForAGSContractRun(t, ctx, store, workspaceID, run.ID)

	require.Equal(t, automationstore.AutomationRunStatusSucceeded, finished.Status)
	jobRuns, err := store.ListJobRuns(ctx, workspaceID, run.ID)
	require.NoError(t, err)
	require.Len(t, jobRuns, 1)
	require.Equal(t, automationstore.AutomationJobRunStatusSucceeded, jobRuns[0].Status, "error_json=%s", jobRuns[0].ErrorJSON)
	events, err := store.ListEvents(ctx, workspaceID, run.ID, 1, 20)
	require.NoError(t, err)
	requireAGSContractEventTypes(t, events, "run_started", "checkout_started", "checkout_succeeded", "job_succeeded", "run_reconciled")
	require.Contains(t, agsContractEvent(t, events, "checkout_succeeded").MetadataJSON, `"strategy":"ags_archive"`)
}

type agsServerArchiveAdapter struct {
	server *agsserver.Server
}

func (a agsServerArchiveAdapter) ArchiveRepository(ctx context.Context, request AutomationRepositoryArchiveRequest) error {
	if a.server == nil {
		return fmt.Errorf("ags server is nil")
	}
	archiveRequest := agsserver.RepositoryArchiveRequest{
		RepositoryFullName: request.RepositoryFullName,
		CommitSHA:          request.CommitSHA,
		Writer:             request.Writer,
	}
	if err := a.server.ArchiveRepository(ctx, archiveRequest); err != nil {
		return fmt.Errorf("archive repository through AGS server: %w", err)
	}
	return nil
}

type agsContractExecutor struct {
	t     *testing.T
	check func(*testing.T, runtime.ExecuteJobRequest)
}

func (e agsContractExecutor) ExecuteJob(ctx context.Context, req runtime.ExecuteJobRequest) (runtime.ExecuteJobResult, error) {
	result := runtime.ExecuteJobResult{Steps: make([]runtime.JobStepResult, 0, len(req.Spec.Steps))}
	for idx, step := range req.Spec.Steps {
		now := req.Now().Format(time.RFC3339Nano)
		exitCode := 0
		var err error
		if strings.TrimSpace(step.Uses) != "" {
			exitCode, err = req.UsesStepHandler(ctx, runtime.UsesStepRequest{Step: step, Index: idx})
		} else if e.check != nil {
			e.check(e.t, req)
		}
		result.Steps = append(result.Steps, runtime.JobStepResult{
			Index:       idx,
			ExitCode:    exitCode,
			StartedAt:   now,
			CompletedAt: now,
		})
		if err != nil {
			return result, err
		}
	}
	return result, nil
}

func newAGSContractServer(t *testing.T) (*agsserver.Server, string) {
	t.Helper()
	token := "meshx-contract-token"
	server, err := agsserver.New(agsconfig.Config{
		Port:        "8080",
		BaseURL:     "http://ags.test",
		DBdsn:       serviceTestSQLiteDSN(t, "ags"),
		GitRepoDir:  t.TempDir(),
		ListenMode:  "production",
		AdminLogin:  "testuser",
		AdminToken:  token,
		Environment: "development",
	})
	require.NoError(t, err)
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		require.NoError(t, server.Shutdown(ctx))
	})
	return server, token
}

func newAGSContractAutomationStore(t *testing.T) *automationstore.Store {
	t.Helper()
	db, err := openServiceTestSQLiteDB(t, "meshx")
	require.NoError(t, err)
	migrateAutomationServiceTestSchema(t, db, &automationstore.AutomationRunner{}, &automationstore.AutomationRunnerSession{}, &automationstore.AutomationLogChunk{}, &automationstore.AutomationArtifact{}, &automationstore.AutomationSecret{})
	return automationstore.New(db)
}

func agsJSON(t *testing.T, handler http.Handler, method string, path string, token string, body map[string]any, wantStatus int) map[string]any {
	t.Helper()
	var reqBody bytes.Buffer
	if body != nil {
		require.NoError(t, json.NewEncoder(&reqBody).Encode(body))
	}
	req := httptest.NewRequest(method, path, &reqBody)
	req.Header.Set("Authorization", "token "+token)
	if body != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	rec := httptest.NewRecorder()
	handler.ServeHTTP(rec, req)
	require.Equal(t, wantStatus, rec.Code, "AGS %s %s body=%s", method, path, rec.Body.String())
	if rec.Body.Len() == 0 {
		return nil
	}
	var payload map[string]any
	require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &payload), "AGS %s %s body=%s", method, path, rec.Body.String())
	return payload
}

func waitForAGSContractRun(t *testing.T, ctx context.Context, store *automationstore.Store, workspaceID string, runID string) automationstore.AutomationRun {
	t.Helper()
	deadline := time.After(5 * time.Second)
	tick := time.NewTicker(10 * time.Millisecond)
	defer tick.Stop()
	for {
		select {
		case <-deadline:
			run, err := store.GetRun(ctx, workspaceID, runID)
			require.NoError(t, err)
			t.Fatalf("automation run did not finish: status=%s", run.Status)
		case <-tick.C:
			run, err := store.GetRun(ctx, workspaceID, runID)
			require.NoError(t, err)
			if run.Status == automationstore.AutomationRunStatusSucceeded || run.Status == automationstore.AutomationRunStatusFailed || run.Status == automationstore.AutomationRunStatusCanceled {
				return run
			}
		}
	}
}

func requireAGSContractEventTypes(t *testing.T, events []automationstore.AutomationEvent, want ...string) {
	t.Helper()
	got := make([]string, 0, len(events))
	for _, event := range events {
		got = append(got, event.EventType)
	}
	require.ElementsMatch(t, want, got)
}

func agsContractEvent(t *testing.T, events []automationstore.AutomationEvent, eventType string) automationstore.AutomationEvent {
	t.Helper()
	for _, event := range events {
		if event.EventType == eventType {
			return event
		}
	}
	t.Fatalf("event %q not found in %#v", eventType, events)
	return automationstore.AutomationEvent{}
}