package message

import (
	"context"
	"fmt"
	"strings"
	"time"
	"unicode/utf8"

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

// channel_attachment_projection is a read projection for the channel
// attachments list. One row per message-bound attachment, flattening the
// immutable message / scope / parent-scope / author-reference / content facts
// so the list can be served by a single ordered query instead of the batched
// 14-JOIN candidate dance.
//
// What is and is not materialized:
//   - Immutable facts (scope ids, message seq/created_at/content at write time,
//     attachment filename/content_type/ordinal, author member id/type, owner
//     flags) are stored.
//   - Mutable presentation is NOT stored. The read path live-joins members /
//     users / agents for the author and the thread root scope_messages row for
//     the thread title, so profile edits and thread-root edits/recalls are
//     reflected immediately -- matching the legacy query and avoiding the
//     stale-denormalization class of bugs.
//
// Write side:
//   - ReconcileChannelAttachmentProjectionForMessage rebuilds the rows for one
//     message from source tables (claim/copy). Idempotent; non channel/thread
//     scopes yield no rows.
//   - The scopeMessageTombstoneProjectionSteps entry deletes rows on
//     rewind/recall/withdraw alongside the attachments hard-delete.
//
// Read side:
//   - ListChannelAttachmentsFromProjection serves the unified read for both
//     channel-scoped (optionally including reply threads) and single-thread
//     requests. Visibility is public-only (ListChannelAttachmentsForUser /
//     ListThreadAttachmentsForUser are human-only, and humans are always public
//     members), checked via an EXISTS on scope_memberships.
const channelAttachmentProjectionTable = "channel_attachment_projection"

const defaultChannelAttachmentProjectionLimit = 30

// ReconcileChannelAttachmentProjectionForMessage rebuilds the projection rows
// for a single message from source tables within the caller's transaction. It
// is the write-side maintenance entry point, called after attachment
// claim/copy. Idempotent: it deletes then re-inserts, so a message whose
// attachments shrank to zero ends up with no rows. Messages in non
// channel/thread scopes produce no rows (filtered by s.type).
func ReconcileChannelAttachmentProjectionForMessage(ctx context.Context, execer coredb.QueryExecer, messageID string) error {
	messageID = strings.TrimSpace(messageID)
	if messageID == "" {
		return nil
	}
	if _, err := execer.Exec(ctx, `DELETE FROM `+channelAttachmentProjectionTable+` WHERE message_id = $1`, messageID); err != nil {
		return fmt.Errorf("delete channel attachment projection for message: %w", err)
	}
	if _, err := execer.Exec(ctx, reconcileChannelAttachmentProjectionForMessageSQL, messageID); err != nil {
		return fmt.Errorf("insert channel attachment projection for message: %w", err)
	}
	return nil
}

// reconcileChannelAttachmentProjectionForMessageSQL re-inserts the projection
// rows for the message bound to $1. attachment_ordinal is the 1-based absolute
// position within the message ordered by (attachment.created_at, id), matching
// the legacy BuildLoadAttachmentsForMessagesQuery ORDER BY so cursor ordinals
// stay comparable across the legacy and projection paths. Mutable author
// presentation and thread-root text are intentionally not stored (live-joined
// at read).
const reconcileChannelAttachmentProjectionForMessageSQL = `
INSERT INTO channel_attachment_projection (
  id, workspace_id, scope_id, scope_type, parent_scope_id,
  message_id, message_seq, message_created_at,
  thread_root_message_id,
  author_member_id, author_type,
  filename, content_type, size_bytes, attachment_ordinal, attachment_created_at,
  is_human_visible_type, is_not_rewound, scope_status_active, scope_not_archived, parent_scope_active,
  scope_owner_member_id, author_is_owner, author_has_task_participant,
  created_at, updated_at
)
SELECT
  a.id, m.workspace_id, s.id, s.type,
  CASE WHEN s.type = 'thread' THEN COALESCE(s.parent_scope_id, '') ELSE s.id END,
  a.message_id, m.seq, m.created_at,
  CASE WHEN s.type = 'thread' THEN COALESCE(s.root_message_id, '') ELSE '' END,
  m.author_member_id, m.author_type,
  a.filename, a.content_type, a.size_bytes,
  ROW_NUMBER() OVER (PARTITION BY a.message_id ORDER BY a.created_at ASC, a.id ASC),
  a.created_at,
  CASE WHEN m.message_type IN ('task_description_attachment', 'task_dispatch') THEN 0 ELSE 1 END,
  CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(CASE WHEN JSON_VALID(m.metadata) THEN m.metadata ELSE JSON_OBJECT() END, '$.messageRecall.status')) = 'rewound' THEN 0 ELSE 1 END,
  CASE WHEN s.status = 'active' THEN 1 ELSE 0 END,
  CASE WHEN s.archived_at IS NULL THEN 1 ELSE 0 END,
  CASE WHEN s.type <> 'thread' OR ps.id IS NOT NULL THEN 1 ELSE 0 END,
  COALESCE(scp.owner_member_id, ''),
  CASE WHEN m.author_member_id = COALESCE(scp.owner_member_id, '') THEN 1 ELSE 0 END,
  0,
  a.created_at, a.created_at
FROM attachments a
JOIN scope_messages m ON m.id = a.message_id AND m.workspace_id = a.workspace_id AND m.scope_id = a.scope_id
JOIN scopes s ON s.id = a.scope_id AND s.workspace_id = a.workspace_id
LEFT JOIN scopes ps ON s.type = 'thread' AND ps.id = s.parent_scope_id AND ps.status = 'active' AND ps.archived_at IS NULL
LEFT JOIN scope_channel_profiles scp ON scp.scope_id = s.id
WHERE a.message_id = $1
  AND a.message_id IS NOT NULL
  AND s.type IN ('channel', 'thread')
`

// BackfillChannelAttachmentProjectionForWorkspace rebuilds the projection for
// every message-bound attachment in a workspace. Intended as a backfill helper
// (idempotent: deletes then re-inserts the workspace's rows, safe to re-run).
func BackfillChannelAttachmentProjectionForWorkspace(ctx context.Context, execer coredb.QueryExecer, workspaceID string) (int64, error) {
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return 0, fmt.Errorf("workspace is required")
	}
	if _, err := execer.Exec(ctx, `DELETE FROM `+channelAttachmentProjectionTable+` WHERE workspace_id = $1`, workspaceID); err != nil {
		return 0, fmt.Errorf("clear channel attachment projection for backfill: %w", err)
	}
	result, err := execer.Exec(ctx, backfillChannelAttachmentProjectionForWorkspaceSQL, workspaceID)
	if err != nil {
		return 0, fmt.Errorf("backfill channel attachment projection: %w", err)
	}
	return result.RowsAffected(), nil
}

// backfillChannelAttachmentProjectionForWorkspaceSQL is the workspace-scoped
// form of reconcileChannelAttachmentProjectionForMessageSQL (no per-message
// filter), keyed by $1 = workspace_id.
const backfillChannelAttachmentProjectionForWorkspaceSQL = `
INSERT INTO channel_attachment_projection (
  id, workspace_id, scope_id, scope_type, parent_scope_id,
  message_id, message_seq, message_created_at,
  thread_root_message_id,
  author_member_id, author_type,
  filename, content_type, size_bytes, attachment_ordinal, attachment_created_at,
  is_human_visible_type, is_not_rewound, scope_status_active, scope_not_archived, parent_scope_active,
  scope_owner_member_id, author_is_owner, author_has_task_participant,
  created_at, updated_at
)
SELECT
  a.id, m.workspace_id, s.id, s.type,
  CASE WHEN s.type = 'thread' THEN COALESCE(s.parent_scope_id, '') ELSE s.id END,
  a.message_id, m.seq, m.created_at,
  CASE WHEN s.type = 'thread' THEN COALESCE(s.root_message_id, '') ELSE '' END,
  m.author_member_id, m.author_type,
  a.filename, a.content_type, a.size_bytes,
  ROW_NUMBER() OVER (PARTITION BY a.message_id ORDER BY a.created_at ASC, a.id ASC),
  a.created_at,
  CASE WHEN m.message_type IN ('task_description_attachment', 'task_dispatch') THEN 0 ELSE 1 END,
  CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(CASE WHEN JSON_VALID(m.metadata) THEN m.metadata ELSE JSON_OBJECT() END, '$.messageRecall.status')) = 'rewound' THEN 0 ELSE 1 END,
  CASE WHEN s.status = 'active' THEN 1 ELSE 0 END,
  CASE WHEN s.archived_at IS NULL THEN 1 ELSE 0 END,
  CASE WHEN s.type <> 'thread' OR ps.id IS NOT NULL THEN 1 ELSE 0 END,
  COALESCE(scp.owner_member_id, ''),
  CASE WHEN m.author_member_id = COALESCE(scp.owner_member_id, '') THEN 1 ELSE 0 END,
  0,
  a.created_at, a.created_at
FROM attachments a
JOIN scope_messages m ON m.id = a.message_id AND m.workspace_id = a.workspace_id AND m.scope_id = a.scope_id
JOIN scopes s ON s.id = a.scope_id AND s.workspace_id = a.workspace_id
LEFT JOIN scopes ps ON s.type = 'thread' AND ps.id = s.parent_scope_id AND ps.status = 'active' AND ps.archived_at IS NULL
LEFT JOIN scope_channel_profiles scp ON scp.scope_id = s.id
WHERE a.workspace_id = $1
  AND a.message_id IS NOT NULL
  AND s.type IN ('channel', 'thread')
`

// ChannelAttachmentProjectionListRequest is the read request for the projection
// path. It mirrors the legacy candidate request but the limit is the page size
// (the caller over-fetches by one to compute HasMore).
//
// Channel mode: ChannelID is a channel; IncludeThreads controls whether reply
// threads under it are included. ThreadScopeID mode: ThreadScopeID is a single
// reply-thread scope and the result is scoped to messages in that thread only.
type ChannelAttachmentProjectionListRequest struct {
	WorkspaceID    string
	ViewerMemberID string
	ChannelID      string
	ThreadScopeID  string
	Kind           ChannelAttachmentKind
	Query          string
	FromMemberID   string
	DateFrom       *time.Time
	DateTo         *time.Time
	IncludeThreads bool
	Cursor         ChannelAttachmentCursor
	Limit          int
}

// BuildChannelAttachmentProjectionListQuery builds the unified single-query
// read. The viewer must hold an open-ended public membership on the message's
// scope (or on the parent channel for threads); owner-private authority is
// intentionally not handled here (this API is human-only).
// channelAttachmentProjectionListQuerySQL is the projection list SELECT body.
// %s is the AND-joined WHERE predicates built by
// channelAttachmentProjectionListPredicates; $%d is the over-fetch LIMIT
// (limit+1) the caller appends as the last positional arg. $1 (viewer member)
// is referenced by the EXISTS membership clause; $2 (workspace) by the base
// predicate.
const channelAttachmentProjectionListQuerySQL = `
SELECT
  p.id, p.message_id, p.message_created_at, msg.content AS message_content,
  p.scope_id, p.scope_type, p.parent_scope_id,
  p.thread_root_message_id, COALESCE(root.content, '') AS thread_root_text,
  am.type, p.author_member_id,
  COALESCE(u.id, ''), COALESCE(ag.id, ''), COALESCE(ag.name, ''),
  COALESCE(u.display_name, ''), COALESCE(am.display_name, ''), COALESCE(u.name, ''),
  COALESCE(u.avatar_url, ''), COALESCE(ag.avatar_url, ''),
  p.filename, p.content_type, p.size_bytes, p.attachment_ordinal
FROM channel_attachment_projection p
JOIN scope_messages msg ON msg.id = p.message_id AND msg.workspace_id = p.workspace_id AND msg.scope_id = p.scope_id
JOIN members am ON am.id = p.author_member_id
LEFT JOIN users u ON am.type = 'human' AND u.id = am.user_id
LEFT JOIN agents ag ON am.type = 'agent' AND ag.member_id = am.id
LEFT JOIN scope_messages root ON p.thread_root_message_id <> '' AND root.id = p.thread_root_message_id AND root.workspace_id = p.workspace_id AND root.scope_id = p.parent_scope_id
WHERE %s
  AND EXISTS (
    SELECT 1
    FROM scope_memberships sm
    WHERE sm.member_id = $1
      AND sm.removed_at IS NULL
      AND sm.visibility = 'public'
      AND (
        (sm.scope_id = p.scope_id
         AND p.message_seq >= sm.effective_from_seq
         AND (sm.effective_to_seq IS NULL OR p.message_seq <= sm.effective_to_seq))
        OR (p.scope_type = 'thread'
            AND sm.scope_id = p.parent_scope_id
            AND sm.archived_at IS NULL
            AND sm.effective_to_seq IS NULL)
      )
  )
ORDER BY p.message_created_at DESC, p.message_id ASC, p.attachment_ordinal ASC
LIMIT $%d
`

func BuildChannelAttachmentProjectionListQuery(request ChannelAttachmentProjectionListRequest) (string, []any, error) {
	workspaceID := strings.TrimSpace(request.WorkspaceID)
	viewerMemberID := strings.TrimSpace(request.ViewerMemberID)
	if workspaceID == "" || viewerMemberID == "" {
		return "", nil, fmt.Errorf("workspace and viewer member are required")
	}
	threadScopeID := strings.TrimSpace(request.ThreadScopeID)
	if threadScopeID == "" && strings.TrimSpace(request.ChannelID) == "" {
		return "", nil, fmt.Errorf("channel or thread scope is required")
	}
	limit := request.Limit
	if limit <= 0 {
		limit = defaultChannelAttachmentProjectionLimit
	}

	where, args := channelAttachmentProjectionListPredicates(request)
	// Over-fetch by one so the caller can compute HasMore without a second query.
	args = append(args, limit+1)
	limitParam := len(args)
	query := fmt.Sprintf(channelAttachmentProjectionListQuerySQL, strings.Join(where, "\n  AND "), limitParam)
	return query, args, nil
}

// channelAttachmentProjectionListPredicates builds the WHERE predicates and the
// positional args for the projection list query. $1 is the viewer member id and
// $2 the workspace id (both referenced by channelAttachmentProjectionListQuerySQL);
// filter args follow in order. The LIMIT arg is appended by the caller after the
// predicates so it becomes the last positional parameter.
func channelAttachmentProjectionListPredicates(request ChannelAttachmentProjectionListRequest) ([]string, []any) {
	threadScopeID := strings.TrimSpace(request.ThreadScopeID)
	threadMode := threadScopeID != ""
	args := []any{
		strings.TrimSpace(request.ViewerMemberID),
		strings.TrimSpace(request.WorkspaceID),
	}
	where := []string{
		"p.workspace_id = $2",
		"p.is_human_visible_type = 1",
		"p.is_not_rewound = 1",
		"p.scope_status_active = 1",
		"p.scope_not_archived = 1",
		"(p.scope_type <> 'thread' OR p.parent_scope_active = 1)",
		channelAttachmentProjectionKindPredicate(NormalizeChannelAttachmentKind(request.Kind)),
	}
	if threadMode {
		args = append(args, threadScopeID)
		where = append(where, fmt.Sprintf("p.scope_id = $%d", len(args)))
		where = append(where, "p.scope_type = 'thread'")
	} else {
		args = append(args, strings.TrimSpace(request.ChannelID), request.IncludeThreads)
		channelParam := len(args) - 1
		includeParam := len(args)
		where = append(where, fmt.Sprintf("(p.parent_scope_id = $%d AND ($%d OR p.scope_type = 'channel'))", channelParam, includeParam))
	}
	if fromMemberID := strings.TrimSpace(request.FromMemberID); fromMemberID != "" {
		args = append(args, fromMemberID)
		where = append(where, fmt.Sprintf("p.author_member_id = $%d", len(args)))
	}
	if request.DateFrom != nil && !request.DateFrom.IsZero() {
		args = append(args, request.DateFrom.UTC())
		where = append(where, fmt.Sprintf("p.message_created_at >= $%d", len(args)))
	}
	if request.DateTo != nil && !request.DateTo.IsZero() {
		args = append(args, request.DateTo.UTC())
		where = append(where, fmt.Sprintf("p.message_created_at <= $%d", len(args)))
	}
	if q := strings.ToLower(strings.TrimSpace(request.Query)); q != "" {
		args = append(args, "%"+q+"%")
		queryParam := len(args)
		// Message-level match (matches legacy): a message qualifies if its body
		// matches OR any of its attachments' filename/content_type matches; then
		// every kind-matching attachment row of that message is returned. The
		// body is shared across the message's rows, so it is tested on the outer
		// row; the attachment match ranges over the message's rows via p2.
		where = append(where, fmt.Sprintf(`(
  LOWER(msg.content) LIKE $%[1]d
  OR EXISTS (
      SELECT 1
      FROM channel_attachment_projection p2
      WHERE p2.message_id = p.message_id
        AND (LOWER(p2.filename) LIKE $%[1]d OR LOWER(p2.content_type) LIKE $%[1]d)
  )
)`, queryParam))
	}
	if !request.Cursor.IsZero() {
		args = append(args, request.Cursor.CreatedAt.UTC(), strings.TrimSpace(request.Cursor.MessageID), request.Cursor.ResourceOrdinal)
		createdAtParam := len(args) - 2
		messageIDParam := len(args) - 1
		ordinalParam := len(args)
		where = append(where, fmt.Sprintf(`(
  p.message_created_at < $%[1]d
  OR (p.message_created_at = $%[1]d AND p.message_id > $%[2]d)
  OR (p.message_created_at = $%[1]d AND p.message_id = $%[2]d AND p.attachment_ordinal > $%[3]d)
)`, createdAtParam, messageIDParam, ordinalParam))
	}
	return where, args
}

// channelAttachmentProjectionKindPredicate filters attachment rows by MIME kind.
// Comparisons are case-insensitive (LOWER) so mixed-case content-type such as
// "Image/PNG" is classified consistently with the row scan and the legacy path.
func channelAttachmentProjectionKindPredicate(kind ChannelAttachmentKind) string {
	switch kind {
	case ChannelAttachmentKindImage:
		return "LOWER(p.content_type) LIKE 'image/%'"
	case ChannelAttachmentKindFile:
		return "LOWER(p.content_type) NOT LIKE 'image/%'"
	default:
		return "1 = 1"
	}
}

// ListChannelAttachmentsFromProjection serves the channel attachments list
// from the projection table (unified read). It preserves the legacy total
// order (message_created_at DESC, message_id ASC, attachment_ordinal ASC) and
// the over-fetch-by-one HasMore/NextCursor contract.
func (store *Store) ListChannelAttachmentsFromProjection(ctx context.Context, request ChannelAttachmentProjectionListRequest) (ChannelAttachmentPage, error) {
	limit := request.Limit
	if limit <= 0 {
		limit = defaultChannelAttachmentProjectionLimit
	}
	query, args, err := BuildChannelAttachmentProjectionListQuery(request)
	if err != nil {
		return ChannelAttachmentPage{}, err
	}
	rows, err := store.db.Query(ctx, query, args...)
	if err != nil {
		return ChannelAttachmentPage{}, fmt.Errorf("list channel attachments from projection: %w", err)
	}
	defer rows.Close()

	items := make([]ChannelAttachmentItem, 0, limit+1)
	for rows.Next() {
		item, err := scanChannelAttachmentProjectionRow(rows)
		if err != nil {
			return ChannelAttachmentPage{}, err
		}
		items = append(items, item)
	}
	if err := rows.Err(); err != nil {
		return ChannelAttachmentPage{}, fmt.Errorf("iterate channel attachment projection: %w", err)
	}

	page := ChannelAttachmentPage{Items: items}
	if len(items) > limit {
		page.HasMore = true
		page.Items = items[:limit]
		if len(page.Items) > 0 {
			page.NextCursor = page.Items[len(page.Items)-1].Cursor
		}
	}
	return page, nil
}

func scanChannelAttachmentProjectionRow(rows coredb.Rows) (ChannelAttachmentItem, error) {
	var (
		item                ChannelAttachmentItem
		memberType          string
		authorMemberID      string
		userID              string
		agentID             string
		agentName           string
		userDisplayName     string
		memberDisplayName   string
		userName            string
		userAvatarURL       string
		agentAvatarURL      string
		messageID           string
		messageContent      string
		scopeID             string
		scopeType           string
		parentScopeID       string
		threadRootMessageID string
		threadRootText      string
		filename            string
		contentType         string
		sizeBytes           int64
		createdAt           time.Time
		ordinal             int
	)
	if err := rows.Scan(
		&item.ID, &messageID, &createdAt, &messageContent,
		&scopeID, &scopeType, &parentScopeID,
		&threadRootMessageID, &threadRootText,
		&memberType, &authorMemberID,
		&userID, &agentID, &agentName,
		&userDisplayName, &memberDisplayName, &userName,
		&userAvatarURL, &agentAvatarURL,
		&filename, &contentType, &sizeBytes, &ordinal,
	); err != nil {
		return ChannelAttachmentItem{}, fmt.Errorf("scan channel attachment projection row: %w", err)
	}
	item.Kind = ChannelAttachmentKindFile
	if strings.HasPrefix(strings.ToLower(strings.TrimSpace(contentType)), "image/") {
		item.Kind = ChannelAttachmentKindImage
	}
	item.Title = strings.TrimSpace(filename)
	if item.Title == "" {
		item.Title = strings.TrimSpace(item.ID)
	}
	item.CreatedAt = createdAt
	item.Author = authorProfileFromResourceRow(memberType, authorMemberID, userID, agentID, agentName, userDisplayName, memberDisplayName, userName, userAvatarURL, agentAvatarURL)
	item.Message = channelAttachmentProjectionLocation(messageID, scopeID, scopeType, parentScopeID, threadRootMessageID, threadRootText)
	item.Snippet = channelAttachmentProjectionSnippet(messageContent)
	item.Attachment = &ChannelAttachmentDetail{
		ID:          strings.TrimSpace(item.ID),
		Filename:    strings.TrimSpace(filename),
		ContentType: strings.TrimSpace(contentType),
		SizeBytes:   sizeBytes,
	}
	item.Cursor = ChannelAttachmentCursor{CreatedAt: createdAt, MessageID: messageID, ResourceOrdinal: ordinal}
	return item, nil
}

func channelAttachmentProjectionLocation(messageID, scopeID, scopeType, parentScopeID, threadRootMessageID, threadRootText string) ChannelAttachmentMessageLocation {
	scopeType = strings.TrimSpace(scopeType)
	parentChannelID := strings.TrimSpace(parentScopeID)
	if parentChannelID == "" {
		parentChannelID = strings.TrimSpace(scopeID)
	}
	location := ChannelAttachmentMessageLocation{
		MessageID:       strings.TrimSpace(messageID),
		ScopeID:         strings.TrimSpace(scopeID),
		ScopeType:       scopeType,
		ParentChannelID: parentChannelID,
	}
	if scopeType == "thread" {
		location.ThreadChannelID = strings.TrimSpace(scopeID)
		location.ThreadRootMessageID = strings.TrimSpace(threadRootMessageID)
		location.ThreadTitle = channelAttachmentProjectionSnippet(threadRootText)
	}
	return location
}

func channelAttachmentProjectionSnippet(content string) string {
	trimmed := strings.TrimSpace(strings.Join(strings.Fields(content), " "))
	if trimmed == "" {
		return ""
	}
	const limit = 180
	if utf8.RuneCountInString(trimmed) <= limit {
		return trimmed
	}
	runes := []rune(trimmed)
	return string(runes[:limit]) + "…"
}