package message

import (
	"context"
	"fmt"
	"strings"

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

const MaxVisibleScopesPerBatch = 64

var visibleScopeReadBatchQuery = `
SELECT
  candidate.id,
  candidate.type,
  COALESCE(parent_scope.type, ''),
  CASE
    WHEN candidate.type = 'channel' THEN EXISTS (
      SELECT 1
      FROM scope_memberships membership
      WHERE membership.scope_id = candidate.id
        AND membership.member_id = $2
        AND membership.removed_at IS NULL
        AND membership.effective_to_seq IS NULL
        AND (
          (
            membership.visibility = 'public'
            AND ($3 = '' OR membership.archived_at IS NULL)
          )
          OR (
            private_agent_owner.id IS NOT NULL
            AND membership.visibility = 'private'
            AND membership.owner_member_id = private_agent_owner.id
            AND membership.archived_at IS NULL
          )
        )
    )
    WHEN candidate.type = 'dm' THEN EXISTS (
      SELECT 1
      FROM scope_memberships membership
      WHERE membership.scope_id = candidate.id
        AND membership.member_id = $2
        AND membership.removed_at IS NULL
        AND membership.archived_at IS NULL
        AND membership.effective_to_seq IS NULL
    )
    WHEN candidate.type = 'task' THEN EXISTS (
      SELECT 1
      FROM tasks task
      JOIN task_boards task_board
        ON task_board.id = task.board_id
       AND task_board.workspace_id = $1
      JOIN scopes source_scope
        ON source_scope.id = task_board.channel_id
       AND source_scope.workspace_id = $1
       AND source_scope.type = 'channel'
       AND source_scope.status = 'active'
       AND source_scope.archived_at IS NULL
      JOIN scope_memberships membership
        ON membership.scope_id = source_scope.id
       AND membership.member_id = $2
      WHERE task.task_scope_id = candidate.id
        AND ` + taskLiveParentAuthoritySQL + `
    )
    WHEN candidate.type = 'thread' AND parent_scope.type = 'channel' THEN
      parent_profile.scope_id IS NOT NULL
      AND (parent_profile.slug = 'all' OR parent_profile.owner_member_id IS NOT NULL)
      AND EXISTS (
        SELECT 1
        FROM scope_memberships membership
        WHERE membership.scope_id = parent_scope.id
          AND membership.member_id = $2
          AND membership.removed_at IS NULL
          AND membership.effective_to_seq IS NULL
          AND (
            (
              membership.visibility = 'public'
              AND ($3 = '' OR membership.archived_at IS NULL)
            )
            OR (
              private_agent_owner.id IS NOT NULL
              AND membership.visibility = 'private'
              AND membership.owner_member_id = private_agent_owner.id
              AND membership.archived_at IS NULL
            )
          )
      )
    WHEN candidate.type = 'thread' AND parent_scope.type = 'dm' THEN
      EXISTS (
        SELECT 1
        FROM scope_memberships membership
        WHERE membership.scope_id IN (candidate.id, parent_scope.id)
          AND membership.member_id = $2
          AND membership.removed_at IS NULL
          AND membership.archived_at IS NULL
          AND membership.effective_to_seq IS NULL
      )
    WHEN candidate.type = 'thread' THEN FALSE
    WHEN candidate.type = 'workflow' THEN EXISTS (
      SELECT 1
      FROM workflow_plans plan
      JOIN members viewer
        ON viewer.workspace_id = plan.workspace_id
       AND viewer.id = $2
       AND viewer.type = 'human'
       AND viewer.status = 'active'
      WHERE plan.workspace_id = $1
        AND plan.scope_id = candidate.id
    )
    ELSE FALSE
  END AS readable
FROM scopes candidate
LEFT JOIN scopes parent_scope
  ON parent_scope.id = candidate.parent_scope_id
 AND parent_scope.workspace_id = candidate.workspace_id
 AND parent_scope.status = 'active'
LEFT JOIN scope_channel_profiles parent_profile
  ON parent_profile.scope_id = parent_scope.id
 AND parent_profile.workspace_id = parent_scope.workspace_id
LEFT JOIN agents private_agent
  ON private_agent.workspace_id = candidate.workspace_id
 AND private_agent.id = $3
 AND private_agent.member_id = $2
 AND private_agent.status <> 'archived'
LEFT JOIN machines private_agent_machine
  ON private_agent_machine.workspace_id = private_agent.workspace_id
 AND private_agent_machine.id = private_agent.machine_id
 AND private_agent_machine.status = 'active'
LEFT JOIN members private_agent_owner
  ON private_agent_owner.workspace_id = private_agent.workspace_id
 AND private_agent_owner.user_id = private_agent_machine.created_by_user_id
 AND private_agent_owner.type = 'human'
 AND private_agent_owner.status = 'active'
WHERE candidate.workspace_id = $1
  AND candidate.id IN (%s)
  AND candidate.status = 'active'
  AND candidate.archived_at IS NULL
`

// SQLVisibleScopeAuthorizer applies the same read gates used by history reads
// before a WebSocket connection can join a live-only visible scope.
type SQLVisibleScopeAuthorizer struct {
	queryer coredb.Queryer
}

// NewSQLVisibleScopeAuthorizer creates a database-backed visible-scope
// authorizer. Nil queryers deny all visible-scope subscriptions.
func NewSQLVisibleScopeAuthorizer(queryer coredb.Queryer) *SQLVisibleScopeAuthorizer {
	return &SQLVisibleScopeAuthorizer{queryer: queryer}
}

func (authorizer *SQLVisibleScopeAuthorizer) CanReadVisibleScope(ctx context.Context, workspaceID string, memberID string, scopeID string) (bool, error) {
	access, err := authorizer.CanReadVisibleScopes(ctx, workspaceID, memberID, []string{scopeID})
	if err != nil {
		return false, err
	}
	return access[strings.TrimSpace(scopeID)].Readable, nil
}

// VisibleScopeReadAccess is the active stored type and read decision for one
// requested scope. Missing or inactive scopes are omitted from a batch result.
type VisibleScopeReadAccess struct {
	ScopeType string
	Readable  bool
}

// CanReadVisibleScopes evaluates a bounded set in one SQL round trip. The CASE
// branches intentionally mirror the ordinary history gates used by
// CanReadVisibleScope: public Channel membership, active DM membership,
// parent-authorized reply threads, and Task source-Channel membership.
func (authorizer *SQLVisibleScopeAuthorizer) CanReadVisibleScopes(
	ctx context.Context,
	workspaceID string,
	memberID string,
	scopeIDs []string,
) (map[string]VisibleScopeReadAccess, error) {
	return authorizer.canReadVisibleScopes(ctx, workspaceID, memberID, "", scopeIDs)
}

// CanReadVisibleScopesForAgent extends the ordinary visible-scope gate with an
// authenticated Agent's active owner-private memberships. The private owner is
// derived from the exact active Agent and Machine instead of being
// caller-supplied. This is the batch read gate used for frozen Workflow
// relatedScopes; the Web subscription gate intentionally calls the public-only
// CanReadVisibleScopes method above.
func (authorizer *SQLVisibleScopeAuthorizer) CanReadVisibleScopesForAgent(
	ctx context.Context,
	workspaceID string,
	agentMemberID string,
	agentID string,
	scopeIDs []string,
) (map[string]VisibleScopeReadAccess, error) {
	return authorizer.canReadVisibleScopes(
		ctx,
		workspaceID,
		agentMemberID,
		strings.TrimSpace(agentID),
		scopeIDs,
	)
}

func (authorizer *SQLVisibleScopeAuthorizer) canReadVisibleScopes(
	ctx context.Context,
	workspaceID string,
	memberID string,
	agentID string,
	scopeIDs []string,
) (map[string]VisibleScopeReadAccess, error) {
	result := make(map[string]VisibleScopeReadAccess)
	if authorizer == nil || authorizer.queryer == nil {
		return result, nil
	}
	workspaceID = strings.TrimSpace(workspaceID)
	memberID = strings.TrimSpace(memberID)
	if workspaceID == "" || memberID == "" || len(scopeIDs) == 0 {
		return result, nil
	}
	if len(scopeIDs) > MaxVisibleScopesPerBatch {
		return nil, fmt.Errorf("visible scope batch must not contain more than %d scopes", MaxVisibleScopesPerBatch)
	}
	args, placeholders := visibleScopeReadBatchArgs(workspaceID, memberID, agentID, scopeIDs)
	if len(placeholders) == 0 {
		return result, nil
	}
	rows, err := authorizer.queryer.Query(
		ctx,
		fmt.Sprintf(visibleScopeReadBatchQuery, strings.Join(placeholders, ", ")),
		args...,
	)
	if err != nil {
		return nil, fmt.Errorf("authorize visible scope batch: %w", err)
	}
	defer rows.Close()
	return scanVisibleScopeReadBatch(rows)
}

func visibleScopeReadBatchArgs(
	workspaceID string,
	memberID string,
	agentID string,
	scopeIDs []string,
) ([]any, []string) {
	args := []any{workspaceID, memberID, agentID}
	placeholders := make([]string, 0, len(scopeIDs))
	seen := make(map[string]struct{}, len(scopeIDs))
	for _, rawScopeID := range scopeIDs {
		scopeID := strings.TrimSpace(rawScopeID)
		if scopeID == "" {
			continue
		}
		if _, duplicate := seen[scopeID]; duplicate {
			continue
		}
		seen[scopeID] = struct{}{}
		args = append(args, scopeID)
		placeholders = append(placeholders, fmt.Sprintf("$%d", len(args)))
	}
	return args, placeholders
}

func scanVisibleScopeReadBatch(rows coredb.Rows) (map[string]VisibleScopeReadAccess, error) {
	result := make(map[string]VisibleScopeReadAccess)
	for rows.Next() {
		var scopeID string
		var scopeType string
		var parentScopeType string
		var readable bool
		if err := rows.Scan(&scopeID, &scopeType, &parentScopeType, &readable); err != nil {
			return nil, fmt.Errorf("scan visible scope batch: %w", err)
		}
		scopeType = strings.TrimSpace(scopeType)
		parentScopeType = strings.TrimSpace(parentScopeType)
		if scopeType == "thread" && parentScopeType != "" &&
			parentScopeType != "channel" && parentScopeType != "dm" {
			return nil, fmt.Errorf("invalid visible thread parent scope type %q", parentScopeType)
		}
		result[strings.TrimSpace(scopeID)] = VisibleScopeReadAccess{
			ScopeType: scopeType,
			Readable:  readable,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate visible scope batch: %w", err)
	}
	return result, nil
}