package message

import (
	"context"
	"fmt"
	"strings"
	"time"

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

const (
	humanScopeSubscriptionStateActive = "active"
	humanScopeSubscriptionStateQuiet  = "quiet"
	humanScopeSubscriptionStateMute   = "mute"
)

func InsertThreadSubscriptionIfMissing(ctx context.Context, execer CommandTagExecer, memberID string, threadChannelID string, subscribedAt time.Time) error {
	return insertScopeNotificationIfMissingWithExecer(ctx, execer, memberID, threadChannelID, humanScopeSubscriptionStateActive, subscribedAt)
}

func normalizeHumanScopeSubscriptionState(state string) (string, error) {
	switch strings.TrimSpace(state) {
	case humanScopeSubscriptionStateActive:
		return humanScopeSubscriptionStateActive, nil
	case humanScopeSubscriptionStateQuiet:
		return humanScopeSubscriptionStateQuiet, nil
	case humanScopeSubscriptionStateMute:
		return humanScopeSubscriptionStateMute, nil
	default:
		return "", badRequest("subscription state must be active, quiet, or mute")
	}
}

func ActivateThreadSubscription(ctx context.Context, execer coredb.QueryExecer, memberID string, threadChannelID string, subscribedAt time.Time) error {
	return ActivateThreadSubscriptionWithInitialRead(ctx, execer, memberID, threadChannelID, 0, 0, subscribedAt)
}

func ActivateThreadSubscriptionWithInitialRead(ctx context.Context, execer coredb.QueryExecer, memberID string, threadChannelID string, initialReadSeq int64, initialReadGlobalSeq int64, subscribedAt time.Time) error {
	return ActivateThreadSubscriptionWithInitialReadAtSeq(ctx, execer, memberID, threadChannelID, initialReadSeq, initialReadGlobalSeq, 0, subscribedAt)
}

func ActivateThreadSubscriptionAtSeq(ctx context.Context, execer coredb.QueryExecer, memberID string, threadChannelID string, effectiveFromSeq int64, subscribedAt time.Time) error {
	return ActivateThreadSubscriptionWithInitialReadAtSeq(ctx, execer, memberID, threadChannelID, 0, 0, effectiveFromSeq, subscribedAt)
}

func ActivateThreadSubscriptionWithInitialReadAtSeq(ctx context.Context, execer coredb.QueryExecer, memberID string, threadChannelID string, initialReadSeq int64, initialReadGlobalSeq int64, effectiveFromSeq int64, subscribedAt time.Time) error {
	if _, err := upsertScopeThreadSubscriptionStateAtSeqWithExecer(ctx, execer, memberID, threadChannelID, humanScopeSubscriptionStateActive, initialReadSeq, initialReadGlobalSeq, effectiveFromSeq, subscribedAt); err != nil {
		return fmt.Errorf("activate thread scope subscription: %w", err)
	}
	return nil
}

func ActivateThreadSubscriptionIfNotExplicitlyUnsubscribed(ctx context.Context, execer coredb.QueryExecer, memberID string, threadChannelID string, subscribedAt time.Time) error {
	return ActivateThreadSubscriptionWithInitialReadIfNotExplicitlyUnsubscribedAtSeq(ctx, execer, memberID, threadChannelID, 0, 0, 0, subscribedAt)
}

func ActivateThreadSubscriptionIfNotExplicitlyUnsubscribedAtSeq(ctx context.Context, execer coredb.QueryExecer, memberID string, threadChannelID string, effectiveFromSeq int64, subscribedAt time.Time) error {
	return ActivateThreadSubscriptionWithInitialReadIfNotExplicitlyUnsubscribedAtSeq(ctx, execer, memberID, threadChannelID, 0, 0, effectiveFromSeq, subscribedAt)
}

func ActivateThreadSubscriptionWithInitialReadIfNotExplicitlyUnsubscribedAtSeq(ctx context.Context, execer coredb.QueryExecer, memberID string, threadChannelID string, initialReadSeq int64, initialReadGlobalSeq int64, effectiveFromSeq int64, subscribedAt time.Time) error {
	explicitlyUnsubscribed, err := isExplicitlyUnsubscribedFromThread(ctx, execer, memberID, threadChannelID)
	if err != nil {
		return err
	}
	if explicitlyUnsubscribed {
		return nil
	}
	return ActivateThreadSubscriptionWithInitialReadAtSeq(ctx, execer, memberID, threadChannelID, initialReadSeq, initialReadGlobalSeq, effectiveFromSeq, subscribedAt)
}

// EnsureReplyThreadParentDMAgentParticipant copies any agent peer from the
// parent DM into the reply thread as an auto-wake participant, unless the
// thread has been explicitly unsubscribed for that agent already.
func EnsureReplyThreadParentDMAgentParticipant(ctx context.Context, execer coredb.QueryExecer, threadChannelID string, participatedAt time.Time) error {
	memberIDs, err := loadReplyThreadParentDMAgentMemberIDs(ctx, execer, threadChannelID)
	if err != nil {
		return err
	}
	for _, memberID := range memberIDs {
		if err := EnsureReplyThreadParticipantProjection(ctx, execer, threadChannelID, memberID, "auto-wake", participatedAt); err != nil {
			return err
		}
	}
	return nil
}

func loadReplyThreadParentDMAgentMemberIDs(ctx context.Context, queryer coredb.Queryer, threadChannelID string) ([]string, error) {
	rows, err := queryer.Query(ctx, `
SELECT DISTINCT parent_membership.member_id
FROM scopes reply_scope
JOIN scopes parent_scope
  ON parent_scope.id = reply_scope.parent_scope_id
 AND parent_scope.workspace_id = reply_scope.workspace_id
JOIN scope_memberships parent_membership
  ON parent_membership.scope_id = parent_scope.id
 AND parent_membership.removed_at IS NULL
 AND parent_membership.archived_at IS NULL
 AND parent_membership.effective_to_seq IS NULL
JOIN members parent_member
  ON parent_member.id = parent_membership.member_id
 AND parent_member.type = 'agent'
WHERE reply_scope.id = $1
  AND reply_scope.type = 'thread'
  AND reply_scope.status = 'active'
  AND reply_scope.archived_at IS NULL
  AND parent_scope.type = 'dm'
  AND parent_scope.status = 'active'
  AND parent_scope.archived_at IS NULL
ORDER BY parent_membership.member_id ASC
`, strings.TrimSpace(threadChannelID))
	if err != nil {
		return nil, fmt.Errorf("load reply thread parent dm agent members: %w", err)
	}
	defer rows.Close()

	memberIDs := make([]string, 0, 1)
	for rows.Next() {
		var memberID string
		if err := rows.Scan(&memberID); err != nil {
			return nil, fmt.Errorf("scan reply thread parent dm agent member: %w", err)
		}
		memberID = strings.TrimSpace(memberID)
		if memberID == "" {
			continue
		}
		memberIDs = append(memberIDs, memberID)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate reply thread parent dm agent members: %w", err)
	}
	return memberIDs, nil
}

func isExplicitlyUnsubscribedFromThread(ctx context.Context, queryer coredb.QueryRower, memberID string, threadChannelID string) (bool, error) {
	trimmedMemberID := strings.TrimSpace(memberID)
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	if trimmedMemberID == "" || trimmedThreadChannelID == "" {
		return false, nil
	}
	row := queryer.QueryRow(ctx, `
SELECT removed_at
FROM scope_memberships
WHERE member_id = $1
  AND scope_id = $2
  AND removed_at IS NOT NULL
ORDER BY updated_at DESC
LIMIT 1
`, trimmedMemberID, trimmedThreadChannelID)
	var removedAt *time.Time
	if err := row.Scan(&removedAt); err != nil {
		if coredb.IsNotFound(err) {
			return false, nil
		}
		return false, fmt.Errorf("load thread unsubscribe state: %w", err)
	}
	return removedAt != nil, nil
}

func loadExplicitlyUnsubscribedThreadMemberIDSet(ctx context.Context, queryer coredb.Queryer, threadChannelID string, memberIDs []string) (map[string]struct{}, error) {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	normalizedMemberIDs := normalizeNonEmptyStrings(memberIDs)
	if trimmedThreadChannelID == "" || len(normalizedMemberIDs) == 0 {
		return map[string]struct{}{}, nil
	}

	requestedMemberIDsCTE, requestedMemberIDArgs, _, err := coredb.BuildStringValuesCTE("requested_member_ids", "member_id", normalizedMemberIDs, 2)
	if err != nil {
		return nil, fmt.Errorf("build thread unsubscribe member query: %w", err)
	}
	rows, err := queryer.Query(ctx, fmt.Sprintf(`
WITH %s
SELECT DISTINCT sm.member_id
FROM requested_member_ids requested
JOIN scope_memberships sm
  ON sm.member_id = requested.member_id
 AND sm.scope_id = $1
 AND sm.removed_at IS NOT NULL
`, requestedMemberIDsCTE), append([]any{trimmedThreadChannelID}, requestedMemberIDArgs...)...)
	if err != nil {
		return nil, fmt.Errorf("load thread unsubscribe member set: %w", err)
	}
	defer rows.Close()

	unsubscribed := make(map[string]struct{}, len(normalizedMemberIDs))
	for rows.Next() {
		var memberID string
		if err := rows.Scan(&memberID); err != nil {
			return nil, fmt.Errorf("scan thread unsubscribe member: %w", err)
		}
		trimmedMemberID := strings.TrimSpace(memberID)
		if trimmedMemberID == "" {
			continue
		}
		unsubscribed[trimmedMemberID] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate thread unsubscribe members: %w", err)
	}
	return unsubscribed, nil
}

func upsertScopeThreadSubscriptionStateWithExecer(ctx context.Context, execer coredb.QueryExecer, memberID string, scopeID string, state string, initialReadSeq int64, initialReadGlobalSeq int64, updatedAt time.Time) (string, error) {
	return upsertScopeThreadSubscriptionStateAtSeqWithExecer(ctx, execer, memberID, scopeID, state, initialReadSeq, initialReadGlobalSeq, 0, updatedAt)
}

func upsertScopeThreadSubscriptionStateAtSeqWithExecer(ctx context.Context, execer coredb.QueryExecer, memberID string, scopeID string, state string, initialReadSeq int64, initialReadGlobalSeq int64, effectiveFromSeq int64, updatedAt time.Time) (string, error) {
	trimmedMemberID := strings.TrimSpace(memberID)
	trimmedScopeID := strings.TrimSpace(scopeID)
	if trimmedMemberID == "" || trimmedScopeID == "" {
		return "", nil
	}
	if initialReadSeq < 0 {
		initialReadSeq = 0
	}
	if initialReadGlobalSeq < 0 {
		initialReadGlobalSeq = 0
	}
	if effectiveFromSeq < 0 {
		effectiveFromSeq = 0
	}
	normalizedState, err := normalizeScopeNotificationState(state)
	if err != nil {
		return "", err
	}
	actionAt := updatedAt.UTC()
	if actionAt.IsZero() {
		actionAt = time.Now().UTC()
	}
	var visibility, ownerMemberID, scopeNotificationState string
	fenceErr := withChannelReplyThreadFence(ctx, execer, trimmedScopeID, trimmedMemberID, func(writeExecer coredb.QueryExecer) error {
		// Resolve defaults under the parent lock (on the fence tx) so a concurrent
		// remover cannot leave the child write with stale owner/visibility.
		defaultsVisibility, defaultsOwner, defaultsState, err := scopeThreadMembershipDefaults(ctx, writeExecer, trimmedScopeID, trimmedMemberID, normalizedState)
		if err != nil {
			return err
		}
		visibility, ownerMemberID, scopeNotificationState = defaultsVisibility, defaultsOwner, defaultsState
		var hasOpenMembership int
		selectErr := writeExecer.QueryRow(ctx, `
SELECT 1
FROM scope_memberships
WHERE scope_id = $1
  AND member_id = $2
  AND effective_to_seq IS NULL
  AND removed_at IS NULL
  AND archived_at IS NULL
LIMIT 1
`, trimmedScopeID, trimmedMemberID).Scan(&hasOpenMembership)
		if selectErr != nil && !coredb.IsNotFound(selectErr) {
			return fmt.Errorf("load open scope thread membership: %w", selectErr)
		}
		if selectErr == nil {
			if _, err := writeExecer.Exec(ctx, `
UPDATE scope_memberships
SET unread_generation = CASE
        WHEN visibility <> $3 OR COALESCE(owner_member_id, '') <> $4 THEN unread_generation + 1
        ELSE unread_generation
    END,
    visibility = $3,
    owner_member_id = NULLIF($4, ''),
    authority_from_at = COALESCE(authority_from_at, created_at),
    updated_at = $5
WHERE scope_id = $1
  AND member_id = $2
  AND effective_to_seq IS NULL
  AND removed_at IS NULL
  AND archived_at IS NULL
`, trimmedScopeID, trimmedMemberID, visibility, ownerMemberID, actionAt); err != nil {
				return fmt.Errorf("update open scope thread membership: %w", err)
			}
		} else if _, err := writeExecer.Exec(ctx, `
INSERT INTO scope_memberships (
  scope_id, member_id, visibility, owner_member_id, last_read_seq,
  last_read_global_seq, last_activity_global_seq,
  owner_context_dismissed_through_global_seq, effective_from_seq,
  effective_to_seq, authority_from_at, authority_to_at,
  created_at, updated_at, removed_at, archived_at
) VALUES ($1, $2, $4, NULLIF($5, ''), $6, $7, 0, 0, $8, NULL, LEAST($3, CURRENT_TIMESTAMP(6)), NULL, $3, $3, NULL, NULL)
ON DUPLICATE KEY UPDATE
  unread_generation = CASE
    WHEN scope_memberships.visibility <> VALUES(visibility)
      OR COALESCE(scope_memberships.owner_member_id, '') <> COALESCE(VALUES(owner_member_id), '')
      OR scope_memberships.effective_to_seq IS NOT NULL
      OR scope_memberships.removed_at IS NOT NULL
      OR scope_memberships.archived_at IS NOT NULL THEN scope_memberships.unread_generation + 1
    ELSE scope_memberships.unread_generation
  END,
  visibility = VALUES(visibility),
  owner_member_id = VALUES(owner_member_id),
  effective_to_seq = NULL,
  authority_from_at = CASE
    WHEN scope_memberships.removed_at IS NOT NULL OR scope_memberships.archived_at IS NOT NULL OR scope_memberships.effective_to_seq IS NOT NULL THEN VALUES(authority_from_at)
    ELSE COALESCE(scope_memberships.authority_from_at, scope_memberships.created_at)
  END,
  authority_to_at = CASE
    WHEN scope_memberships.removed_at IS NOT NULL OR scope_memberships.archived_at IS NOT NULL OR scope_memberships.effective_to_seq IS NOT NULL THEN NULL
    ELSE scope_memberships.authority_to_at
  END,
  removed_at = NULL,
  archived_at = NULL,
  updated_at = VALUES(updated_at)
`, trimmedScopeID, trimmedMemberID, actionAt, visibility, ownerMemberID, initialReadSeq, initialReadGlobalSeq, effectiveFromSeq); err != nil {
			return fmt.Errorf("upsert scope thread membership: %w", err)
		}
		if _, err := writeExecer.Exec(ctx, `
INSERT INTO scope_notification_settings (
  scope_id, member_id, state, created_at, updated_at, unread_generation
) VALUES ($1, $2, $3, $4, $4, 1)
ON DUPLICATE KEY UPDATE
  unread_generation = CASE
      WHEN (CASE
        WHEN VALUES(state) = 'auto-wake' AND state IN ('auto-wake', 'no-wake') THEN state
        ELSE VALUES(state)
      END) <> state THEN unread_generation + 1
      ELSE unread_generation
    END,
  state = CASE
      WHEN VALUES(state) = 'auto-wake' AND state IN ('auto-wake', 'no-wake') THEN state
      ELSE VALUES(state)
    END,
  updated_at = VALUES(updated_at)
`, trimmedScopeID, trimmedMemberID, scopeNotificationState, actionAt); err != nil {
			return fmt.Errorf("upsert scope thread notification state: %w", err)
		}
		return nil
	})
	if fenceErr != nil {
		return "", fenceErr
	}
	return scopeNotificationState, nil
}

func normalizeScopeNotificationState(state string) (string, error) {
	switch strings.TrimSpace(state) {
	case "active", "quiet", "mute", "auto-wake", "no-wake":
		return strings.TrimSpace(state), nil
	default:
		return "", badRequest("scope notification state must be active, quiet, mute, auto-wake, or no-wake")
	}
}

func scopeThreadMembershipDefaults(ctx context.Context, queryer coredb.QueryRower, scopeID string, memberID string, state string) (string, string, string, error) {
	visibility := "public"
	ownerMemberID := ""
	err := queryer.QueryRow(ctx, `
SELECT parent_membership.visibility, COALESCE(parent_membership.owner_member_id, '')
FROM scopes child_scope
JOIN scope_memberships parent_membership
  ON parent_membership.scope_id = child_scope.parent_scope_id
 AND parent_membership.member_id = $2
 AND parent_membership.removed_at IS NULL
 AND parent_membership.archived_at IS NULL
 AND parent_membership.effective_to_seq IS NULL
WHERE child_scope.id = $1
  AND child_scope.parent_scope_id IS NOT NULL
LIMIT 1
`, strings.TrimSpace(scopeID), strings.TrimSpace(memberID)).Scan(&visibility, &ownerMemberID)
	if err != nil && !coredb.IsNotFound(err) {
		return "", "", "", fmt.Errorf("load parent scope membership: %w", err)
	}
	var memberType string
	err = queryer.QueryRow(ctx, `
SELECT type
FROM members
WHERE id = $1
LIMIT 1
`, strings.TrimSpace(memberID)).Scan(&memberType)
	if err != nil && !coredb.IsNotFound(err) {
		return "", "", "", fmt.Errorf("load subscription member type: %w", err)
	}
	notificationState := strings.TrimSpace(state)
	if strings.TrimSpace(memberType) == "agent" && notificationState == humanScopeSubscriptionStateActive {
		notificationState = "auto-wake"
	}
	return strings.TrimSpace(visibility), strings.TrimSpace(ownerMemberID), notificationState, nil
}

func upsertScopeNotificationStateWithExecer(ctx context.Context, execer CommandTagExecer, memberID string, scopeID string, state string, updatedAt time.Time) error {
	trimmedMemberID := strings.TrimSpace(memberID)
	trimmedScopeID := strings.TrimSpace(scopeID)
	if trimmedMemberID == "" || trimmedScopeID == "" {
		return nil
	}
	normalizedState := strings.TrimSpace(state)
	if normalizedState == "" {
		return nil
	}
	actionAt := updatedAt.UTC()
	if actionAt.IsZero() {
		actionAt = time.Now().UTC()
	}
	if _, err := execer.Exec(ctx, `
INSERT INTO scope_notification_settings (
  scope_id, member_id, state, created_at, updated_at, unread_generation
) VALUES ($1, $2, $3, $4, $4, 1)
ON DUPLICATE KEY UPDATE
  unread_generation = CASE
      WHEN state <> VALUES(state) THEN unread_generation + 1
      ELSE unread_generation
    END,
  updated_at = CASE
      WHEN state <> VALUES(state) THEN VALUES(updated_at)
      ELSE updated_at
    END,
  state = CASE
      WHEN state <> VALUES(state) THEN VALUES(state)
      ELSE state
    END
`, trimmedScopeID, trimmedMemberID, normalizedState, actionAt); err != nil {
		return fmt.Errorf("upsert scope notification state: %w", err)
	}
	return nil
}

func insertScopeNotificationIfMissingWithExecer(ctx context.Context, execer CommandTagExecer, memberID string, scopeID string, state string, createdAt time.Time) error {
	trimmedMemberID := strings.TrimSpace(memberID)
	trimmedScopeID := strings.TrimSpace(scopeID)
	if trimmedMemberID == "" || trimmedScopeID == "" {
		return nil
	}
	normalizedState, err := normalizeScopeNotificationState(state)
	if err != nil {
		return err
	}
	actionAt := createdAt.UTC()
	if actionAt.IsZero() {
		actionAt = time.Now().UTC()
	}
	if _, err := execer.Exec(ctx, `
INSERT IGNORE INTO scope_notification_settings (
  scope_id, member_id, state, created_at, updated_at
) VALUES ($1, $2, $3, $4, $4)
`, trimmedScopeID, trimmedMemberID, normalizedState, actionAt); err != nil {
		return fmt.Errorf("insert scope notification if missing: %w", err)
	}
	return nil
}

func UpsertThreadSubscription(ctx context.Context, execer CommandTagExecer, memberID string, threadChannelID string, subscribedAt time.Time, unsubscribedAt *time.Time) error {
	trimmedMemberID := strings.TrimSpace(memberID)
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	if trimmedMemberID == "" || trimmedThreadChannelID == "" {
		return nil
	}

	if unsubscribedAt != nil {
		unsubscribedUTC := unsubscribedAt.UTC()
		if _, err := execer.Exec(ctx, `
DELETE FROM scope_notification_settings
WHERE member_id = $1 AND scope_id = $2
`, trimmedMemberID, trimmedThreadChannelID); err != nil {
			return fmt.Errorf("delete thread scope notification: %w", err)
		}
		if _, err := execer.Exec(ctx, `
UPDATE scope_memberships
SET unread_generation = unread_generation + 1,
    removed_at = $3,
    updated_at = $3
WHERE member_id = $1
  AND scope_id = $2
  AND removed_at IS NULL
`, trimmedMemberID, trimmedThreadChannelID, unsubscribedUTC); err != nil {
			return fmt.Errorf("mark thread scope membership removed: %w", err)
		}
		return nil
	}
	return upsertScopeNotificationStateWithExecer(ctx, execer, trimmedMemberID, trimmedThreadChannelID, humanScopeSubscriptionStateActive, subscribedAt)
}