package rewind

import (
	"context"
	"database/sql"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/testfixture"
)

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

	ctx := context.Background()
	db := startRewindRollbackDB(t, ctx)
	now := time.Date(2026, 5, 14, 7, 30, 0, 0, time.UTC)

	const (
		workspaceID     = "wsp_rewind_reply_topology"
		userID          = "usr_rewind_reply_topology"
		memberID        = "mem_rewind_reply_topology"
		parentScopeID   = "chn_rewind_reply_parent"
		childScopeID    = "chn_rewind_reply_child"
		firstMessageID  = "msg_rewind_reply_first"
		parentMessageID = "msg_rewind_reply_parent"
		childMessageID  = "msg_rewind_reply_child"
	)

	mustExecRecovery(t, db, `
INSERT INTO users (id, email, name, display_name, avatar_url, is_admin, password_hash, email_verified_at, created_at, updated_at)
VALUES ($1, 'rewind-reply-topology@example.com', 'rewind-reply-topology', 'Rewind Reply Topology', '', TRUE, 'hash', $2, $2, $2)
`, userID, now)
	mustExecRecovery(t, db, `
INSERT INTO workspaces (id, organization_id, slug, name, latest_message_seq, created_at, updated_at)
VALUES ($1, $2, 'rewind-reply-topology', 'Rewind Reply Topology', 3, $3, $3)
`, workspaceID, testfixture.SeedWorkspaceOrganization(t, ctx, db, workspaceID, now), now)
	mustExecRecovery(t, db, `
INSERT INTO members (id, workspace_id, type, display_name, user_id, workspace_role, created_at, updated_at)
VALUES ($1, $2, 'human', 'Human', $3, 'owner', $4, $4)
`, memberID, workspaceID, userID, now)
	mustExecRecovery(t, db, `
INSERT INTO scopes (
    id, workspace_id, type, identity_key, latest_seq, latest_global_seq,
    latest_message_at, archived_until_seq, status, created_at, updated_at
) VALUES (
    $1, $2, 'channel', CONCAT('channel:', $1), 2, 2,
    $3, 0, 'active', $3, $3
)
`, parentScopeID, workspaceID, now)
	mustExecRecovery(t, db, `
INSERT INTO scopes (
    id, workspace_id, type, parent_scope_id, root_message_id, root_message_short_id,
    identity_key, latest_seq, latest_global_seq, latest_message_at,
    archived_until_seq, status, created_at, updated_at
) VALUES (
    $1, $2, 'thread', $3, $4, 'replytop',
    CONCAT('thread:', $3, ':', $4), 1, 3, $5,
    0, 'active', $5, $5
)
`, childScopeID, workspaceID, parentScopeID, parentMessageID, now.Add(time.Second))
	mustExecRecovery(t, db, `
INSERT INTO scope_messages (
    id, workspace_id, scope_id, seq, workspace_global_seq,
    author_member_id, author_type, message_type, content, created_at, updated_at
) VALUES
    ($1, $3, $4, 1, 1, $5, 'human', 'text', 'first', $6, $6),
    ($2, $3, $4, 2, 2, $5, 'human', 'text', 'parent', $6, $6)
`, firstMessageID, parentMessageID, workspaceID, parentScopeID, memberID, now)
	mustExecRecovery(t, db, `
INSERT INTO scope_messages (
    id, workspace_id, scope_id, seq, workspace_global_seq,
    author_member_id, author_type, message_type, content, created_at, updated_at
) VALUES ($1, $2, $3, 1, 3, $4, 'human', 'text', 'child', $5, $5)
`, childMessageID, workspaceID, childScopeID, memberID, now.Add(time.Second))

	store := New(db, Options{})
	cascade, err := store.loadCascadeReplyThreads(ctx, workspaceID, parentScopeID, []SuffixMessage{
		{ID: firstMessageID, MessageType: "text"},
		{ID: parentMessageID, MessageType: "text"},
	}, memberID)
	require.NoError(t, err)
	require.Len(t, cascade, 1)
	require.Equal(t, childScopeID, cascade[0].ScopeID)
	require.Equal(t, parentMessageID, cascade[0].ParentMessageID)
	require.Equal(t, []string{childMessageID}, cascade[0].MessageIDs)
}

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

	db := &cascadeReplyThreadCaptureDB{
		rows: []coredb.Rows{
			&cascadeReplyThreadRows{values: [][]string{
				{"thr_child_a", "msg_tail", "thr_child_a"},
				{"thr_child_b", "msg_tail", "thr_child_b"},
			}},
			&cascadeReplyThreadRows{values: [][]string{
				{"thr_child_a", "msg_child_a1"},
				{"thr_child_a", "msg_child_a2"},
				{"thr_child_b", "msg_child_b1"},
			}},
		},
	}

	store := New(db, Options{})
	cascade, err := store.loadCascadeReplyThreads(context.Background(), "wsp_batch", "chn_parent", []SuffixMessage{
		{ID: "msg_keep", MessageType: "text"},
		{ID: "msg_tail", MessageType: "text"},
	}, "")

	require.NoError(t, err)
	require.Equal(t, []CascadeReplyThread{
		{ScopeID: "thr_child_a", ParentMessageID: "msg_tail", MessageIDs: []string{"msg_child_a1", "msg_child_a2"}},
		{ScopeID: "thr_child_b", ParentMessageID: "msg_tail", MessageIDs: []string{"msg_child_b1"}},
	}, cascade)
	require.Len(t, db.queries, 2)
	discoveryQuery := strings.Join(strings.Fields(db.queries[0]), " ")
	require.Contains(t, discoveryQuery, "WITH suffix_message_ids AS")
	require.NotContains(t, discoveryQuery, "EXISTS ( SELECT 1 FROM scope_messages child_message")
	messageQuery := strings.Join(strings.Fields(db.queries[1]), " ")
	require.Contains(t, messageQuery, "WITH cascade_thread_ids AS")
	require.Contains(t, messageQuery, "JOIN scope_messages m")
	require.Contains(t, messageQuery, "ORDER BY requested.thread_channel_id ASC, m.seq ASC")
}

type cascadeReplyThreadCaptureDB struct {
	queries []string
	args    [][]any
	rows    []coredb.Rows
	nextRow int
}

func (db *cascadeReplyThreadCaptureDB) Query(_ context.Context, query string, args ...any) (coredb.Rows, error) {
	db.queries = append(db.queries, query)
	db.args = append(db.args, append([]any(nil), args...))
	if db.nextRow >= len(db.rows) {
		return &cascadeReplyThreadRows{}, nil
	}
	rows := db.rows[db.nextRow]
	db.nextRow++
	return rows, nil
}

func (db *cascadeReplyThreadCaptureDB) QueryRow(context.Context, string, ...any) coredb.Row {
	return cascadeReplyThreadRow{err: sql.ErrNoRows}
}

func (db *cascadeReplyThreadCaptureDB) Exec(context.Context, string, ...any) (coredb.Result, error) {
	return cascadeReplyThreadResult{}, nil
}

func (db *cascadeReplyThreadCaptureDB) BeginTx(context.Context, coredb.TxOptions) (coredb.Tx, error) {
	return nil, sql.ErrNoRows
}

type cascadeReplyThreadRows struct {
	values [][]string
	index  int
}

func (rows *cascadeReplyThreadRows) Close() {}

func (rows *cascadeReplyThreadRows) Err() error { return nil }

func (rows *cascadeReplyThreadRows) Next() bool {
	return rows.index < len(rows.values)
}

func (rows *cascadeReplyThreadRows) Scan(dest ...any) error {
	if rows.index >= len(rows.values) {
		return sql.ErrNoRows
	}
	values := rows.values[rows.index]
	if len(dest) != len(values) {
		return sql.ErrNoRows
	}
	for index, value := range values {
		target, ok := dest[index].(*string)
		if !ok {
			return sql.ErrNoRows
		}
		*target = value
	}
	rows.index++
	return nil
}

type cascadeReplyThreadRow struct {
	err error
}

func (row cascadeReplyThreadRow) Scan(...any) error {
	if row.err != nil {
		return row.err
	}
	return nil
}

type cascadeReplyThreadResult struct{}

func (cascadeReplyThreadResult) RowsAffected() int64 { return 0 }