package server

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	activityservice "github.com/tidbcloud/AgentTeam/internal/service/resource/activity"

	agsserver "github.com/pingcap/agent-git-service/server"
	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/database"
	"github.com/tidbcloud/AgentTeam/internal/store/content"
	activitystore "github.com/tidbcloud/AgentTeam/internal/store/resource/activity"
	reposstore "github.com/tidbcloud/AgentTeam/internal/store/resource/repos"
)

type fakeKBSourceEventAuthorityReader map[string]string

func (reader fakeKBSourceEventAuthorityReader) ListAuthoritiesForBindings(_ context.Context, workspaceID, domain string, bindingKeys []string) (map[string]content.Authority, error) {
	out := make(map[string]content.Authority)
	for _, bindingKey := range bindingKeys {
		if state, ok := reader[bindingKey]; ok {
			out[bindingKey] = content.Authority{WorkspaceID: workspaceID, Domain: domain, BindingKey: bindingKey, State: state}
		}
	}
	return out, nil
}

func newResourceActivityBridgeTestStores(t *testing.T) (*reposstore.Store, *activitystore.Store, func()) {
	t.Helper()
	db, err := database.Open(database.Config{DSN: meshxDatabaseDSN(t)})
	require.NoError(t, err)
	return reposstore.New(db), activitystore.New(db), func() { require.NoError(t, database.Close(db)) }
}

func seedResourceActivityBridgeRepo(t *testing.T, store *reposstore.Store, resourceType, resourceID, agsRepoID string) {
	t.Helper()
	now := time.Date(2026, 6, 29, 12, 0, 0, 0, time.UTC)
	_, err := store.Upsert(context.Background(), reposstore.ResourceRepository{
		ID:           "rr_" + resourceID,
		WorkspaceID:  "ws_1",
		ResourceType: resourceType,
		ResourceID:   resourceID,
		RepoRole:     reposstore.RepoRolePrimary,
		OwnerKind:    "personal",
		RepoOwner:    "owner",
		RepoName:     "repo",
		RepoFullName: "owner/repo",
		AGSRepoID:    agsRepoID,
		DefaultRef:   "main",
		CreatedBy:    "usr_1",
		UpdatedBy:    "usr_1",
		CreatedAt:    now,
		UpdatedAt:    now,
	})
	require.NoError(t, err)
}

func TestAGSResourceActivityEventBridgeWritesPushAndReleaseEvents(t *testing.T) {
	repos, events, cleanup := newResourceActivityBridgeTestStores(t)
	defer cleanup()
	seedResourceActivityBridgeRepo(t, repos, reposstore.ResourceTypeMemory, "mem_1", "42")
	bridge := newAGSResourceActivityEventBridge(repos, events, 4)
	defer bridge.Close()

	occurred := time.Date(2026, 6, 29, 11, 30, 0, 0, time.UTC)
	pushPayload := map[string]any{
		"ref":     "refs/heads/main",
		"before":  "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
		"after":   "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		"created": false,
		"deleted": false,
		"repository": map[string]any{
			"full_name": "owner/repo",
		},
		"commits": []map[string]any{{
			"id":        "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
			"message":   "Update docs\n\nBody",
			"timestamp": occurred.Format(time.RFC3339),
			"url":       "https://example.test/owner/repo/commit/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
			"author": map[string]any{
				"name":  "Alice",
				"email": "alice@example.test",
			},
		}},
		"sender": map[string]any{"login": "meshx-internal"},
	}
	require.NoError(t, bridge.process(context.Background(), agsserver.RepositoryEvent{RepositoryID: 42, Event: "push", PayloadJSON: mustActivityJSON(t, pushPayload)}))

	commitID := activityservice.ResourceActivitySourceEventID(activityservice.ResourceActivityDataPlaneRepository{RepoFullName: "owner/repo"}, "commit", "refs/heads/main", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
	commit, err := events.GetBySourceEventID(context.Background(), "ws_1", activitystore.SourceGitea, commitID)
	require.NoError(t, err)
	require.Equal(t, "commit_pushed", commit.EventType)
	require.Equal(t, "alice@example.test", commit.ActorID)
	require.Equal(t, "Alice", commit.ActorName)
	require.Equal(t, "mem_1", commit.AssetID)
	require.Equal(t, activitystore.AssetTypeMemory, commit.AssetType)

	releasePayload := map[string]any{
		"action": "published",
		"release": map[string]any{
			"id":           99,
			"tag_name":     "v1.0.0",
			"name":         "v1.0.0",
			"html_url":     "https://example.test/owner/repo/releases/tag/v1.0.0",
			"published_at": occurred.Format(time.RFC3339),
			"author":       map[string]any{"login": "bob", "name": "Bob"},
		},
		"repository": map[string]any{"full_name": "owner/repo"},
	}
	require.NoError(t, bridge.process(context.Background(), agsserver.RepositoryEvent{RepositoryID: 42, Event: "release", Action: "published", PayloadJSON: mustActivityJSON(t, releasePayload)}))
	releaseID := activityservice.ResourceActivitySourceEventID(activityservice.ResourceActivityDataPlaneRepository{RepoFullName: "owner/repo"}, "release", "99", "published")
	release, err := events.GetBySourceEventID(context.Background(), "ws_1", activitystore.SourceGitea, releaseID)
	require.NoError(t, err)
	require.Equal(t, "release_published", release.EventType)
	require.Equal(t, "bob", release.ActorID)
	require.Equal(t, "release", release.TargetKind)
}

func TestAGSResourceActivityEventBridgeMapsGitHubCompatibleEvents(t *testing.T) {
	repo := activityservice.ResourceActivityDataPlaneRepository{WorkspaceID: "ws_1", ResourceType: reposstore.ResourceTypeMemory, ResourceID: "mem_1", AssetType: activitystore.AssetTypeMemory, AssetID: "mem_1", RepoRole: reposstore.RepoRolePrimary, RepoFullName: "owner/repo"}
	now := time.Date(2026, 6, 29, 12, 0, 0, 0, time.UTC)
	cases := []struct {
		name      string
		event     agsserver.RepositoryEvent
		payload   agsWebhookPayload
		eventType string
	}{
		{name: "branch create", event: agsserver.RepositoryEvent{Event: "push"}, payload: agsWebhookPayload{Ref: "refs/heads/feature", Created: true, After: "abc", Sender: agsWebhookUser{Login: "alice"}}, eventType: "branch_created"},
		{name: "tag delete", event: agsserver.RepositoryEvent{Event: "push"}, payload: agsWebhookPayload{Ref: "refs/tags/v1", Deleted: true, Before: "def", Sender: agsWebhookUser{Login: "alice"}}, eventType: "tag_deleted"},
		{name: "issue edited", event: agsserver.RepositoryEvent{Event: "issues", Action: "edited"}, payload: agsWebhookPayload{Issue: agsWebhookIssue{Number: 1, Title: "Bug", User: agsWebhookUser{Login: "alice"}, UpdatedAt: &now}}, eventType: "issue_edited"},
		{name: "issue comment", event: agsserver.RepositoryEvent{Event: "issue_comment", Action: "created"}, payload: agsWebhookPayload{Issue: agsWebhookIssue{Number: 1}, Comment: agsWebhookComment{ID: 11, User: agsWebhookUser{Login: "alice"}, CreatedAt: &now}}, eventType: "issue_comment_created"},
		{name: "pull ready", event: agsserver.RepositoryEvent{Event: "pull_request", Action: "ready_for_review"}, payload: agsWebhookPayload{PullRequest: agsWebhookPull{Number: 2, Title: "PR", User: agsWebhookUser{Login: "alice"}, UpdatedAt: &now}}, eventType: "pr_ready_for_review"},
		{name: "pull review", event: agsserver.RepositoryEvent{Event: "pull_request_review", Action: "submitted"}, payload: agsWebhookPayload{PullRequest: agsWebhookPull{Number: 2, Title: "PR"}, Review: agsWebhookReview{ID: 21, State: "APPROVED", User: agsWebhookUser{Login: "alice"}, SubmittedAt: &now}}, eventType: "pr_reviewed"},
		{name: "pull review comment", event: agsserver.RepositoryEvent{Event: "pull_request_review_comment", Action: "resolved"}, payload: agsWebhookPayload{PullRequest: agsWebhookPull{Number: 2}, Comment: agsWebhookComment{ID: 31, User: agsWebhookUser{Login: "alice"}, UpdatedAt: &now}}, eventType: "pr_review_comment_resolved"},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			events := mapAGSEventToResourceActivity(repo, tc.event, tc.payload, now)
			require.NotEmpty(t, events)
			require.Equal(t, tc.eventType, events[0].EventType)
			require.NotEmpty(t, events[0].SourceEventID)
		})
	}
}

func TestAGSResourceActivityEventBridgeSkipsUnknownAndAutomationRepos(t *testing.T) {
	repos, events, cleanup := newResourceActivityBridgeTestStores(t)
	defer cleanup()
	seedResourceActivityBridgeRepo(t, repos, reposstore.ResourceTypeAutomationRepository, reposstore.AutomationRepositoryResourceID("owner/repo"), "42")
	bridge := newAGSResourceActivityEventBridge(repos, events, 1)
	defer bridge.Close()

	payload := map[string]any{
		"ref":        "refs/heads/main",
		"repository": map[string]any{"full_name": "owner/repo"},
		"commits":    []map[string]any{{"id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "timestamp": time.Now().UTC().Format(time.RFC3339)}},
	}
	require.NoError(t, bridge.process(context.Background(), agsserver.RepositoryEvent{RepositoryID: 42, Event: "push", PayloadJSON: mustActivityJSON(t, payload)}))
	page, err := events.ListEvents(context.Background(), activitystore.ListOptions{WorkspaceID: "ws_1", AssetType: activitystore.AssetTypeMemory, VisibleAssetIDs: []string{"mem_1"}})
	require.NoError(t, err)
	require.Empty(t, page.Items)

	require.NoError(t, bridge.process(context.Background(), agsserver.RepositoryEvent{RepositoryID: 404, Event: "push", PayloadJSON: mustActivityJSON(t, payload)}))
}

func TestAGSResourceActivityEventBridgeSkipsNativeKBButPreservesLegacyDomains(t *testing.T) {
	repos, events, cleanup := newResourceActivityBridgeTestStores(t)
	defer cleanup()
	seedResourceActivityBridgeRepo(t, repos, reposstore.ResourceTypeKBSource, "src_1", "42")
	seedResourceActivityBridgeRepo(t, repos, reposstore.ResourceTypeMemory, "mem_1", "42")
	bridge := newAGSResourceActivityEventBridge(repos, events, 1)
	bridge.contentAuthority = fakeKBSourceEventAuthorityReader{"src_1": content.AuthorityStateNativeActive}
	defer bridge.Close()

	payload := map[string]any{
		"ref":        "refs/heads/main",
		"repository": map[string]any{"full_name": "owner/repo"},
		"commits":    []map[string]any{{"id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "timestamp": time.Now().UTC().Format(time.RFC3339)}},
	}
	require.NoError(t, bridge.process(context.Background(), agsserver.RepositoryEvent{RepositoryID: 42, Event: "push", PayloadJSON: mustActivityJSON(t, payload)}))

	kbEvents, err := events.ListEvents(context.Background(), activitystore.ListOptions{WorkspaceID: "ws_1", AssetType: activitystore.AssetTypeKB, VisibleAssetIDs: []string{"src_1"}})
	require.NoError(t, err)
	require.Empty(t, kbEvents.Items, "native KB sources must not consume AGS activity events")
	memoryEvents, err := events.ListEvents(context.Background(), activitystore.ListOptions{WorkspaceID: "ws_1", AssetType: activitystore.AssetTypeMemory, VisibleAssetIDs: []string{"mem_1"}})
	require.NoError(t, err)
	require.Len(t, memoryEvents.Items, 1, "legacy product domains must keep the shared AGS bridge until their own cutover")
}

func TestAGSResourceActivityEventBridgeQueueFullDrops(t *testing.T) {
	bridge := &agsResourceActivityEventBridge{queue: make(chan agsserver.RepositoryEvent, 1), started: true}
	require.NoError(t, bridge.HandleRepositoryEvent(context.Background(), agsserver.RepositoryEvent{RepositoryID: 1, Event: "push"}))
	require.NoError(t, bridge.HandleRepositoryEvent(context.Background(), agsserver.RepositoryEvent{RepositoryID: 1, Event: "push"}))
	require.EqualValues(t, 1, bridge.droppedTotal.Load())
}

func TestAGSResourceActivityEventBridgeQuiesceCanResumeAfterTimeout(t *testing.T) {
	started := make(chan struct{})
	release := make(chan struct{})
	bridge := newAGSResourceActivityEventBridge(nil, nil, 1)
	bridge.processEvent = func(context.Context, agsserver.RepositoryEvent) error {
		close(started)
		<-release
		return nil
	}
	bridge.Start()
	require.NoError(t, bridge.HandleRepositoryEvent(context.Background(), agsserver.RepositoryEvent{Event: "push"}))
	<-started
	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond)
	defer cancel()
	require.ErrorIs(t, bridge.Quiesce(ctx), context.DeadlineExceeded)
	close(release)
	require.NoError(t, bridge.Quiesce(context.Background()))
}

func mustActivityJSON(t *testing.T, payload any) []byte {
	t.Helper()
	data, err := json.Marshal(payload)
	require.NoError(t, err)
	return data
}