package app_test

import (
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

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

	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	. "github.com/tidbcloud/AgentTeam/internal/api/testkit"
	"github.com/tidbcloud/AgentTeam/internal/service/appruntime"
	"github.com/tidbcloud/AgentTeam/internal/store/app"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/model"
)

// TestExternalAppThreadReplyUsesRuntimeCommittedMessageFanout exercises the
// production API-test composition end to end. It guards the failure mode where
// an autonomous Store producer commits a visible reply and outbox row but no
// caller owns the first unread/message.created projection.
func TestExternalAppThreadReplyUsesRuntimeCommittedMessageFanout(t *testing.T) {
	t.Parallel()

	runtime := StartRuntime(t)
	router := gin.New()
	runtime.RegisterRoutes(router)
	server := httptest.NewServer(router)
	defer server.Close()

	client := server.Client()
	admin := RegisterUserForOrganizationTests(t, client, server.URL, "app-runtime-fanout@example.com", "app-runtime-fanout")
	var workspaces []dto.WorkspaceResponse
	response := DoJSON(t, client, http.MethodGet, server.URL+"/api/v1/workspaces", nil, AuthHeaders(admin.AccessToken), &workspaces)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.NotEmpty(t, workspaces)
	workspaceID := workspaces[0].ID
	ownerMemberID := TaskWorkspaceMemberIDForUser(t, runtime, workspaceID, admin.User.ID)

	var channel ChannelResponse
	response = DoJSON(t, client, http.MethodPost, server.URL+"/api/v1/workspaces/"+workspaceID+"/channels", map[string]any{
		"name":        "app-runtime-fanout",
		"description": "external App reply fanout integration",
	}, WorkspaceHeaders(admin.AccessToken, workspaceID), &channel)
	require.Equal(t, http.StatusOK, response.StatusCode)
	require.NotEmpty(t, channel.ID)

	installation := requestAndApproveRunnerAppForAPIIntegrationTest(t, runtime, admin.User.ID, workspaceID, "runtime-fanout-app")
	response = DoJSON(t, client, http.MethodPost, server.URL+"/api/v1/workspaces/"+workspaceID+"/channels/"+channel.ID+"/members", map[string]any{
		"serviceMemberId": installation.MemberID,
	}, WorkspaceHeaders(admin.AccessToken, workspaceID), nil)
	require.Equal(t, http.StatusOK, response.StatusCode)

	credential, err := runtime.Store.App().CreateAppRunnerCredentialForUser(
		runtime.Ctx,
		admin.User.ID,
		workspaceID,
		installation.ID,
		app.AppRunnerCredentialCreateParams{Name: "fanout integration runner"},
	)
	require.NoError(t, err)
	runnerService := appruntime.NewRunnerService(appruntime.RunnerServiceOptions{Store: runtime.Store.Graph.ServiceActor, Authority: runtime.Store.App()})
	runnerAuth, err := runnerService.AuthenticateRunner(runtime.Ctx, credential.Token)
	require.NoError(t, err)

	author := model.User{ID: admin.User.ID, Name: admin.User.Name, DisplayName: admin.User.DisplayName}
	parent, _, err := runtime.Store.App().SendMessageForUser(runtime.Ctx, author, workspaceID, channel.ID, "parent for external App reply", nil)
	require.NoError(t, err)
	replyThread, err := runtime.Store.OpenReplyThreadForUser(runtime.Ctx, admin.User.ID, workspaceID, channel.ID, parent.ID)
	require.NoError(t, err)
	threadID := replyThread.ThreadChannelID
	require.NotEmpty(t, threadID)

	_, _, err = runtime.Store.App().SendMessageForUser(
		runtime.Ctx,
		author,
		workspaceID,
		threadID,
		"@"+installation.Handle+" inspect this reply thread",
		nil,
	)
	require.NoError(t, err)

	claimed, err := runnerService.ClaimDeliveries(runtime.Ctx, runnerAuth, appruntime.RunnerClaimParams{
		RunnerID:      "runtime-fanout-runner",
		RunnerVersion: "1.0.0",
		Limit:         1,
		LeaseDuration: time.Minute,
	})
	require.NoError(t, err)
	require.Len(t, claimed.Envelopes, 1)
	envelope := claimed.Envelopes[0]
	deliveryAuth, err := runtime.Store.App().AuthenticateAppRuntimeDelivery(
		runtime.Ctx,
		envelope.DeliveryToken,
		envelope.Delivery.ID,
		envelope.Attempt.ID,
	)
	require.NoError(t, err)

	require.Zero(t, runtime.UnreadRuntime.ScopeUnreadCounts(workspaceID, ownerMemberID, []string{threadID})[threadID])
	params := app.AppRuntimeReplyParams{
		AttemptID:      envelope.Attempt.ID,
		Content:        "external App reply projected through real runtime fanout",
		IdempotencyKey: "runtime-fanout-reply-1",
	}
	reply, err := runtime.Store.App().ReplyAppRuntimeDelivery(runtime.Ctx, deliveryAuth, params)
	require.NoError(t, err)
	require.NotEmpty(t, reply.MessageID)

	require.Eventually(t, func() bool {
		return runtime.UnreadRuntime.ScopeUnreadCounts(workspaceID, ownerMemberID, []string{threadID})[threadID] == 1
	}, 5*time.Second, 25*time.Millisecond)

	assertExternalAppReplyFanoutState(t, runtime, workspaceID, ownerMemberID, threadID, reply.MessageID)

	replayed, err := runtime.Store.App().ReplyAppRuntimeDelivery(runtime.Ctx, deliveryAuth, params)
	require.NoError(t, err)
	require.Equal(t, reply.MessageID, replayed.MessageID)
	require.Equal(t, int64(1), runtime.UnreadRuntime.ScopeUnreadCounts(workspaceID, ownerMemberID, []string{threadID})[threadID])
	assertExternalAppReplyFanoutState(t, runtime, workspaceID, ownerMemberID, threadID, reply.MessageID)
}

func assertExternalAppReplyFanoutState(
	t *testing.T,
	runtime Runtime,
	workspaceID string,
	ownerMemberID string,
	threadID string,
	messageID string,
) {
	t.Helper()

	var projectionCount int64
	require.NoError(t, runtime.Pool.QueryRow(runtime.Ctx, `
SELECT COUNT(*)
FROM im_stream_events
WHERE workspace_id = $1
  AND member_id = $2
  AND scope_id = $3
  AND type = 'message.created'
  AND entity_id = $4
`, workspaceID, ownerMemberID, threadID, messageID).Scan(&projectionCount))
	require.Equal(t, int64(1), projectionCount)

	var outboxCount int64
	var status string
	var payload []byte
	require.Eventually(t, func() bool {
		outboxCount = 0
		status = ""
		payload = nil
		if err := runtime.Pool.QueryRow(runtime.Ctx, `
SELECT COUNT(*)
FROM scope_fanout_outbox
WHERE message_id = $1
  AND event_ref_type = 'message'
  AND event_ref_id = $1
  AND fanout_type = 'scope_fanout'
`, messageID).Scan(&outboxCount); err != nil || outboxCount != 1 {
			return false
		}
		if err := runtime.Pool.QueryRow(runtime.Ctx, `
SELECT status, payload
FROM scope_fanout_outbox
WHERE message_id = $1
  AND event_ref_type = 'message'
  AND event_ref_id = $1
  AND fanout_type = 'scope_fanout'
`, messageID).Scan(&status, &payload); err != nil {
			return false
		}
		return status == "delivered" && len(payload) > 0
	}, 5*time.Second, 25*time.Millisecond)
	require.Equal(t, int64(1), outboxCount)
	require.Equal(t, "delivered", status)
	require.NotEmpty(t, payload)

	decoded, err := message.UnmarshalScopeFanoutPayload(payload)
	require.NoError(t, err)
	require.Equal(t, messageID, decoded.AppMessage.ID)
	require.Equal(t, threadID, decoded.AppMessage.ChannelID)
	require.Equal(t, messageID, decoded.Delivery.MessageID)
	require.Equal(t, threadID, decoded.Delivery.Message.TargetKey)
	require.True(t, decoded.Delivery.RecoverableStreamProjection)
	require.Empty(t, decoded.Delivery.WakeRecipients, "silent external App replies must not wake Agents")
}