package app

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

	"github.com/tidbcloud/AgentTeam/internal/core/identifiers"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
)

const (
	maxMessageSyncDirtyScopes = 50
	// Dirty discovery reads an ordered candidate window before applying
	// visibility and cursor frontier filters. Keep the scan budget tied to the
	// requested page so old covered or currently invisible dirty rows cannot
	// force a full-workspace dirty scan.
	messageSyncDirtyScopeScanMultiplier = 8
	// Dirty candidate reads inspect frontier items across the cursor ancestry, so
	// compact before ordinary browser sync chains can grow into hundreds of
	// generations.
	messageSyncCursorCompactionGenerationThreshold = 128
	messageSyncCursorFrontierLoadBatchSize         = 200
	messageSyncCursorFrontierInsertBatchSize       = 100
)

type messageSyncCursorState struct {
	ID             string
	WorkspaceID    string
	MemberID       string
	UserID         string
	ClientID       string
	RootCursorID   string
	ParentCursorID string
	Generation     int64
	CreatedAt      time.Time
}

type messageSyncCursorChainRow struct {
	ID             string
	ParentCursorID string
	Generation     int64
	CreatedAt      time.Time
}

type messageSyncCursorFrontier struct {
	ScopeID             string
	FrontierSeq         int64
	FrontierActivitySeq int64
}

func (store *Store) CreateMessageSyncCursorForUser(ctx context.Context, userID string, workspaceID string, clientID string) (message.MessageSyncCursor, error) {
	membership, err := store.requireWorkspaceAccess(ctx, userID, workspaceID)
	if err != nil {
		return message.MessageSyncCursor{}, err
	}

	cursorID, err := identifiers.NewEntityID("msc")
	if err != nil {
		return message.MessageSyncCursor{}, fmt.Errorf("generate message sync cursor id: %w", err)
	}
	now := time.Now().UTC()
	if err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		if _, err := tx.Exec(ctx, `
INSERT INTO message_sync_cursors (
  id, workspace_id, member_id, user_id, client_id, root_cursor_id,
  parent_cursor_id, generation, created_at, expires_at
) VALUES ($1, $2, $3, $4, $5, $1, NULL, 0, $6, NULL)
`, cursorID, strings.TrimSpace(workspaceID), membership.Member.ID, strings.TrimSpace(userID), strings.TrimSpace(clientID), now); err != nil {
			return fmt.Errorf("create message sync cursor: %w", err)
		}

		query, args, err := message.BuildMessageSyncInitialCursorItemsQuery(workspaceID, membership.Member.ID, cursorID, now)
		if err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, query, args...); err != nil {
			return fmt.Errorf("create message sync cursor frontier: %w", err)
		}
		return nil
	}); err != nil {
		return message.MessageSyncCursor{}, err
	}

	return message.MessageSyncCursor{Cursor: cursorID}, nil
}

func (store *Store) SyncMessagesForUserCursor(ctx context.Context, userID string, workspaceID string, cursor string, limit int) (message.MessageSyncPage, error) {
	membership, err := store.requireWorkspaceAccess(ctx, userID, workspaceID)
	if err != nil {
		return message.MessageSyncPage{}, err
	}

	normalizedLimit, err := normalizeMessageSyncCursorLimit(limit)
	if err != nil {
		return message.MessageSyncPage{}, err
	}

	state, err := store.loadMessageSyncCursorState(ctx, store.db, cursor, workspaceID, membership.Member.ID)
	if err != nil {
		return message.MessageSyncPage{}, err
	}
	if state.Generation >= messageSyncCursorCompactionGenerationThreshold {
		state, err = store.compactMessageSyncCursor(ctx, state, userID, workspaceID, membership.Member.ID, nil)
		if err != nil {
			return message.MessageSyncPage{}, err
		}
	}

	dirtyScopeLimit := normalizedLimit
	if dirtyScopeLimit > maxMessageSyncDirtyScopes {
		dirtyScopeLimit = maxMessageSyncDirtyScopes
	}
	dirtyScopeScanLimit := messageSyncDirtyScopeScanLimit(dirtyScopeLimit)
	candidates, err := store.loadMessageSyncDirtyScopeCandidates(ctx, workspaceID, membership.Member.ID, state, dirtyScopeLimit, dirtyScopeScanLimit)
	if err != nil {
		return message.MessageSyncPage{}, err
	}
	if len(candidates) == 0 {
		return message.MessageSyncPage{Cursor: state.ID, Scopes: []message.MessageSyncScopePage{}}, nil
	}

	messagesByScope, maxSeqByScope, err := store.loadMessageSyncScopeMessages(ctx, workspaceID, membership.Member.ID, candidates, normalizedLimit)
	if err != nil {
		return message.MessageSyncPage{}, err
	}

	scopePages := make([]message.MessageSyncScopePage, 0, len(candidates))
	frontierUpdates := make(map[string]int64, len(candidates))
	for _, candidate := range candidates {
		messages := messagesByScope[candidate.ScopeID]
		if len(messages) == 0 {
			continue
		}
		nextSeq := maxSeqByScope[candidate.ScopeID]
		if nextSeq < candidate.FrontierSeq {
			nextSeq = candidate.FrontierSeq
		}
		frontierUpdates[candidate.ScopeID] = nextSeq
		scopePages = append(scopePages, message.MessageSyncScopePage{
			ScopeID:  candidate.ScopeID,
			Messages: messages,
			NextSeq:  nextSeq,
			HasMore:  nextSeq < candidate.LatestMessageSeq,
		})
	}
	if len(scopePages) == 0 {
		return message.MessageSyncPage{Cursor: state.ID, Scopes: []message.MessageSyncScopePage{}}, nil
	}

	nextCursor, err := store.createMessageSyncChildCursor(ctx, state, userID, workspaceID, membership.Member.ID, frontierUpdates)
	if err != nil {
		return message.MessageSyncPage{}, err
	}
	return message.MessageSyncPage{Cursor: nextCursor, Scopes: scopePages}, nil
}

func normalizeMessageSyncCursorLimit(limit int) (int, error) {
	if limit < 0 {
		return 0, badRequest("invalid limit")
	}
	if limit == 0 {
		return defaultSyncMessagesLimit, nil
	}
	if limit > maxSyncMessagesLimit {
		return maxSyncMessagesLimit, nil
	}
	return limit, nil
}

func messageSyncDirtyScopeScanLimit(limit int) int {
	if limit <= 0 {
		return 0
	}
	return limit * messageSyncDirtyScopeScanMultiplier
}

func (store *Store) loadMessageSyncCursorState(ctx context.Context, queryer coredb.Queryer, cursor string, workspaceID string, memberID string) (messageSyncCursorState, error) {
	var state messageSyncCursorState
	var parent sql.NullString
	var clientID sql.NullString
	row := queryer.QueryRow(ctx, `
SELECT id, workspace_id, member_id, user_id, client_id, root_cursor_id,
       parent_cursor_id, generation, created_at
FROM message_sync_cursors
WHERE id = $1
  AND workspace_id = $2
  AND member_id = $3
`, strings.TrimSpace(cursor), strings.TrimSpace(workspaceID), strings.TrimSpace(memberID))
	if err := row.Scan(
		&state.ID,
		&state.WorkspaceID,
		&state.MemberID,
		&state.UserID,
		&clientID,
		&state.RootCursorID,
		&parent,
		&state.Generation,
		&state.CreatedAt,
	); err != nil {
		if coredb.IsNotFound(err) {
			return messageSyncCursorState{}, badRequest("cursor not found")
		}
		return messageSyncCursorState{}, fmt.Errorf("load message sync cursor: %w", err)
	}
	state.ParentCursorID = parent.String
	state.ClientID = clientID.String
	return state, nil
}

func (store *Store) loadMessageSyncDirtyScopeCandidates(ctx context.Context, workspaceID string, memberID string, state messageSyncCursorState, limit int, scanLimit int) ([]message.MessageSyncDirtyScopeCandidate, error) {
	cursorIDs, err := loadMessageSyncCursorAncestorIDs(ctx, store.db, state)
	if err != nil {
		return nil, err
	}
	query, args, err := message.BuildMessageSyncDirtyScopeCandidatesQuery(workspaceID, memberID, cursorIDs, limit, scanLimit)
	if err != nil {
		return nil, err
	}
	rows, err := store.db.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("load message sync dirty scopes: %w", err)
	}
	defer rows.Close()

	candidates := make([]message.MessageSyncDirtyScopeCandidate, 0)
	for rows.Next() {
		var candidate message.MessageSyncDirtyScopeCandidate
		if err := rows.Scan(&candidate.ScopeID, &candidate.FrontierSeq, &candidate.LatestMessageSeq); err != nil {
			return nil, fmt.Errorf("scan message sync dirty scope: %w", err)
		}
		if strings.TrimSpace(candidate.ScopeID) == "" || candidate.LatestMessageSeq <= candidate.FrontierSeq {
			continue
		}
		candidates = append(candidates, candidate)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate message sync dirty scopes: %w", err)
	}
	return candidates, nil
}

func (store *Store) loadMessageSyncScopeMessages(ctx context.Context, workspaceID string, memberID string, candidates []message.MessageSyncDirtyScopeCandidate, perScopeLimit int) (map[string][]message.AppMessage, map[string]int64, error) {
	query, args, err := message.BuildMessageSyncMessagesForScopesQuery(workspaceID, candidates, perScopeLimit)
	if err != nil {
		return nil, nil, err
	}
	if strings.TrimSpace(query) == "" {
		return map[string][]message.AppMessage{}, map[string]int64{}, nil
	}

	rows, err := store.db.Query(ctx, query, args...)
	if err != nil {
		return nil, nil, fmt.Errorf("load message sync scope messages: %w", err)
	}
	defer rows.Close()

	allMessages := make([]message.AppMessage, 0)
	for rows.Next() {
		appMessage, err := scanAppMessageRow(rows)
		if err != nil {
			return nil, nil, err
		}
		allMessages = append(allMessages, appMessage)
	}
	if err := rows.Err(); err != nil {
		return nil, nil, fmt.Errorf("iterate message sync scope messages: %w", err)
	}

	attachmentLookup, err := store.loadAttachmentsForMessages(ctx, workspaceID, collectAppMessageIDs(allMessages))
	if err != nil {
		return nil, nil, err
	}
	for i := range allMessages {
		allMessages[i].Attachments = attachmentLookup[allMessages[i].ID]
	}
	if err := store.attachAppMessageReadDecorations(ctx, workspaceID, strings.TrimSpace(memberID), allMessages); err != nil {
		return nil, nil, err
	}
	messagesByScope := make(map[string][]message.AppMessage, len(candidates))
	maxSeqByScope := make(map[string]int64, len(candidates))
	for _, appMessage := range allMessages {
		messagesByScope[appMessage.ChannelID] = append(messagesByScope[appMessage.ChannelID], appMessage)
		if appMessage.Seq > maxSeqByScope[appMessage.ChannelID] {
			maxSeqByScope[appMessage.ChannelID] = appMessage.Seq
		}
	}
	return messagesByScope, maxSeqByScope, nil
}

func (store *Store) createMessageSyncChildCursor(ctx context.Context, state messageSyncCursorState, userID string, workspaceID string, memberID string, frontierUpdates map[string]int64) (string, error) {
	if state.Generation+1 >= messageSyncCursorCompactionGenerationThreshold {
		compacted, err := store.compactMessageSyncCursor(ctx, state, userID, workspaceID, memberID, frontierUpdates)
		if err != nil {
			return "", err
		}
		return compacted.ID, nil
	}

	nextCursorID, err := identifiers.NewEntityID("msc")
	if err != nil {
		return "", fmt.Errorf("generate message sync child cursor id: %w", err)
	}
	now := time.Now().UTC()
	if err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		if _, err := tx.Exec(ctx, `
INSERT INTO message_sync_cursors (
  id, workspace_id, member_id, user_id, client_id, root_cursor_id,
  parent_cursor_id, generation, created_at, expires_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NULL)
`, nextCursorID, strings.TrimSpace(workspaceID), strings.TrimSpace(memberID), strings.TrimSpace(userID), state.ClientID, state.RootCursorID, state.ID, state.Generation+1, now); err != nil {
			return fmt.Errorf("create message sync child cursor: %w", err)
		}
		frontiers := make(map[string]messageSyncCursorFrontier, len(frontierUpdates))
		for scopeID, frontierSeq := range frontierUpdates {
			trimmedScopeID := strings.TrimSpace(scopeID)
			if trimmedScopeID == "" || frontierSeq < 0 {
				continue
			}
			frontiers[trimmedScopeID] = messageSyncCursorFrontier{
				ScopeID:             trimmedScopeID,
				FrontierSeq:         frontierSeq,
				FrontierActivitySeq: 0,
			}
		}
		if err := insertMessageSyncCursorFrontiers(ctx, tx, nextCursorID, frontiers, now); err != nil {
			return err
		}
		return nil
	}); err != nil {
		return "", err
	}
	return nextCursorID, nil
}

func (store *Store) compactMessageSyncCursor(ctx context.Context, state messageSyncCursorState, userID string, workspaceID string, memberID string, frontierUpdates map[string]int64) (messageSyncCursorState, error) {
	nextCursorID, err := identifiers.NewEntityID("msc")
	if err != nil {
		return messageSyncCursorState{}, fmt.Errorf("generate compacted message sync cursor id: %w", err)
	}

	now := time.Now().UTC()
	compacted := messageSyncCursorState{
		ID:           nextCursorID,
		WorkspaceID:  strings.TrimSpace(workspaceID),
		MemberID:     strings.TrimSpace(memberID),
		UserID:       strings.TrimSpace(userID),
		ClientID:     state.ClientID,
		RootCursorID: nextCursorID,
		Generation:   0,
		CreatedAt:    now,
	}
	if err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		frontiers, err := store.loadMessageSyncEffectiveFrontiers(ctx, tx, state)
		if err != nil {
			return err
		}
		for scopeID, frontierSeq := range frontierUpdates {
			trimmedScopeID := strings.TrimSpace(scopeID)
			if trimmedScopeID == "" || frontierSeq < 0 {
				continue
			}
			frontiers[trimmedScopeID] = messageSyncCursorFrontier{
				ScopeID:             trimmedScopeID,
				FrontierSeq:         frontierSeq,
				FrontierActivitySeq: 0,
			}
		}

		if _, err := tx.Exec(ctx, `
INSERT INTO message_sync_cursors (
  id, workspace_id, member_id, user_id, client_id, root_cursor_id,
  parent_cursor_id, generation, created_at, expires_at
) VALUES ($1, $2, $3, $4, $5, $1, NULL, 0, $6, NULL)
`, nextCursorID, compacted.WorkspaceID, compacted.MemberID, compacted.UserID, compacted.ClientID, now); err != nil {
			return fmt.Errorf("create compacted message sync cursor: %w", err)
		}
		if err := insertMessageSyncCursorFrontiers(ctx, tx, nextCursorID, frontiers, now); err != nil {
			return err
		}
		return nil
	}); err != nil {
		return messageSyncCursorState{}, err
	}
	return compacted, nil
}

func (store *Store) loadMessageSyncEffectiveFrontiers(ctx context.Context, queryer coredb.Queryer, state messageSyncCursorState) (map[string]messageSyncCursorFrontier, error) {
	chainIDs, err := loadMessageSyncCursorAncestorIDs(ctx, queryer, state)
	if err != nil {
		return nil, err
	}
	itemsByCursorID, err := loadMessageSyncCursorFrontiersByCursor(ctx, queryer, chainIDs)
	if err != nil {
		return nil, err
	}

	frontiers := make(map[string]messageSyncCursorFrontier)
	for _, cursorID := range chainIDs {
		for _, frontier := range itemsByCursorID[cursorID] {
			if strings.TrimSpace(frontier.ScopeID) == "" || frontier.FrontierSeq < 0 {
				continue
			}
			if _, exists := frontiers[frontier.ScopeID]; exists {
				continue
			}
			frontiers[frontier.ScopeID] = frontier
		}
	}
	return frontiers, nil
}

func loadMessageSyncCursorAncestorIDs(ctx context.Context, queryer coredb.Queryer, state messageSyncCursorState) ([]string, error) {
	rootCursorID := strings.TrimSpace(state.RootCursorID)
	workspaceID := strings.TrimSpace(state.WorkspaceID)
	memberID := strings.TrimSpace(state.MemberID)
	current := messageSyncCursorChainRow{
		ID:             strings.TrimSpace(state.ID),
		ParentCursorID: strings.TrimSpace(state.ParentCursorID),
		Generation:     state.Generation,
		CreatedAt:      state.CreatedAt,
	}
	chainIDs := make([]string, 0)
	visited := make(map[string]struct{})
	for current.ID != "" {
		if _, exists := visited[current.ID]; exists {
			return nil, fmt.Errorf("message sync cursor chain contains a cycle at %s", current.ID)
		}
		visited[current.ID] = struct{}{}
		chainIDs = append(chainIDs, current.ID)

		parentID := strings.TrimSpace(current.ParentCursorID)
		if parentID == "" {
			break
		}
		parent, err := loadMessageSyncCursorChainRow(ctx, queryer, parentID, rootCursorID, workspaceID, memberID)
		if err != nil {
			return nil, err
		}
		if parent.Generation >= current.Generation {
			return nil, fmt.Errorf("message sync cursor chain has non-decreasing generation at %s", parent.ID)
		}
		current = parent
	}
	if len(chainIDs) == 0 {
		return nil, fmt.Errorf("message sync cursor chain is empty")
	}
	return chainIDs, nil
}

func loadMessageSyncCursorChainRow(ctx context.Context, queryer coredb.Queryer, cursorID string, rootCursorID string, workspaceID string, memberID string) (messageSyncCursorChainRow, error) {
	var row messageSyncCursorChainRow
	var parent sql.NullString
	if err := queryer.QueryRow(ctx, `
SELECT id, parent_cursor_id, generation, created_at
FROM message_sync_cursors
WHERE id = $1
  AND root_cursor_id = $2
  AND workspace_id = $3
  AND member_id = $4
`, strings.TrimSpace(cursorID), rootCursorID, workspaceID, memberID).Scan(&row.ID, &parent, &row.Generation, &row.CreatedAt); err != nil {
		if coredb.IsNotFound(err) {
			return messageSyncCursorChainRow{}, fmt.Errorf("message sync cursor chain is missing cursor %s", strings.TrimSpace(cursorID))
		}
		return messageSyncCursorChainRow{}, fmt.Errorf("load message sync cursor chain row: %w", err)
	}
	row.ParentCursorID = strings.TrimSpace(parent.String)
	return row, nil
}

func loadMessageSyncCursorFrontiersByCursor(ctx context.Context, queryer coredb.Queryer, cursorIDs []string) (map[string][]messageSyncCursorFrontier, error) {
	itemsByCursorID := make(map[string][]messageSyncCursorFrontier, len(cursorIDs))
	for start := 0; start < len(cursorIDs); start += messageSyncCursorFrontierLoadBatchSize {
		end := start + messageSyncCursorFrontierLoadBatchSize
		if end > len(cursorIDs) {
			end = len(cursorIDs)
		}
		query, args := buildMessageSyncCursorFrontiersForCursorsQuery(cursorIDs[start:end])
		rows, err := queryer.Query(ctx, query, args...)
		if err != nil {
			return nil, fmt.Errorf("load message sync cursor frontiers: %w", err)
		}
		for rows.Next() {
			var cursorID string
			var frontier messageSyncCursorFrontier
			if err := rows.Scan(&cursorID, &frontier.ScopeID, &frontier.FrontierSeq, &frontier.FrontierActivitySeq); err != nil {
				rows.Close()
				return nil, fmt.Errorf("scan message sync cursor frontier: %w", err)
			}
			frontier.ScopeID = strings.TrimSpace(frontier.ScopeID)
			cursorID = strings.TrimSpace(cursorID)
			if cursorID != "" && frontier.ScopeID != "" {
				itemsByCursorID[cursorID] = append(itemsByCursorID[cursorID], frontier)
			}
		}
		if err := rows.Err(); err != nil {
			rows.Close()
			return nil, fmt.Errorf("iterate message sync cursor frontiers: %w", err)
		}
		rows.Close()
	}
	return itemsByCursorID, nil
}

func buildMessageSyncCursorFrontiersForCursorsQuery(cursorIDs []string) (string, []any) {
	placeholders := make([]string, 0, len(cursorIDs))
	args := make([]any, 0, len(cursorIDs))
	for _, cursorID := range cursorIDs {
		placeholders = append(placeholders, fmt.Sprintf("$%d", len(args)+1))
		args = append(args, strings.TrimSpace(cursorID))
	}
	return fmt.Sprintf(`
SELECT cursor_id, scope_id, frontier_seq, frontier_activity_seq
FROM message_sync_cursor_items
WHERE cursor_id IN (%s)
`, strings.Join(placeholders, ", ")), args
}

func insertMessageSyncCursorFrontiers(ctx context.Context, tx coredb.Tx, cursorID string, frontiers map[string]messageSyncCursorFrontier, updatedAt time.Time) error {
	if len(frontiers) == 0 {
		return nil
	}

	batch := make([]messageSyncCursorFrontier, 0, messageSyncCursorFrontierInsertBatchSize)
	flush := func() error {
		if len(batch) == 0 {
			return nil
		}
		values := make([]string, 0, len(batch))
		args := make([]any, 0, len(batch)*5)
		for _, frontier := range batch {
			base := len(args) + 1
			values = append(values, fmt.Sprintf("($%d, $%d, $%d, $%d, $%d)", base, base+1, base+2, base+3, base+4))
			args = append(args, cursorID, frontier.ScopeID, frontier.FrontierSeq, frontier.FrontierActivitySeq, updatedAt)
		}
		query := fmt.Sprintf(`
INSERT INTO message_sync_cursor_items (
  cursor_id, scope_id, frontier_seq, frontier_activity_seq, updated_at
) VALUES %s
ON DUPLICATE KEY UPDATE
  frontier_seq = VALUES(frontier_seq),
  frontier_activity_seq = VALUES(frontier_activity_seq),
  updated_at = VALUES(updated_at)
`, strings.Join(values, ",\n"))
		if _, err := tx.Exec(ctx, query, args...); err != nil {
			return fmt.Errorf("create message sync cursor frontiers: %w", err)
		}
		batch = batch[:0]
		return nil
	}

	for _, frontier := range frontiers {
		frontier.ScopeID = strings.TrimSpace(frontier.ScopeID)
		if frontier.ScopeID == "" || frontier.FrontierSeq < 0 {
			continue
		}
		batch = append(batch, frontier)
		if len(batch) == messageSyncCursorFrontierInsertBatchSize {
			if err := flush(); err != nil {
				return err
			}
		}
	}
	return flush()
}