package im

import (
	"context"
	"fmt"
	"strings"

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

const conversationIndexProjectionSelect = `
idx.workspace_id, idx.member_id, idx.conversation_id, idx.scope_id, idx.kind, idx.sort_at,
idx.last_event_seq, idx.pinned_at, idx.archived_at,
idx.projection_json, idx.created_at, idx.updated_at,
CASE
  WHEN idx.kind = 'task' THEN COALESCE((
    SELECT CASE WHEN sns.task_watch_state = 'active' THEN 'active' ELSE 'mute' END
    FROM scope_notification_settings sns
    WHERE sns.scope_id = idx.scope_id
      AND sns.member_id = idx.member_id
    LIMIT 1
  ), 'mute')
  ELSE ''
END AS subscription_state`

// conversationIndexProjectedSourceValidityPredicate is the lightweight source
// integrity filter for the established list and single-row projection reads. It
// deliberately does not recompute membership authorization: ListConversations
// must remain an index-backed member-local page read.
const conversationIndexProjectedSourceValidityPredicate = `
idx.kind IN ('channel', 'dm', 'thread', 'task', 'service_account', 'replyThread', 'humanDm', 'agentDm', 'serviceDm', 'serviceAccount')
AND (
  idx.kind <> 'task'
  OR EXISTS (
    SELECT 1
    FROM tasks task
    JOIN scopes task_scope
      ON task_scope.id = task.task_scope_id
     AND task_scope.workspace_id = idx.workspace_id
     AND task_scope.type = 'task'
     AND task_scope.status = 'active'
     AND task_scope.archived_at IS NULL
    WHERE task.task_scope_id = idx.scope_id
  )
)
AND (
  idx.kind NOT IN ('service_account', 'serviceAccount')
  OR EXISTS (
    SELECT 1
    FROM service_accounts service_account
    WHERE service_account.workspace_id = idx.workspace_id
      AND service_account.message_scope_id = idx.scope_id
  )
)`

// conversationIndexBoundedExactLookupAuthorizationPredicate is only for
// LookupMemberVisibleConversationProjectionsByIDs after the caller-supplied IDs
// have been deduplicated and bounded to MaxConversationLookupIDs. Never use it
// in list, pagination, or other index-scan queries.
const conversationIndexBoundedExactLookupAuthorizationPredicate = `
idx.kind IN ('channel', 'dm', 'thread', 'task', 'service_account', 'replyThread', 'humanDm', 'agentDm', 'serviceDm', 'serviceAccount')
AND (
  (
    idx.kind IN ('channel', 'dm', 'thread', 'replyThread', 'humanDm', 'agentDm', 'serviceDm')
    AND NOT EXISTS (
      SELECT 1
      FROM scopes source_scope
      WHERE source_scope.workspace_id = idx.workspace_id
        AND source_scope.id = idx.scope_id
    )
  )
  OR EXISTS (
    SELECT 1
    FROM scopes scope
    WHERE scope.workspace_id = idx.workspace_id
      AND scope.id = idx.scope_id
      AND scope.status = 'active'
      AND scope.archived_at IS NULL
      AND (
        (idx.kind = 'channel' AND scope.type = 'channel')
        OR (idx.kind IN ('dm', 'humanDm', 'agentDm', 'serviceDm') AND scope.type = 'dm')
        OR (idx.kind IN ('thread', 'replyThread') AND scope.type = 'thread')
        OR (idx.kind = 'task' AND scope.type = 'task')
        OR (idx.kind IN ('service_account', 'serviceAccount') AND scope.type = 'service_account')
      )
      AND (
        (
          scope.type <> 'thread'
          AND EXISTS (
            SELECT 1
            FROM scope_memberships self_membership
            WHERE self_membership.scope_id = scope.id
              AND self_membership.member_id = idx.member_id
              AND self_membership.removed_at IS NULL
              AND self_membership.archived_at IS NULL
              AND self_membership.effective_to_seq IS NULL
          )
        )
        OR (
          scope.type = 'thread'
          AND EXISTS (
            SELECT 1
            FROM scope_notification_settings notification
            WHERE notification.scope_id = scope.id
              AND notification.member_id = idx.member_id
              AND notification.state IN ('active', 'quiet', 'mute')
          )
          AND EXISTS (
            SELECT 1
            FROM scopes parent_scope
            WHERE parent_scope.workspace_id = idx.workspace_id
              AND parent_scope.id = scope.parent_scope_id
              AND parent_scope.status = 'active'
              AND parent_scope.archived_at IS NULL
              AND (
                (
                  parent_scope.type = 'channel'
                  AND EXISTS (
                    SELECT 1
                    FROM scope_memberships parent_membership
                    WHERE parent_membership.scope_id = parent_scope.id
                      AND parent_membership.member_id = idx.member_id
                      AND parent_membership.removed_at IS NULL
                      AND parent_membership.archived_at IS NULL
                      AND parent_membership.effective_to_seq IS NULL
                      AND parent_membership.visibility = 'public'
                  )
                )
                OR (
                  parent_scope.type = 'dm'
                  AND (
                    EXISTS (
                      SELECT 1
                      FROM scope_memberships child_membership
                      WHERE child_membership.scope_id = scope.id
                        AND child_membership.member_id = idx.member_id
                        AND child_membership.removed_at IS NULL
                        AND child_membership.archived_at IS NULL
                        AND child_membership.effective_to_seq IS NULL
                    )
                    OR EXISTS (
                      SELECT 1
                      FROM scope_memberships parent_membership
                      WHERE parent_membership.scope_id = parent_scope.id
                        AND parent_membership.member_id = idx.member_id
                        AND parent_membership.removed_at IS NULL
                        AND parent_membership.archived_at IS NULL
                        AND parent_membership.effective_to_seq IS NULL
                    )
                  )
                )
              )
          )
        )
      )
      AND (
        scope.type <> 'task'
        OR EXISTS (
          SELECT 1
          FROM tasks task
          WHERE task.task_scope_id = scope.id
        )
      )
      AND (
        scope.type <> 'service_account'
        OR EXISTS (
          SELECT 1
          FROM service_accounts service_account
          WHERE service_account.workspace_id = idx.workspace_id
            AND service_account.message_scope_id = scope.id
        )
      )
  )
)`

// UpsertConversation stores one member-local conversation projection.
func (store *Store) UpsertConversation(ctx context.Context, workspaceID string, memberID string, item ConversationIndexItem) error {
	if err := store.Validate(); err != nil {
		return err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return err
	}
	conversationID := normalizedString(item.ConversationID)
	if conversationID == "" {
		return fmt.Errorf("conversation id is required")
	}
	kind := normalizedString(item.Kind)
	if kind == "" {
		return fmt.Errorf("conversation kind is required")
	}
	sortAt := item.SortAt.UTC()
	if sortAt.IsZero() {
		sortAt = nowUTC()
	}
	item.ConversationID = conversationID
	item.Kind = kind
	item.SortAt = sortAt
	return upsertConversationTx(ctx, store.db, workspaceID, memberID, item)
}

// ListConversations reads one backend-sorted member-local conversation index page.
func (store *Store) ListConversations(ctx context.Context, workspaceID string, memberID string, limit int, cursor ConversationCursor, kinds []string) (ConversationIndexPage, error) {
	if err := store.Validate(); err != nil {
		return ConversationIndexPage{}, err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return ConversationIndexPage{}, err
	}
	listQuery, err := buildConversationIndexListQuery(workspaceID, memberID, limit, cursor, kinds)
	if err != nil {
		return ConversationIndexPage{}, err
	}

	rows, err := store.db.Query(ctx, listQuery.statement, listQuery.args...)
	if err != nil {
		return ConversationIndexPage{}, fmt.Errorf("list im conversation index: %w", err)
	}
	defer rows.Close()

	items := make([]ConversationIndexItem, 0, listQuery.fetchLimit)
	for rows.Next() {
		item, err := scanConversationIndexItem(rows)
		if err != nil {
			return ConversationIndexPage{}, err
		}
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return ConversationIndexPage{}, fmt.Errorf("iterate im conversation index: %w", err)
	}
	page := ConversationIndexPage{Items: items}
	if len(items) > listQuery.limit {
		page.HasMore = true
		page.Items = items[:listQuery.limit]
		last := page.Items[len(page.Items)-1]
		page.NextCursor = ConversationCursor{PinnedAt: last.PinnedAt, SortAt: last.SortAt, ConversationID: last.ConversationID}
	}
	return page, nil
}

type conversationIndexListQuery struct {
	statement  string
	args       []any
	limit      int
	fetchLimit int
}

func buildConversationIndexListQuery(workspaceID string, memberID string, limit int, cursor ConversationCursor, kinds []string) (conversationIndexListQuery, error) {
	cursor.ConversationID = strings.TrimSpace(cursor.ConversationID)
	cursorIsZero := cursor.SortAt.IsZero() && cursor.ConversationID == "" && cursor.PinnedAt == nil
	if !cursorIsZero && (cursor.SortAt.IsZero() || cursor.ConversationID == "") {
		return conversationIndexListQuery{}, fmt.Errorf("invalid conversation index cursor")
	}
	if cursor.PinnedAt != nil {
		normalizedPinnedAt := cursor.PinnedAt.UTC()
		if normalizedPinnedAt.IsZero() {
			return conversationIndexListQuery{}, fmt.Errorf("invalid conversation index cursor")
		}
		cursor.PinnedAt = &normalizedPinnedAt
	}
	cursor.SortAt = cursor.SortAt.UTC()
	limit = normalizeLimit(limit, defaultConversationIndexListSize, maxConversationIndexListSize)
	fetchLimit := limit + 1

	// Explicit Task Unwatch is a member-local serving filter. Keep the
	// index row durable so Watch can reveal the same projection again; do not
	// reuse conversation_member_states hidden/archive semantics here.
	query := `
SELECT ` + conversationIndexProjectionSelect + `
FROM im_conversation_index idx
WHERE idx.workspace_id = $1
  AND idx.member_id = $2
  AND idx.archived_at IS NULL
  AND (
    idx.kind <> 'task'
    OR NOT EXISTS (
      SELECT 1
      FROM scope_notification_settings sns
      WHERE sns.scope_id = idx.scope_id
        AND sns.member_id = idx.member_id
        AND sns.task_watch_state = 'mute'
    )
  )
  AND ` + conversationIndexProjectedSourceValidityPredicate + `
`
	args := []any{workspaceID, memberID}
	normalizedKinds := normalizeConversationIndexKinds(kinds)
	if len(normalizedKinds) > 0 {
		placeholders := make([]string, 0, len(normalizedKinds))
		for _, kind := range normalizedKinds {
			args = append(args, kind)
			placeholders = append(placeholders, fmt.Sprintf("$%d", len(args)))
		}
		query += fmt.Sprintf("  AND idx.kind IN (%s)\n", strings.Join(placeholders, ", "))
	}
	if !cursor.SortAt.IsZero() {
		if cursor.PinnedAt != nil {
			args = append(args, cursor.SortAt, cursor.ConversationID)
			sortAtArg := len(args) - 1
			conversationIDArg := len(args)
			query += fmt.Sprintf(`  AND (
    idx.pinned_at IS NULL
    OR (
      idx.pinned_at IS NOT NULL
      AND (idx.sort_at < $%d OR (idx.sort_at = $%d AND idx.conversation_id > $%d))
    )
  )
`, sortAtArg, sortAtArg, conversationIDArg)
		} else {
			args = append(args, cursor.SortAt, cursor.ConversationID)
			query += fmt.Sprintf("  AND idx.pinned_at IS NULL\n  AND (idx.sort_at < $%d OR (idx.sort_at = $%d AND idx.conversation_id > $%d))\n", len(args)-1, len(args)-1, len(args))
		}
	}
	query += fmt.Sprintf("ORDER BY CASE WHEN idx.pinned_at IS NULL THEN 1 ELSE 0 END ASC, idx.sort_at DESC, idx.conversation_id ASC\nLIMIT $%d\n", len(args)+1)
	args = append(args, fetchLimit)
	return conversationIndexListQuery{
		statement:  query,
		args:       args,
		limit:      limit,
		fetchLimit: fetchLimit,
	}, nil
}

// GetConversation returns one conversation projection by member-local id.
func (store *Store) GetConversation(ctx context.Context, workspaceID string, memberID string, conversationID string) (ConversationIndexItem, error) {
	if err := store.Validate(); err != nil {
		return ConversationIndexItem{}, err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return ConversationIndexItem{}, err
	}
	conversationID = normalizedString(conversationID)
	if conversationID == "" {
		return ConversationIndexItem{}, fmt.Errorf("conversation id is required")
	}
	row := store.db.QueryRow(ctx, `
SELECT `+conversationIndexProjectionSelect+`
FROM im_conversation_index idx
WHERE idx.workspace_id = $1
  AND idx.member_id = $2
  AND idx.conversation_id = $3
  AND `+conversationIndexProjectedSourceValidityPredicate+`
`, workspaceID, memberID, conversationID)
	item, err := scanConversationIndexItem(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return ConversationIndexItem{}, fmt.Errorf("conversation not found: %w", err)
		}
		return ConversationIndexItem{}, err
	}
	return item, nil
}

// LookupMemberVisibleConversationProjectionsByIDs returns active member-visible
// projections for a bounded set of exact conversation IDs. Its cost is tied
// only to the request IDs and it never hydrates one source object per row.
func (store *Store) LookupMemberVisibleConversationProjectionsByIDs(
	ctx context.Context,
	workspaceID string,
	memberID string,
	conversationIDs []string,
) ([]ConversationIndexItem, error) {
	if err := store.Validate(); err != nil {
		return nil, err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return nil, err
	}
	uniqueIDs := make([]string, 0, len(conversationIDs))
	seen := make(map[string]struct{}, len(conversationIDs))
	for _, conversationID := range conversationIDs {
		conversationID = normalizedString(conversationID)
		if conversationID == "" {
			continue
		}
		if _, exists := seen[conversationID]; exists {
			continue
		}
		seen[conversationID] = struct{}{}
		uniqueIDs = append(uniqueIDs, conversationID)
	}
	if len(uniqueIDs) == 0 {
		return []ConversationIndexItem{}, nil
	}
	if len(uniqueIDs) > MaxConversationLookupIDs {
		return nil, fmt.Errorf(
			"conversation lookup must not exceed %d ids",
			MaxConversationLookupIDs,
		)
	}
	requestedConversationsCTE, requestedConversationArgs, _, err :=
		coredb.BuildIndexedStringValuesCTE(
			"requested_conversations",
			"conversation_id",
			"ordinality",
			uniqueIDs,
			3,
		)
	if err != nil {
		return nil, err
	}
	args := make([]any, 0, len(requestedConversationArgs)+2)
	args = append(args, workspaceID, memberID)
	args = append(args, requestedConversationArgs...)
	rows, err := store.db.Query(ctx, `
WITH `+requestedConversationsCTE+`
SELECT `+conversationIndexProjectionSelect+`
FROM requested_conversations requested
JOIN im_conversation_index idx
  ON idx.workspace_id = $1
 AND idx.member_id = $2
 AND idx.conversation_id = requested.conversation_id
WHERE idx.archived_at IS NULL
  AND `+conversationIndexBoundedExactLookupAuthorizationPredicate+`
ORDER BY requested.ordinality
`, args...)
	if err != nil {
		return nil, fmt.Errorf("lookup im conversations: %w", err)
	}
	defer rows.Close()

	items := make([]ConversationIndexItem, 0, len(uniqueIDs))
	for rows.Next() {
		item, err := scanConversationIndexItem(rows)
		if err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate im conversation lookup: %w", err)
	}
	return items, nil
}

func scanConversationIndexItem(row interface{ Scan(...any) error }) (ConversationIndexItem, error) {
	var item ConversationIndexItem
	var projectionRaw []byte
	if err := row.Scan(
		&item.WorkspaceID,
		&item.MemberID,
		&item.ConversationID,
		&item.ScopeID,
		&item.Kind,
		&item.SortAt,
		&item.LastEventSeq,
		&item.PinnedAt,
		&item.ArchivedAt,
		&projectionRaw,
		&item.CreatedAt,
		&item.UpdatedAt,
		&item.SubscriptionState,
	); err != nil {
		return ConversationIndexItem{}, fmt.Errorf("scan im conversation index item: %w", err)
	}
	projection, err := decodeJSONMap(projectionRaw)
	if err != nil {
		return ConversationIndexItem{}, err
	}
	item.Projection = projection
	return item, nil
}