apps.py

import json
import logging
import os
import time
from html import escape
from datetime import datetime, timezone

import httpx
from typing import List, Optional
from urllib.parse import urlparse
from pydantic import BaseModel as PydanticBaseModel, ConfigDict, Field, ValidationError
from ulid import ULID
from fastapi import APIRouter, Body, Depends, Form, UploadFile, File, HTTPException, Header, Query
from fastapi.responses import HTMLResponse

from langchain_core.messages import SystemMessage, HumanMessage
from utils.apps import _clamp_review_score, fetch_app_chat_tools_from_manifest
from utils.executors import (
    critical_executor,
    db_executor,
    llm_executor,
    storage_executor,
    run_blocking,
    start_background_task,
)
from utils.http_client import get_webhook_client
from utils.multipart import APP_IMAGE_MAX_PART_SIZE, MultipartMaxPartSizeRoute, max_part_size
from utils.mcp_client import (
    discover_oauth_metadata,
    register_oauth_client,
    build_authorization_url,
    exchange_oauth_code,
    refresh_oauth_token,
    discover_mcp_tools,
    fetch_brandfetch_logo,
    generate_state_token,
    parse_state_token,
    generate_pkce_pair,
)

from database.apps import (
    change_app_approval_status,
    get_unapproved_public_apps_db,
    get_app_by_id_db,
    add_app_to_db,
    update_app_in_db,
    delete_app_from_db,
    update_app_visibility_in_db,
    get_personas_by_username_db,
    get_persona_by_id_db,
    delete_persona_db,
    get_persona_by_twitter_handle_db,
    get_persona_by_username_db,
    migrate_app_owner_id_db,
    get_user_persona_by_uid,
    get_omi_persona_apps_by_uid_db,
    create_api_key_db,
    list_api_keys_db,
    delete_api_key_db,
    set_app_popular_db,
    search_apps_db,
)
from database.webhook_health import clear_app_webhook_health
from database.auth import get_user_from_uid
from database.redis_db import (
    get_generic_cache,
    set_generic_cache,
    get_specific_user_review,
    increase_app_installs_count,
    decrease_app_installs_count,
    enable_app,
    disable_app,
    is_app_enabled,
    delete_app_cache_by_id,
    save_username,
    get_enabled_apps,
    get_conversation_summary_app_ids,
    add_conversation_summary_app_id,
    remove_conversation_summary_app_id,
    get_apps_installs_count,
    get_apps_reviews,
)
from utils.apps import (
    get_available_apps,
    get_available_app_by_id,
    get_approved_available_apps,
    invalidate_approved_apps_cache,
    invalidate_popular_apps_cache,
    get_available_app_by_id_with_reviews,
    set_app_review,
    get_app_reviews,
    add_tester,
    is_tester,
    add_app_access_for_tester,
    remove_app_access_for_tester,
    upsert_app_payment_link,
    get_is_user_paid_app,
    is_permit_payment_plan_get,
    generate_persona_prompt,
    generate_persona_desc,
    get_persona_by_uid,
    increment_username,
    generate_api_key,
    get_popular_apps,
    paginate_apps,
    build_pagination_metadata,
    get_capabilities_list,
    normalize_app_numeric_fields,
    filter_apps_by_capability,
    sort_apps_by_installs,
    group_apps_by_capability,
    build_capability_groups_response,
    group_capability_apps_by_category,
    build_capability_category_groups_response,
    validate_app_endpoints_for_reenable,
)

from database.memories import migrate_memories

from utils.llm.persona import generate_persona_intro_message
from utils.llm.app_generator import generate_description, generate_description_and_emoji
from utils.llm.app_generation_prompts import app_generation_prompts_from_llm_payload, app_generation_prompts_response
from utils.subscription import enforce_chat_quota
from utils.llm.usage_tracker import track_usage, Features
from utils.notifications import send_notification, send_app_review_reply_notification, send_new_app_review_notification
from utils.other import endpoints as auth
from utils.request_validation import (
    backfill_app_home_url_from_auth_steps,
    normalize_required_webhook_url,
    parse_form_json,
)
from models.app import (
    App,
    ActionType,
    AppCreate,
    AppUpdate,
    AppBaseModel,
    AppReview,
    AppCatalogItem,
)
from utils.other.storage import upload_app_logo, delete_app_logo, upload_app_thumbnail, get_app_thumbnail_url
from utils.social import (
    get_twitter_profile,
    verify_latest_tweet,
    upsert_persona_from_twitter_profile,
    add_twitter_to_persona,
)

logger = logging.getLogger(__name__)

router = APIRouter(route_class=MultipartMaxPartSizeRoute)


class AppSelectOption(PydanticBaseModel):
    title: str
    id: str


class AppCapabilityAction(AppSelectOption):
    doc_url: Optional[str] = None
    description: Optional[str] = None


class AppCapabilityResponse(AppSelectOption):
    triggers: List[AppSelectOption] = Field(default_factory=list)
    actions: List[AppCapabilityAction] = Field(default_factory=list)
    scopes: List[AppSelectOption] = Field(default_factory=list)


class AppThumbnailUploadResponse(PydanticBaseModel):
    thumbnail_url: str
    thumbnail_id: str


class AppMutationResponse(PydanticBaseModel):
    status: str


class AppStatusMessageResponse(AppMutationResponse):
    message: str


class AppManifestRefreshResponse(AppMutationResponse):
    tools_count: int = 0


class AppCreateResponse(AppMutationResponse):
    app_id: str


class AppMigrationResponse(AppMutationResponse):
    message: str


class McpAddServerResponse(PydanticBaseModel):
    app_id: str
    requires_oauth: bool
    auth_url: Optional[str] = None
    tools_count: Optional[int] = None
    tool_names: List[str] = Field(default_factory=list)


class McpRefreshToolsResponse(PydanticBaseModel):
    tools_count: int
    tool_names: List[str] = Field(default_factory=list)


class AppTesterCheckResponse(PydanticBaseModel):
    is_tester: bool


class UnapprovedPublicAppResponse(PydanticBaseModel):
    model_config = ConfigDict(extra='allow')

    id: Optional[str] = None
    name: Optional[str] = None
    uid: Optional[str] = None
    private: Optional[bool] = None
    approved: Optional[bool] = None
    status: Optional[str] = None
    category: Optional[str] = None
    author: Optional[str] = None
    description: Optional[str] = None
    image: Optional[str] = None
    capabilities: List[str] = Field(default_factory=list)


class AppDescriptionGenerationResponse(PydanticBaseModel):
    description: str


class AppDescriptionEmojiGenerationResponse(PydanticBaseModel):
    description: str
    emoji: str


class AppPromptsGenerationResponse(PydanticBaseModel):
    prompts: List[str]


class AppDraftGenerationResponse(PydanticBaseModel):
    name: str
    description: str
    category: str
    capabilities: List[str]
    chat_prompt: Optional[str] = None
    memory_prompt: Optional[str] = None


class AppGenerationResponse(PydanticBaseModel):
    status: str
    app: AppDraftGenerationResponse


class AppIconGenerationResponse(PydanticBaseModel):
    status: str
    icon_base64: str
    mime_type: str


class AppPaginationLinks(PydanticBaseModel):
    next: Optional[str] = None
    previous: Optional[str] = None


class AppPagination(PydanticBaseModel):
    total: int
    count: int
    offset: int
    limit: int
    hasNext: bool
    hasPrevious: bool
    links: Optional[AppPaginationLinks] = None


class AppCatalogGroup(PydanticBaseModel):
    capability: Optional[AppSelectOption] = None
    category: Optional[AppSelectOption] = None
    data: List[AppCatalogItem] = Field(default_factory=list)
    pagination: Optional[AppPagination] = None
    count: Optional[int] = None


class AppCatalogMeta(PydanticBaseModel):
    capabilities: List[AppSelectOption] = Field(default_factory=list)
    groupCount: int = 0
    limit: Optional[int] = None
    offset: Optional[int] = None
    totalApps: Optional[int] = None


class AppCatalogResponse(PydanticBaseModel):
    data: List[AppCatalogItem] = Field(default_factory=list)
    pagination: Optional[AppPagination] = None
    capability: Optional[AppSelectOption] = None
    category: Optional[AppSelectOption] = None
    groups: List[AppCatalogGroup] = Field(default_factory=list)
    meta: Optional[AppCatalogMeta] = None


class AppSearchFilters(PydanticBaseModel):
    query: Optional[str] = None
    category: Optional[str] = None
    rating: Optional[float] = None
    capability: Optional[str] = None
    sort: str
    my_apps: Optional[bool] = None
    installed_apps: Optional[bool] = None


class AppSearchResponse(PydanticBaseModel):
    data: List[AppCatalogItem] = Field(default_factory=list)
    pagination: AppPagination
    filters: AppSearchFilters


class AppApiKeyResponse(PydanticBaseModel):
    id: str
    label: str
    created_at: Optional[datetime] = None
    secret: Optional[str] = None


class PersonaMutationResponse(AppMutationResponse):
    app_id: str
    username: str


class TwitterProfileResponse(PydanticBaseModel):
    name: str
    profile: str
    rest_id: str
    avatar: str
    desc: str
    friends: int
    sub_count: int
    id: str
    status: str
    persona_id: Optional[str] = None
    persona_username: Optional[str] = None


class TwitterOwnershipVerificationResponse(PydanticBaseModel):
    tweet: str
    verified: bool
    persona_id: Optional[str] = None


class TwitterInitialMessageResponse(PydanticBaseModel):
    message: str


class ConversationSummaryAppIdsResponse(PydanticBaseModel):
    app_ids: List[str] = Field(default_factory=list)


class PersonaRecordResponse(App):
    doc_id: Optional[str] = None


# ******************************************************
# ******************* REQUEST MODELS *******************
# ******************************************************


class ReviewAppRequest(PydanticBaseModel):
    score: float
    review: Optional[str] = None
    username: Optional[str] = None
    response: Optional[str] = None


class ReplyToReviewRequest(PydanticBaseModel):
    reviewer_uid: str
    response: str


class GenerateDescriptionRequest(PydanticBaseModel):
    name: str
    description: str


class GenerateDescriptionEmojiRequest(PydanticBaseModel):
    name: str
    prompt: str


class GenerateAppRequest(PydanticBaseModel):
    prompt: str = ''


class GenerateAppIconRequest(PydanticBaseModel):
    name: str = ''
    description: str = ''
    category: str = 'other'


class AddTesterRequest(PydanticBaseModel):
    model_config = ConfigDict(extra='allow')

    uid: str
    apps: List[str]


class TesterAccessRequest(PydanticBaseModel):
    uid: str
    app_id: str


def _write_file(path: str, data: bytes):
    """Write bytes to file — offloaded to storage_executor."""
    with open(path, 'wb') as f:
        f.write(data)


def _process_chat_tools_manifest(external_integration: dict, app_dict: dict) -> dict:
    """Fetch and process chat tools manifest, updating and returning app_dict.

    Fetches the manifest from chat_tools_manifest_url, resolves relative endpoints
    to absolute URLs using app_home_url, and stores chat_messages config.

    Args:
        external_integration: The external_integration dict from app data
        app_dict: The app dict to update with chat_tools and chat_messages config

    Returns:
        The updated app_dict
    """
    manifest_url = external_integration.get('chat_tools_manifest_url')
    if not manifest_url:
        return app_dict

    manifest_result = fetch_app_chat_tools_from_manifest(manifest_url)
    if not manifest_result:
        return app_dict

    fetched_tools = manifest_result.get('tools')
    if fetched_tools:
        # Resolve relative endpoints to absolute URLs
        base_url = (external_integration.get('app_home_url') or '').rstrip('/')
        if base_url:
            for tool in fetched_tools:
                endpoint = tool.get('endpoint', '')
                if endpoint.startswith('/') and not endpoint.startswith('//'):
                    tool['endpoint'] = f"{base_url}{endpoint}"
        app_dict['chat_tools'] = fetched_tools

    # Store chat_messages config in external_integration
    chat_messages = manifest_result.get('chat_messages')
    if 'external_integration' not in app_dict:
        app_dict['external_integration'] = {}
    if chat_messages:
        app_dict['external_integration']['chat_messages_enabled'] = chat_messages.get('enabled', False)
        app_dict['external_integration']['chat_messages_target'] = chat_messages.get('target', 'app')
        app_dict['external_integration']['chat_messages_notify'] = chat_messages.get('notify', False)
    else:
        # Reset all chat_messages fields to defaults when not in manifest
        app_dict['external_integration']['chat_messages_enabled'] = False
        app_dict['external_integration']['chat_messages_target'] = 'app'
        app_dict['external_integration']['chat_messages_notify'] = False

    return app_dict


def _get_categories():
    return [
        {'title': 'Popular', 'id': 'popular'},
        {'title': 'Conversation Analysis', 'id': 'conversation-analysis'},
        {'title': 'Personality Clone', 'id': 'personality-emulation'},
        {'title': 'Health', 'id': 'health-and-wellness'},
        {'title': 'Education', 'id': 'education-and-learning'},
        {'title': 'Communication', 'id': 'communication-improvement'},
        {'title': 'Emotional Support', 'id': 'emotional-and-mental-support'},
        {'title': 'Productivity', 'id': 'productivity-and-organization'},
        {'title': 'Entertainment', 'id': 'entertainment-and-fun'},
        {'title': 'Financial', 'id': 'financial'},
        {'title': 'Travel', 'id': 'travel-and-exploration'},
        {'title': 'Safety', 'id': 'safety-and-security'},
        {'title': 'Shopping', 'id': 'shopping-and-commerce'},
        {'title': 'Social', 'id': 'social-and-relationships'},
        {'title': 'News', 'id': 'news-and-information'},
        {'title': 'Utilities', 'id': 'utilities-and-tools'},
        {'title': 'Other', 'id': 'other'},
    ]


# ******************************************************
# ********************* APPS CRUD **********************
# ******************************************************


@router.get('/v1/apps', tags=['v1'], response_model=List[AppBaseModel])
def get_apps(uid: str = Depends(auth.get_current_user_uid), include_reviews: bool = True):
    apps = get_available_apps(uid, include_reviews=include_reviews)
    return [normalize_app_numeric_fields(app.to_reduced_dict()) for app in apps]


@router.get('/v1/apps/enabled', tags=['v1'], response_model=List[str])
def get_user_enabled_apps(uid: str = Depends(auth.get_current_user_uid)):
    """Returns the list of app IDs the user has enabled/installed."""
    return get_enabled_apps(uid)


@router.get('/v2/apps', tags=['v2'], response_model=AppCatalogResponse)
def get_apps_v2(
    capability: str | None = Query(default=None, description='Filter by capability id'),
    category: str | None = Query(default=None, description='Filter by category id'),
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=20, ge=1, le=100),
    include_reviews: bool = Query(default=False),
):
    """Public omi apps, paginated by capability groups.

    Notes:
    - Uses approved public apps only (no private/tester apps).
    - Groups: Popular, Integrations, Chat Assistants, Summary Apps, Realtime Notifications.
    - Popular section is shown first.
    - Always excludes persona type apps.
    """

    capabilities = get_capabilities_list()

    if capability:
        cache_key = f"apps:capability:v2:{capability}:offset={offset}:limit={limit}:reviews={int(include_reviews)}"
    elif category:
        cache_key = f"apps:category:v2:{category}:offset={offset}:limit={limit}:reviews={int(include_reviews)}"
    else:
        cache_key = f"apps:capability_groups:v2:offset={offset}:limit={limit}:reviews={int(include_reviews)}"

    cached = get_generic_cache(cache_key)
    if cached:
        return cached

    # Fetch and filter approved public apps
    apps = get_approved_available_apps(include_reviews=include_reviews)
    approved_apps = [a for a in apps if a.approved and (a.private is None or not a.private)]
    # Always exclude persona type apps
    approved_apps = [a for a in approved_apps if not a.is_a_persona()]

    # Capability-specific response
    if capability:
        filtered_apps = filter_apps_by_capability(approved_apps, capability)
        sorted_apps = sort_apps_by_installs(filtered_apps)
        page = paginate_apps(sorted_apps, offset, limit)

        res = {
            'data': [normalize_app_numeric_fields(app.to_reduced_dict()) for app in page],
            'pagination': build_pagination_metadata(len(sorted_apps), offset, limit, capability),
            'capability': {
                'id': capability,
                'title': next(
                    (c['title'] for c in capabilities if c['id'] == capability), capability.title().replace('_', ' ')
                ),
            },
        }
        set_generic_cache(cache_key, res, ttl=60 * 10)
        return res

    if category:
        filtered_apps = [app for app in approved_apps if app.category == category]
        sorted_apps = sort_apps_by_installs(filtered_apps)
        page = paginate_apps(sorted_apps, offset, limit)
        categories = _get_categories()

        res = {
            'data': [normalize_app_numeric_fields(app.to_reduced_dict()) for app in page],
            'pagination': build_pagination_metadata(len(sorted_apps), offset, limit, category),
            'category': {
                'id': category,
                'title': next(
                    (c['title'] for c in categories if c['id'] == category), category.title().replace('-', ' ')
                ),
            },
        }
        set_generic_cache(cache_key, res, ttl=60 * 10)
        return res

    # Grouped response by capability
    grouped_apps = group_apps_by_capability(approved_apps, capabilities)
    groups = build_capability_groups_response(grouped_apps, capabilities, offset, limit)

    res = {
        'groups': groups,
        'meta': {
            'capabilities': capabilities,
            'groupCount': len(groups),
            'limit': limit,
            'offset': offset,
        },
    }
    set_generic_cache(cache_key, res, ttl=60 * 10)
    return res


@router.get('/v2/apps/capability/{capability_id}/grouped', tags=['v2'], response_model=AppCatalogResponse)
def get_capability_apps_grouped_by_category(
    capability_id: str,
    include_reviews: bool = Query(default=True),
):
    """Get all apps for a specific capability, grouped by master category.

    Returns apps grouped into master categories like:
    - For chat: Personality Clones, Productivity & Lifestyle, Social & Entertainment
    - For others: Productivity & Tools, Personal & Lifestyle, Social & Entertainment
    """

    cache_key = f"apps:capability:{capability_id}:grouped:reviews={int(include_reviews)}"

    cached = get_generic_cache(cache_key)
    if cached:
        return cached

    capabilities = get_capabilities_list()

    # Fetch and filter approved public apps
    apps = get_approved_available_apps(include_reviews=include_reviews)
    approved_apps = [a for a in apps if a.approved and (a.private is None or not a.private)]
    # Always exclude persona type apps
    approved_apps = [a for a in approved_apps if not a.is_a_persona()]

    # Filter apps by capability
    filtered_apps = filter_apps_by_capability(approved_apps, capability_id)

    # Group filtered apps by master category
    grouped_apps = group_capability_apps_by_category(filtered_apps, capability_id)
    groups = build_capability_category_groups_response(grouped_apps, capability_id)

    res = {
        'groups': groups,
        'capability': {
            'id': capability_id,
            'title': next(
                (c['title'] for c in capabilities if c['id'] == capability_id),
                capability_id.title().replace('_', ' '),
            ),
        },
        'meta': {
            'totalApps': len(filtered_apps),
            'groupCount': len(groups),
        },
    }
    set_generic_cache(cache_key, res, ttl=60 * 10)
    return res


def _matches_search_text(app: App, query: str) -> bool:
    """Whether `app` matches a lowercased search query.

    Name *or* description — the contract the `q` parameter documents, and the same fields the
    clients' offline fallback ranks over (desktop `appRanking.ts`). Matching the name alone made
    the remote endpoint strictly narrower than that fallback: an app found offline by a word in
    its description returned "No apps found" once the endpoint answered.
    """
    return query in app.name.lower() or query in (app.description or '').lower()


def _name_match_tier(app: App, query: str) -> int:
    """Relevance tier for a search hit: 0 exact name, 1 name prefix, 2 matched elsewhere.

    Mirrors `nameMatchTier` in the desktop client so both orderings agree.
    """
    name = app.name.lower()
    if name == query:
        return 0
    if name.startswith(query):
        return 1
    return 2


@router.get('/v2/apps/search', tags=['v2'], response_model=AppSearchResponse)
def search_apps(
    q: str | None = Query(default=None, description='Search query for app name or description'),
    category: str | None = Query(default=None, description='Filter by category id'),
    rating: float | None = Query(default=None, ge=0, le=5, description='Minimum rating filter'),
    capability: str | None = Query(default=None, description='Filter by capability id'),
    sort: str | None = Query(
        default=None, description='Sort order: installs, rating_asc, rating_desc, name_asc, name_desc'
    ),
    my_apps: bool | None = Query(default=None, description='Filter to show only user\'s apps'),
    installed_apps: bool | None = Query(default=None, description='Filter to show only installed/enabled apps'),
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=20, ge=1, le=100),
    uid: str = Depends(auth.get_current_user_uid),
):
    """Search and filter apps with pagination.

    Returns a flat list of apps matching the search and filter criteria.
    """

    enabled_app_ids = None
    if installed_apps:
        enabled_app_ids = list(get_enabled_apps(uid))

    apps_data = search_apps_db(
        uid=uid,
        category=category,
        capability=capability,
        my_apps=my_apps or False,
        installed_apps=installed_apps or False,
        enabled_app_ids=enabled_app_ids,
    )

    user_enabled = set(get_enabled_apps(uid))

    # Drop any malformed record missing an id before enrichment: id drives the installs/reviews/
    # enabled lookups below and the pre-loop app_ids list, so a missing id would KeyError before the
    # per-record ValidationError guard can catch it.
    valid_apps_data = [a for a in apps_data if a.get('id')]
    skipped_no_id = len(apps_data) - len(valid_apps_data)
    if skipped_no_id:
        logger.warning("Skipping %d malformed app record(s) without an id in search results", skipped_no_id)
    apps_data = valid_apps_data

    app_ids = [app['id'] for app in apps_data]
    apps_installs = get_apps_installs_count(app_ids)
    apps_reviews = get_apps_reviews(app_ids)

    apps = []

    for app_dict in apps_data:
        app_dict['enabled'] = app_dict['id'] in user_enabled
        app_dict['rejected'] = app_dict.get('approved') is False
        app_dict['installs'] = apps_installs.get(app_dict['id'], 0)

        # Calculate average from reviews
        reviews = apps_reviews.get(app_dict['id'], {})
        scores = [_clamp_review_score(x['score']) for x in reviews.values()]
        app_dict['rating_avg'] = sum(scores) / len(scores) if scores else None
        app_dict['rating_count'] = len(scores)

        # Skip a malformed/legacy app document rather than 500 the whole search page.
        try:
            apps.append(App(**app_dict))
        except ValidationError as e:
            logger.warning(
                "Skipping malformed app %s in search results: %s",
                app_dict.get('id'),
                [err['loc'][0] for err in e.errors() if err.get('loc')],
            )

    # Always exclude persona type apps from results
    filtered_apps = [app for app in apps if not app.is_a_persona()]

    # Apply text search filter
    if q and q.strip():
        search_query = q.strip().lower()
        filtered_apps = [app for app in filtered_apps if _matches_search_text(app, search_query)]

    # Apply rating filter
    if rating is not None:
        filtered_apps = [app for app in filtered_apps if (app.rating_avg or 0) >= rating]

    # Apply sorting
    if sort == 'rating_desc':
        filtered_apps = sorted(filtered_apps, key=lambda a: (a.rating_avg or 0), reverse=True)
    elif sort == 'rating_asc':
        filtered_apps = sorted(filtered_apps, key=lambda a: (a.rating_avg or 0))
    elif sort == 'name_asc':
        filtered_apps = sorted(filtered_apps, key=lambda a: a.name.lower())
    elif sort == 'name_desc':
        filtered_apps = sorted(filtered_apps, key=lambda a: a.name.lower(), reverse=True)
    elif sort == 'installs_desc':
        filtered_apps = sorted(filtered_apps, key=lambda a: (a.installs or 0), reverse=True)
    else:
        # sort by installs when searching, otherwise by name
        if q and q.strip():
            search_query = q.strip().lower()
            # Name matches rank above description-only matches before popularity: results are
            # paginated, so an exact-name app must not be pushed off page 1 by a more-installed
            # app that only mentions the query in its description.
            filtered_apps = sorted(filtered_apps, key=lambda a: (_name_match_tier(a, search_query), -(a.installs or 0)))
        else:
            filtered_apps = sorted(filtered_apps, key=lambda a: a.name.lower())

    # Paginate results
    total = len(filtered_apps)
    page = paginate_apps(filtered_apps, offset, limit)

    return {
        'data': [normalize_app_numeric_fields(app.to_reduced_dict()) for app in page],
        'pagination': build_pagination_metadata(total, offset, limit),
        'filters': {
            'query': q,
            'category': category,
            'rating': rating,
            'capability': capability,
            'sort': sort or 'name',
            'my_apps': my_apps,
            'installed_apps': installed_apps,
        },
    }


@router.get('/v1/approved-apps', tags=['v1'], response_model=List[AppBaseModel])
def get_approved_apps(include_reviews: bool = False):
    apps = get_approved_available_apps(include_reviews=include_reviews)
    # Always exclude persona type apps
    filtered_apps = [app for app in apps if not app.is_a_persona()]
    return [normalize_app_numeric_fields(app.to_reduced_dict()) for app in filtered_apps]


@router.get('/v1/apps/popular', tags=['v1'], response_model=List[AppBaseModel])
def get_popular_apps_endpoint(uid: str = Depends(auth.get_current_user_uid)):
    apps = get_popular_apps()
    # Always exclude persona type apps
    filtered_apps = [app for app in apps if not app.is_a_persona()]
    return [normalize_app_numeric_fields(app.to_reduced_dict()) for app in filtered_apps]


@router.post('/v1/apps', tags=['v1'], response_model=AppCreateResponse)
@max_part_size(APP_IMAGE_MAX_PART_SIZE)
def create_app(app_data: str = Form(...), file: UploadFile = File(...), uid=Depends(auth.get_current_user_uid)):
    data = parse_form_json(dict, app_data, 'app_data')
    data['approved'] = False
    data['status'] = 'under-review'
    data['name'] = (data.get('name') or '').strip()
    data['id'] = str(ULID())
    data['uid'] = uid
    if not data.get('author') and not data.get('email'):
        user = get_user_from_uid(uid) or {}
        email = user.get('email')
        # author is required + non-null on AppCreate; display_name/email can both be null.
        data['author'] = user.get('display_name') or (email.split('@')[0] if email else None) or 'Anonymous'
        data['email'] = email
    if not data.get('is_paid'):
        data['is_paid'] = False
    else:
        if data['is_paid'] is True:
            if data.get('price') is None:
                raise HTTPException(status_code=422, detail='App price is required')
            if data.get('price') < 0.0:
                raise HTTPException(status_code=422, detail='Price cannot be a negative value')
            if data.get('payment_plan') is None:
                raise HTTPException(status_code=422, detail='Payment plan is required')

    if external_integration := data.get('external_integration'):
        if external_integration.get('triggers_on') is None and len(external_integration.get('actions', [])) == 0:
            raise HTTPException(status_code=422, detail='Triggers on or actions is required')
        # Trigger on
        if external_integration.get('triggers_on'):
            normalize_required_webhook_url(external_integration)
            if external_integration.get('setup_instructions_file_path'):
                external_integration['setup_instructions_file_path'] = external_integration[
                    'setup_instructions_file_path'
                ].strip()
                if external_integration['setup_instructions_file_path'].startswith('http'):
                    external_integration['is_instructions_url'] = True
                else:
                    external_integration['is_instructions_url'] = False

        # Actions
        if actions := external_integration.get('actions'):
            for action in actions:
                if not action.get('action'):
                    raise HTTPException(status_code=422, detail='Action field is required for each action')
                if action.get('action') not in [action_type.value for action_type in ActionType]:
                    raise HTTPException(
                        status_code=422,
                        detail=f'Unsupported action type. Supported types: {", ".join([action_type.value for action_type in ActionType])}',
                    )
    os.makedirs('_temp/apps', exist_ok=True)
    file_path = f"_temp/apps/{file.filename}"
    with open(file_path, 'wb') as f:
        f.write(file.file.read())
    img_url = upload_app_logo(file_path, data['id'])
    data['image'] = img_url
    data['created_at'] = datetime.now(timezone.utc)
    # Backward compatibility: Set app_home_url from first auth step if not provided
    if 'external_integration' in data:
        backfill_app_home_url_from_auth_steps(data['external_integration'])

    try:
        app = AppCreate.model_validate(data)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=str(e))

    # Build app dict
    app_dict = app.model_dump(exclude_unset=True)

    # Fetch chat tools from manifest URL (only way to add chat tools)
    if external_integration := data.get('external_integration'):
        app_dict = _process_chat_tools_manifest(external_integration, app_dict)

    add_app_to_db(app_dict)

    # payment link
    upsert_app_payment_link(app.id, app.is_paid, app.price, app.payment_plan, app.uid)

    return {'status': 'ok', 'app_id': app.id}


@router.post('/v1/personas', tags=['v1'], response_model=PersonaMutationResponse)
@max_part_size(APP_IMAGE_MAX_PART_SIZE)
async def create_persona(
    persona_data: str = Form(...), file: UploadFile = File(...), uid=Depends(auth.get_current_user_uid)
):
    data = parse_form_json(dict, persona_data, 'persona_data')
    data['approved'] = False
    data['status'] = 'under-review'
    data['category'] = 'personality-emulation'
    data['name'] = (data.get('name') or '').strip()
    data['id'] = str(ULID())
    data['uid'] = uid
    data['capabilities'] = ['persona']
    user = await run_blocking(db_executor, get_user_from_uid, uid) or {}
    data['author'] = user.get('display_name', '')
    data['email'] = user.get('email')

    if 'username' not in data or data['username'] == '' or data['username'] is None:
        data['username'] = data['name'].replace(' ', '').lower()
        data['username'] = await run_blocking(db_executor, increment_username, data['username'])
    await run_blocking(db_executor, save_username, data['username'], uid)

    if 'connected_accounts' not in data or data['connected_accounts'] is None:
        data['connected_accounts'] = ['omi']
    data['persona_prompt'] = await generate_persona_prompt(uid, data)
    data['description'] = await run_blocking(llm_executor, generate_persona_desc, uid, data['name'])
    os.makedirs('_temp/apps', exist_ok=True)
    file_path = f"_temp/apps/{file.filename}"
    contents = await file.read()
    await run_blocking(storage_executor, _write_file, file_path, contents)
    img_url = await run_blocking(storage_executor, upload_app_logo, file_path, data['id'])
    data['image'] = img_url
    data['created_at'] = datetime.now(timezone.utc)

    try:
        app_create = AppCreate.model_validate(data)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=str(e))

    await run_blocking(db_executor, add_app_to_db, app_create.model_dump(exclude_unset=True))

    return {'status': 'ok', 'app_id': data['id'], 'username': data['username']}


@router.patch('/v1/personas/{persona_id}', tags=['v1'], response_model=PersonaMutationResponse)
@max_part_size(APP_IMAGE_MAX_PART_SIZE)
async def update_persona(
    persona_id: str,
    persona_data: str = Form(...),
    file: UploadFile = File(None),
    uid=Depends(auth.get_current_user_uid),
):
    data = parse_form_json(dict, persona_data, 'persona_data')
    persona = await run_blocking(db_executor, get_available_app_by_id, persona_id, uid)
    if not persona:
        raise HTTPException(status_code=404, detail='Persona not found')
    if persona['uid'] != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')

    # Image
    if file:
        if (
            'image' in persona
            and len(persona['image']) > 0
            and persona['image'].startswith('https://storage.googleapis.com/')
        ):
            await run_blocking(storage_executor, delete_app_logo, persona['image'])
        os.makedirs('_temp/apps', exist_ok=True)
        file_path = f"_temp/apps/{file.filename}"
        contents = await file.read()
        await run_blocking(storage_executor, _write_file, file_path, contents)
        img_url = await run_blocking(storage_executor, upload_app_logo, file_path, persona_id)
        data['image'] = img_url

    # Partial update: released clients PATCH the whole persona, but a client that
    # sends only the fields it changed must not have the rest silently rewritten.
    # `username` claims the handle, and `name` drives an LLM description rewrite —
    # both are destructive to do on a field the caller never mentioned.
    if 'username' in data and data['username'] and data['username'] != persona.get('username'):
        await run_blocking(db_executor, save_username, data['username'], uid)

    if 'name' in data and data['name'] and data['name'] != persona.get('name'):
        # The name changed, so the generated description no longer matches it,
        # unless the caller supplied its own.
        if 'description' not in data:
            data['description'] = await run_blocking(llm_executor, generate_persona_desc, uid, data['name'])

    # AppUpdate needs the identity fields even when the caller omitted them.
    data['id'] = persona_id
    data['updated_at'] = datetime.now(timezone.utc)

    # Update 'omi' connected_accounts
    if 'omi' in data.get('connected_accounts', []) and 'omi' not in persona.get('connected_accounts', []):
        data['persona_prompt'] = await generate_persona_prompt(uid, persona)

    try:
        update_app = AppUpdate.model_validate(data)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=str(e))

    await run_blocking(db_executor, update_app_in_db, update_app.model_dump(exclude_unset=True))

    if persona['approved'] and (persona['private'] is None or persona['private'] is False):
        await run_blocking(db_executor, invalidate_approved_apps_cache)
    await run_blocking(db_executor, delete_app_cache_by_id, persona_id)
    username = data.get('username', persona.get('username'))
    return {'status': 'ok', 'app_id': persona_id, 'username': username}


@router.get('/v1/personas', tags=['v1'], response_model=App)
def get_persona_details(uid: str = Depends(auth.get_current_user_uid)):
    app = get_persona_by_uid(uid)
    # print(app)
    app = App(**app) if app else None
    if not app:
        raise HTTPException(status_code=404, detail='Persona not found')
    if app.uid != uid:
        raise HTTPException(status_code=404, detail='Persona not found')
    if app.private is not None:
        if app.private and app.uid != uid:
            raise HTTPException(status_code=403, detail='You are not authorized to view this Persona')

    return app


@router.post('/v1/user/persona', tags=['v1'], response_model=App)
async def get_or_create_user_persona(uid: str = Depends(auth.get_current_user_uid)):
    """Get or create a user persona.

    If the user already has a persona, return it.
    If not, create a new one with default values.
    """
    # Check if user already has a persona
    persona = await run_blocking(db_executor, get_user_persona_by_uid, uid)
    if persona:
        # Return existing persona
        return persona

    # Create a new persona for the user
    user = await run_blocking(db_executor, get_user_from_uid, uid)

    # Generate a unique ID for the persona
    persona_id = str(ULID())

    # Create persona data
    persona_data = {
        'id': persona_id,
        'name': user.get('display_name', 'My Persona'),
        'username': await run_blocking(
            db_executor, increment_username, (user.get('display_name') or 'MyPersona').replace(' ', '').lower()
        ),
        'description': f"This is {user.get('display_name', 'my')} personal AI clone.",
        'image': '',  # Empty image as specified in the task
        'uid': uid,
        'author': user.get('display_name', ''),
        'email': user.get('email', ''),
        'approved': False,
        'status': 'under-review',
        'category': 'personality-emulation',
        'capabilities': ['persona'],
        'connected_accounts': ['omi'],
        'created_at': datetime.now(timezone.utc),
        'private': True,
    }

    # Generate persona prompt
    persona_data['persona_prompt'] = await generate_persona_prompt(uid, persona_data)

    try:
        persona_create = AppCreate.model_validate(persona_data)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=str(e))

    # Save username
    await run_blocking(db_executor, save_username, persona_data['username'], uid)

    # Add persona to database
    await run_blocking(db_executor, add_app_to_db, persona_create.model_dump(exclude_unset=True))

    return persona_data


@router.patch('/v1/apps/{app_id}', tags=['v1'], response_model=AppMutationResponse)
@max_part_size(APP_IMAGE_MAX_PART_SIZE)
def update_app(
    app_id: str, app_data: str = Form(...), file: UploadFile = File(None), uid=Depends(auth.get_current_user_uid)
):
    data = parse_form_json(dict, app_data, 'app_data')
    app = get_available_app_by_id(app_id, uid)
    if not app:
        raise HTTPException(status_code=404, detail='App not found')
    if app['uid'] != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    if file:
        if 'image' in app and len(app['image']) > 0 and app['image'].startswith('https://storage.googleapis.com/'):
            delete_app_logo(app['image'])
        os.makedirs('_temp/apps', exist_ok=True)
        file_path = f"_temp/apps/{file.filename}"
        with open(file_path, 'wb') as f:
            f.write(file.file.read())
        img_url = upload_app_logo(file_path, app_id)
        data['image'] = img_url
    data['updated_at'] = datetime.now(timezone.utc)

    # Backward compatibility: Set app_home_url from first auth step if not provided
    if 'external_integration' in data:
        backfill_app_home_url_from_auth_steps(data['external_integration'])

    try:
        update_app = AppUpdate.model_validate(data)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=str(e))

    # Build update dict
    update_dict = update_app.model_dump(exclude_unset=True)

    # Fetch chat tools from manifest URL (only way to add/update chat tools)
    if external_integration := data.get('external_integration'):
        update_dict = _process_chat_tools_manifest(external_integration, update_dict)

    if update_dict.get('disabled') is False and app.get('disabled'):
        validate_app_endpoints_for_reenable(app, update_dict, app_id)
        clear_app_webhook_health(app_id)
        update_dict.setdefault('disabled_reason', '')
        update_dict.setdefault('disabled_error', '')
        update_dict.setdefault('disabled_at', '')
        update_dict.setdefault('disabled_failure_duration_hours', 0)

    update_app_in_db(update_dict)

    # payment link
    upsert_app_payment_link(
        data.get('id'),
        data.get('is_paid', False),
        data.get('price'),
        data.get('payment_plan'),
        data.get('uid'),
        previous_price=app.get("price", 0),
    )

    if app['approved'] and (app['private'] is None or app['private'] is False):
        invalidate_approved_apps_cache()
    delete_app_cache_by_id(app_id)
    return {'status': 'ok'}


@router.post('/v1/apps/{app_id}/refresh-manifest', tags=['v1'], response_model=AppManifestRefreshResponse)
def refresh_app_manifest(app_id: str, uid: str = Depends(auth.get_current_user_uid)):
    """
    Refresh chat tools manifest for an app.

    Forces a fresh fetch of the manifest from the external URL, bypassing cache.
    Only the app owner can refresh their app's manifest.
    """
    app = get_available_app_by_id(app_id, uid)
    if not app:
        raise HTTPException(status_code=404, detail='App not found')
    if app['uid'] != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')

    external_integration = app.get('external_integration')
    if not external_integration:
        raise HTTPException(status_code=400, detail='App does not have external integration')

    manifest_url = external_integration.get('chat_tools_manifest_url')
    if not manifest_url:
        raise HTTPException(status_code=400, detail='App does not have a chat tools manifest URL')

    manifest_result = fetch_app_chat_tools_from_manifest(manifest_url, force_refresh=True)
    if not manifest_result:
        raise HTTPException(status_code=502, detail='Failed to fetch manifest from external URL')

    update_dict = {'id': app_id, 'updated_at': datetime.now(timezone.utc)}

    fetched_tools = manifest_result.get('tools')
    if fetched_tools:
        base_url = (external_integration.get('app_home_url') or '').rstrip('/')
        if base_url:
            for tool in fetched_tools:
                endpoint = tool.get('endpoint', '')
                if endpoint.startswith('/') and not endpoint.startswith('//'):
                    tool['endpoint'] = f"{base_url}{endpoint}"
        update_dict['chat_tools'] = fetched_tools

    chat_messages = manifest_result.get('chat_messages')
    ext_int_update = {}
    if chat_messages:
        ext_int_update['chat_messages_enabled'] = chat_messages.get('enabled', False)
        ext_int_update['chat_messages_target'] = chat_messages.get('target', 'app')
        ext_int_update['chat_messages_notify'] = chat_messages.get('notify', False)
    else:
        ext_int_update['chat_messages_enabled'] = False
        ext_int_update['chat_messages_target'] = 'app'
        ext_int_update['chat_messages_notify'] = False
    update_dict['external_integration'] = ext_int_update

    update_app_in_db(update_dict)

    if app['approved'] and (app['private'] is None or app['private'] is False):
        invalidate_approved_apps_cache()
    delete_app_cache_by_id(app_id)

    tools_count = len(fetched_tools) if fetched_tools else 0
    return {'status': 'ok', 'tools_count': tools_count}


@router.delete('/v1/apps/{app_id}', tags=['v1'], response_model=AppMutationResponse)
def delete_app(app_id: str, uid: str = Depends(auth.get_current_user_uid)):
    app = get_available_app_by_id(app_id, uid)
    if not app:
        raise HTTPException(status_code=404, detail='App not found')
    if app['uid'] != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    delete_app_from_db(app_id)
    if app['approved']:
        invalidate_approved_apps_cache()
    delete_app_cache_by_id(app_id)
    return {'status': 'ok'}


@router.get('/v1/apps/{app_id}', tags=['v1'], response_model=App)
def get_app_details(app_id: str, uid: str = Depends(auth.get_current_user_uid)):
    app = get_available_app_by_id_with_reviews(app_id, uid)
    app = App(**app) if app else None
    if not app:
        raise HTTPException(status_code=404, detail='App not found')
    if not app.approved and app.uid != uid:
        raise HTTPException(status_code=404, detail='App not found')
    if app.private is not None:
        if app.private and app.uid != uid:
            raise HTTPException(status_code=403, detail='You are not authorized to view this app')

    # is user paid
    app.is_user_paid = get_is_user_paid_app(app.id, uid)

    # payment link
    if app.payment_link:
        app.payment_link = f'{app.payment_link}?client_reference_id=uid_{uid}'

    # Generate thumbnail URLs if thumbnails exist
    if app.thumbnails:
        app.thumbnail_urls = [get_app_thumbnail_url(thumbnail_id) for thumbnail_id in app.thumbnails]

    return app


@router.get('/v1/app-categories', tags=['v1'], response_model=List[AppSelectOption])
def get_app_categories():
    return [
        {'title': 'Conversation Analysis', 'id': 'conversation-analysis'},
        {'title': 'Personality Clone', 'id': 'personality-emulation'},
        {'title': 'Health', 'id': 'health-and-wellness'},
        {'title': 'Education', 'id': 'education-and-learning'},
        {'title': 'Communication', 'id': 'communication-improvement'},
        {'title': 'Emotional Support', 'id': 'emotional-and-mental-support'},
        {'title': 'Productivity', 'id': 'productivity-and-organization'},
        {'title': 'Entertainment', 'id': 'entertainment-and-fun'},
        {'title': 'Financial', 'id': 'financial'},
        {'title': 'Travel', 'id': 'travel-and-exploration'},
        {'title': 'Safety', 'id': 'safety-and-security'},
        {'title': 'Shopping', 'id': 'shopping-and-commerce'},
        {'title': 'Social', 'id': 'social-and-relationships'},
        {'title': 'News', 'id': 'news-and-information'},
        {'title': 'Utilities', 'id': 'utilities-and-tools'},
        {'title': 'Other', 'id': 'other'},
    ]


@router.post('/v1/apps/review', tags=['v1'], response_model=AppMutationResponse)
def review_app(app_id: str, data: ReviewAppRequest, uid: str = Depends(auth.get_current_user_uid)):
    app = get_available_app_by_id(app_id, uid)
    app = App(**app) if app else None
    if not app:
        raise HTTPException(status_code=404, detail='App not found')

    if app.uid == uid:
        raise HTTPException(status_code=403, detail='You are not authorized to review your own app')

    if app.private and app.uid != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to review this app')

    review_data = {
        'score': data.score,
        'review': data.review or '',
        'username': data.username or '',
        'response': data.response or '',
        'rated_at': datetime.now(timezone.utc).isoformat(),
        'uid': uid,
    }
    set_app_review(app_id, uid, review_data)

    # Send notification to app owner
    if review_body := data.review or '':
        send_new_app_review_notification(
            app_owner_uid=app.uid,
            reviewer_uid=uid,
            app_id=app_id,
            app_name=app.name,
            review_body=review_body,
        )

    return {'status': 'ok'}


@router.patch('/v1/apps/{app_id}/review', tags=['v1'], response_model=AppMutationResponse)
def update_app_review(app_id: str, data: ReviewAppRequest, uid: str = Depends(auth.get_current_user_uid)):
    app = get_available_app_by_id(app_id, uid)
    app = App(**app) if app else None
    if not app:
        raise HTTPException(status_code=404, detail='App not found')

    if app.uid == uid:
        raise HTTPException(status_code=403, detail='You are not authorized to review your own app')

    if app.private and app.uid != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to review this app')
    old_review = get_specific_user_review(app_id, uid)
    if not old_review:
        raise HTTPException(status_code=404, detail='Review not found')
    review_data = {
        'score': data.score,
        'review': data.review or '',
        'updated_at': datetime.now(timezone.utc).isoformat(),
        'rated_at': old_review['rated_at'],
        'username': data.username if data.username is not None else old_review.get('username', ''),
        'response': old_review.get('response', ''),
        'uid': uid,
    }
    set_app_review(app_id, uid, review_data)

    # Send notification to app owner
    if review_body := data.review or '':
        send_new_app_review_notification(
            app_owner_uid=app.uid,
            reviewer_uid=uid,
            app_id=app_id,
            app_name=app.name,
            review_body=review_body,
        )

    return {'status': 'ok'}


@router.patch('/v1/apps/{app_id}/review/reply', tags=['v1'], response_model=AppMutationResponse)
def reply_to_review(app_id: str, data: ReplyToReviewRequest, uid: str = Depends(auth.get_current_user_uid)):
    app = get_available_app_by_id(app_id, uid)
    app = App(**app) if app else None
    if not app:
        raise HTTPException(status_code=404, detail='App not found')

    if app.uid != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to reply to this app review')

    if app.private and app.uid != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to reply to this app review')

    if not data.reviewer_uid:
        raise HTTPException(status_code=422, detail='Reviewer UID is required')

    if not data.response.strip():
        raise HTTPException(status_code=422, detail='Response is required')

    review = get_specific_user_review(app_id, data.reviewer_uid)
    if not review:
        raise HTTPException(status_code=404, detail='Review not found')

    review['response'] = data.response
    review['responded_at'] = datetime.now(timezone.utc).isoformat()
    set_app_review(app_id, data.reviewer_uid, review)

    # Send notification to reviewer
    send_app_review_reply_notification(
        data.reviewer_uid,
        app.uid,
        data.response,
        app_id,
        app.name,
    )

    return {'status': 'ok'}


@router.get('/v1/apps/{app_id}/reviews', tags=['v1'], response_model=List[AppReview])
def app_reviews(app_id: str):
    reviews = get_app_reviews(app_id)
    reviews = [details for details in reviews.values() if details.get('review')]
    return reviews


@router.patch('/v1/apps/{app_id}/change-visibility', tags=['v1'], response_model=AppMutationResponse)
def change_app_visibility(app_id: str, private: bool, uid: str = Depends(auth.get_current_user_uid)):
    app = get_available_app_by_id(app_id, uid)
    app = App(**app) if app else None
    if not app:
        raise HTTPException(status_code=404, detail='App not found')
    if app.uid != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    update_app_visibility_in_db(app_id, private)
    # Toggling visibility of an approved app changes whether it appears in the public marketplace
    # list, so invalidate that cache too (mirrors approve/reject/delete). Otherwise a newly public
    # app does not show, and a newly private one keeps showing, until the list cache TTL expires
    # (issue #3783).
    if app.approved:
        invalidate_approved_apps_cache()
    delete_app_cache_by_id(app_id)
    return {'status': 'ok'}


@router.get('/v1/app/proactive-notification-scopes', tags=['v1'], response_model=List[AppSelectOption])
def get_notification_scopes():
    return [
        {'title': 'User Name', 'id': 'user_name'},
        {'title': 'User Memories', 'id': 'user_facts'},
        {'title': 'User Conversations', 'id': 'user_context'},
        {'title': 'User Chat', 'id': 'user_chat'},
    ]


@router.get('/v1/app-capabilities', tags=['v1'], response_model=List[AppCapabilityResponse])
def get_app_capabilities():
    return [
        {'title': 'Chat', 'id': 'chat'},
        {'title': 'Conversations', 'id': 'memories'},
        {
            'title': 'External Integration',
            'id': 'external_integration',
            'triggers': [
                {'title': 'Audio Bytes', 'id': 'audio_bytes'},
                {'title': 'Conversation Creation', 'id': 'memory_creation'},
                {'title': 'Transcript Processed', 'id': 'transcript_processed'},
            ],
            'actions': [
                {
                    'title': 'Create conversations',
                    'id': 'create_conversation',
                    'doc_url': 'https://docs.omi.me/doc/developer/apps/Import',
                    'description': 'Extend user conversations by making a POST request to the OMI System.',
                },
                {
                    'title': 'Create memories',
                    'id': 'create_facts',
                    'doc_url': 'https://docs.omi.me/doc/developer/apps/Import',
                    'description': 'Create new memories for the user through the OMI System.',
                },
                {
                    'title': 'Read conversations',
                    'id': 'read_conversations',
                    'doc_url': 'https://docs.omi.me/doc/developer/apps/Import',
                    'description': 'Access and read all user conversations through the OMI System. This gives the app access to all conversation history.',
                },
                {
                    'title': 'Read memories',
                    'id': 'read_memories',
                    'doc_url': 'https://docs.omi.me/doc/developer/apps/Import',
                    'description': 'Access and read all user memories through the OMI System. This gives the app access to all stored memories.',
                },
                {
                    'title': 'Read tasks',
                    'id': 'read_tasks',
                    'doc_url': 'https://docs.omi.me/doc/developer/apps/Import',
                    'description': 'Access and read all user tasks (to-dos) through the OMI System. This gives the app access to all stored tasks.',
                },
            ],
        },
        {
            'title': 'Notification',
            'id': 'proactive_notification',
            'scopes': [
                {'title': 'User Name', 'id': 'user_name'},
                {'title': 'User Facts', 'id': 'user_facts'},
                {'title': 'User Conversations', 'id': 'user_context'},
                {'title': 'User Chat', 'id': 'user_chat'},
            ],
        },
    ]


# @deprecated
@router.get('/v1/app/payment-plans', tags=['v1'], response_model=List[AppSelectOption])
def get_payment_plans_v1():
    return [
        {'title': 'Monthly Recurring', 'id': 'monthly_recurring'},
    ]


@router.get('/v1/app/plans', tags=['v1'], response_model=List[AppSelectOption])
def get_payment_plans(uid: str = Depends(auth.get_current_user_uid)):
    if not uid or len(uid) == 0 or not is_permit_payment_plan_get(uid):
        return []
    return [
        {'title': 'Monthly Recurring', 'id': 'monthly_recurring'},
    ]


@router.post('/v1/app/generate-description', tags=['v1'], response_model=AppDescriptionGenerationResponse)
def generate_description_endpoint(
    data: GenerateDescriptionRequest,
    uid: str = Depends(auth.get_current_user_uid),
    x_app_platform: Optional[str] = Header(None, alias='X-App-Platform'),
):
    # User-initiated LLM generation — same free-tier gate as chat (402 past cap).
    enforce_chat_quota(uid, platform=x_app_platform)
    if data.name == '':
        raise HTTPException(status_code=422, detail='App Name is required')
    if data.description == '':
        raise HTTPException(status_code=422, detail='App Description is required')
    with track_usage(uid, Features.APP_GENERATOR):
        desc = generate_description(data.name, data.description)
    return {
        'description': desc,
    }


@router.post('/v1/app/generate-description-emoji', tags=['v1'], response_model=AppDescriptionEmojiGenerationResponse)
def generate_description_and_emoji_endpoint(
    data: GenerateDescriptionEmojiRequest,
    uid: str = Depends(auth.get_current_user_uid),
    x_app_platform: Optional[str] = Header(None, alias='X-App-Platform'),
):
    """
    Generate an app description and representative emoji.
    Used by the quick template creator feature.
    """
    # User-initiated LLM generation — same free-tier gate as chat (402 past cap).
    enforce_chat_quota(uid, platform=x_app_platform)
    if not data.name:
        raise HTTPException(status_code=422, detail='App Name is required')
    if not data.prompt:
        raise HTTPException(status_code=422, detail='App Prompt is required')

    with track_usage(uid, Features.APP_GENERATOR):
        result = generate_description_and_emoji(data.name, data.prompt)
    return result


# ******************************************************
# ****************** AI APP GENERATOR ******************
# ******************************************************


@router.get('/v1/app/generate-prompts', tags=['v1'], response_model=AppPromptsGenerationResponse)
async def generate_sample_prompts_endpoint(
    uid: str = Depends(auth.with_rate_limit(auth.get_current_user_uid, "apps:generate_prompts")),
    x_app_platform: Optional[str] = Header(None, alias='X-App-Platform'),
):
    """
    Generate sample app prompts for the AI app generator.
    Uses a fast model to generate creative suggestions.
    """
    from utils.llm.clients import get_llm
    import json

    # User-initiated LLM generation — same free-tier gate as chat (402 past cap).
    enforce_chat_quota(uid, platform=x_app_platform)

    system_prompt = """Generate 5 creative and diverse ideas for apps that are either:
1. Conversation summary based apps - analyze user's recorded conversations and extract/organize information
2. Chat assistant based apps - AI personas or assistants users can chat with

Generate exactly 3 conversation-based and 2 chat-based app ideas.

Examples:
- Conversation based: "Mind map generator from my conversations", "Jokes and funny moments extractor", "Meeting action items tracker"
- Chat based: "Elon Musk personality clone", "Strict accountability mentor", "Socratic philosophy tutor"

Return ONLY a JSON array of 5 strings, each being a short app description (max 50 characters).
Format: ["idea 1", "idea 2", "idea 3", "idea 4", "idea 5"]

First 3 should be conversation-based, last 2 should be chat-based.
Be creative, fun, and varied. No generic ideas."""

    try:
        with track_usage(uid, Features.APP_GENERATOR):
            response = await get_llm('app_integration').ainvoke(
                [
                    SystemMessage(content=system_prompt),
                    HumanMessage(content="Generate 5 creative app ideas now"),
                ]
            )

        content = response.content.strip()

        # Parse JSON from response
        if content.startswith("```"):
            lines = content.split("\n")
            content = "\n".join(lines[1:-1] if lines[-1] == "```" else lines[1:])

        prompts = json.loads(content)

        return app_generation_prompts_from_llm_payload(prompts)
    except Exception as e:
        logger.error(f"Error generating prompts: {e}")
        return app_generation_prompts_response()


@router.post('/v1/app/generate', tags=['v1'], response_model=AppGenerationResponse)
async def generate_app_endpoint(
    data: GenerateAppRequest,
    uid: str = Depends(auth.get_current_user_uid),
    x_app_platform: Optional[str] = Header(None, alias='X-App-Platform'),
):
    """
    Generate an app configuration from a natural language prompt.
    This is an experimental feature that uses AI to create app configurations.
    """
    from utils.llm.app_generator import generate_app_from_prompt

    # User-initiated LLM generation — same free-tier gate as chat (402 past cap).
    enforce_chat_quota(uid, platform=x_app_platform)

    prompt = data.prompt.strip()
    if not prompt:
        raise HTTPException(status_code=422, detail='Prompt is required')

    if len(prompt) < 10:
        raise HTTPException(status_code=422, detail='Prompt is too short. Please provide more details.')

    if len(prompt) > 2000:
        raise HTTPException(status_code=422, detail='Prompt is too long. Please keep it under 2000 characters.')

    try:
        # Generate app configuration using LLM
        with track_usage(uid, Features.APP_GENERATOR):
            generated_app = await generate_app_from_prompt(prompt)

        return {
            'status': 'ok',
            'app': {
                'name': generated_app.name,
                'description': generated_app.description,
                'category': generated_app.category,
                'capabilities': generated_app.capabilities,
                'chat_prompt': generated_app.chat_prompt,
                'memory_prompt': generated_app.memory_prompt,
            },
        }
    except Exception as e:
        logger.error(f"Error generating app: {e}")
        raise HTTPException(status_code=500, detail=f'Failed to generate app: {str(e)}')


@router.post('/v1/app/generate-icon', tags=['v1'], response_model=AppIconGenerationResponse)
async def generate_app_icon_endpoint(
    data: GenerateAppIconRequest,
    uid: str = Depends(auth.get_current_user_uid),
    x_app_platform: Optional[str] = Header(None, alias='X-App-Platform'),
):
    """
    Generate an app icon using AI (DALL-E).
    Returns the icon as a base64 encoded PNG image.
    """
    from utils.llm.app_generator import generate_app_icon
    import base64

    # User-initiated LLM generation — same free-tier gate as chat (402 past cap).
    enforce_chat_quota(uid, platform=x_app_platform)

    app_name = data.name.strip()
    app_description = data.description.strip()
    category = data.category.strip()

    if not app_name:
        raise HTTPException(status_code=422, detail='App name is required')

    if not app_description:
        raise HTTPException(status_code=422, detail='App description is required')

    try:
        # Generate icon using DALL-E
        with track_usage(uid, Features.APP_GENERATOR):
            icon_bytes = await generate_app_icon(app_name, app_description, category)

        # Return as base64
        icon_base64 = base64.b64encode(icon_bytes).decode('utf-8')

        return {'status': 'ok', 'icon_base64': icon_base64, 'mime_type': 'image/png'}
    except Exception as e:
        logger.error(f"Error generating icon: {e}")
        raise HTTPException(status_code=500, detail=f'Failed to generate icon: {str(e)}')


# ******************************************************
# ********************** SOCIAL ************************
# ******************************************************


@router.get('/v1/personas/twitter/profile', tags=['v1'], response_model=TwitterProfileResponse)
async def get_twitter_profile_data(handle: str, uid: str = Depends(auth.get_current_user_uid)):
    if handle.startswith('@'):
        handle = handle[1:]
    profile = await get_twitter_profile(handle)

    # Convert TwitterProfile to dict for response
    res = {
        "name": profile.name,
        "profile": profile.profile,
        "rest_id": profile.rest_id,
        "avatar": profile.avatar,
        "desc": profile.desc,
        "friends": profile.friends,
        "sub_count": profile.sub_count,
        "id": profile.id,
        "status": profile.status,
    }

    # By user persona first
    persona = await run_blocking(db_executor, get_user_persona_by_uid, uid)

    # Get matching persona if exists
    if not persona:
        persona = await run_blocking(db_executor, get_persona_by_twitter_handle_db, handle)

    if persona:
        res['persona_id'] = persona['id']
        res['persona_username'] = persona['username']

    return res


@router.get('/v1/personas/twitter/verify-ownership', tags=['v1'], response_model=TwitterOwnershipVerificationResponse)
async def verify_twitter_ownership_tweet(
    username: str, handle: str, uid: str = Depends(auth.get_current_user_uid), persona_id: str | None = None
):
    # Get user info to check auth provider
    user = await run_blocking(db_executor, get_user_from_uid, uid)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    # Get provider info from Firebase
    user_info = auth.get_user(uid)
    provider_data = [p.provider_id for p in user_info.provider_data]

    # Verify handle
    if handle.startswith('@'):
        handle = handle[1:]
    if username.startswith('@'):
        username = username[1:]
    persona = None
    res = await verify_latest_tweet(username, handle)
    if res['verified']:
        if not ('google.com' in provider_data or 'apple.com' in provider_data):
            persona = await upsert_persona_from_twitter_profile(username, handle, uid)
        else:
            if persona_id:
                persona = await add_twitter_to_persona(handle, persona_id)

    if persona:
        res['persona_id'] = persona['id']

    return res


@router.get('/v1/personas/twitter/initial-message', tags=['v1'], response_model=TwitterInitialMessageResponse)
def get_twitter_initial_message(
    username: str,
    uid: str = Depends(auth.with_rate_limit(auth.get_current_user_uid, "apps:twitter_initial_message")),
):
    persona = get_persona_by_username_db(username)
    if persona:
        with track_usage(uid, Features.PERSONA):
            message = generate_persona_intro_message(persona['persona_prompt'], persona['name'])
        return {'message': message}
    return {'message': ''}


@router.post('/v1/apps/migrate-owner', tags=['v1'], response_model=AppMigrationResponse)
async def migrate_app_owner(
    old_id,
    source_token: Optional[str] = Body(default=None, embed=True),
    uid: str = Depends(auth.get_current_user_uid),
):
    # The client captures this token while it still owns the anonymous Firebase
    # session, then calls this route after signing into the destination account.
    # A providerless UserRecord is not affirmative proof of anonymity (for example,
    # custom-token accounts can be providerless), so a bare ``old_id`` never grants
    # access.  The source token must be a currently valid Firebase credential for
    # that exact uid and attest to the anonymous sign-in provider.
    if old_id == uid:
        raise HTTPException(status_code=400, detail='Source identity must differ from the authenticated identity')
    if not source_token:
        raise HTTPException(status_code=403, detail='Source identity is not eligible for migration')

    try:
        source_claims = await run_blocking(
            critical_executor, auth.auth.verify_id_token, source_token, check_revoked=True
        )
        source_user = await run_blocking(critical_executor, auth.get_user, old_id)
    except Exception:
        # Invalid/revoked tokens, missing/deleted users, and Admin lookup failures
        # are deliberately indistinguishable to callers. Neither may mutate state.
        raise HTTPException(status_code=403, detail='Source identity is not eligible for migration')

    source_uid = source_claims.get('uid')
    firebase_claims = source_claims.get('firebase')
    source_provider = firebase_claims.get('sign_in_provider') if isinstance(firebase_claims, dict) else None
    if source_uid != old_id or source_provider != 'anonymous' or source_user.disabled or source_user.provider_data:
        raise HTTPException(status_code=403, detail='Source identity is not eligible for migration')

    try:
        await run_blocking(db_executor, auth.enforce_account_deletion_http_access, old_id)
    except HTTPException:
        raise HTTPException(status_code=403, detail='Source identity is not eligible for migration')

    await run_blocking(db_executor, migrate_app_owner_id_db, uid, old_id)

    # Tracked background tasks (not bare asyncio.create_task): keeps a live reference
    # and logs a failed migration instead of silently dropping it.
    start_background_task(run_blocking(db_executor, migrate_memories, old_id, uid), name=f"migrate_memories:{uid}")
    start_background_task(update_omi_persona_connected_accounts(uid), name=f"migrate_persona_accounts:{uid}")

    return {"status": "ok", "message": "Migration started"}


async def update_omi_persona_connected_accounts(uid: str):
    try:
        personas = await run_blocking(db_executor, get_omi_persona_apps_by_uid_db, uid)

        for persona in personas:
            connected_accounts = persona.get('connected_accounts', [])
            if 'omi' not in connected_accounts:
                connected_accounts.append('omi')

                update_data = persona
                update_data['connected_accounts'] = connected_accounts
                update_data['updated_at'] = datetime.now(timezone.utc)
                update_data['persona_prompt'] = await generate_persona_prompt(uid, update_data)
                update_data['description'] = await run_blocking(
                    llm_executor, generate_persona_desc, uid, update_data['name']
                )

                await run_blocking(db_executor, update_app_in_db, update_data)
                await run_blocking(db_executor, delete_app_cache_by_id, persona['id'])
    except Exception as e:
        logger.error(f"Error updating persona connected accounts: {e}")


# ******************************************************
# ******************* MCP SERVERS **********************
# ******************************************************


class McpServerRequest(PydanticBaseModel):
    name: str
    mcp_server_url: str
    description: Optional[str] = None


def _serialize_chat_tools_for_firestore(tools) -> list:
    """Serialize ChatTool objects for Firestore, converting parameters dict to JSON string.

    Firestore has nesting depth limits that MCP tool schemas can exceed,
    so we store the parameters field as a JSON string.
    """
    result = []
    for t in tools:
        d = t.model_dump()
        if d.get('parameters') is not None:
            d['parameters'] = json.dumps(d['parameters'])
        result.append(d)
    return result


@router.post('/v1/apps/mcp', tags=['v1'], response_model=McpAddServerResponse)
async def add_mcp_server(data: McpServerRequest, uid: str = Depends(auth.get_current_user_uid)):
    """Add a remote MCP server as a private app with chat tools.

    1. Extracts domain from URL and fetches logo via Brandfetch / logo.dev
    2. Checks for OAuth metadata at /.well-known/oauth-authorization-server
    3. If OAuth required: registers client, returns auth URL for the user
    4. If no OAuth: discovers tools directly, creates app immediately
    """
    server_url = data.mcp_server_url.strip().rstrip('/')
    app_name = data.name.strip()
    app_description = data.description.strip() if data.description else f"MCP server tools from {app_name}"

    if not app_name:
        raise HTTPException(status_code=422, detail='App name is required')
    if not server_url:
        raise HTTPException(status_code=422, detail='MCP server URL is required')

    # Extract domain for logo
    parsed = urlparse(server_url)
    domain = parsed.netloc
    logo_url = await fetch_brandfetch_logo(domain) or ''

    app_id = str(ULID())
    user = await run_blocking(db_executor, get_user_from_uid, uid)

    # Check for OAuth metadata
    oauth_meta = await discover_oauth_metadata(server_url)

    if oauth_meta and oauth_meta.get('authorization_endpoint'):
        # OAuth required — register client and return auth URL
        base_url = os.getenv('BASE_API_URL', '').rstrip('/')
        if not base_url:
            raise HTTPException(status_code=500, detail='BASE_API_URL not configured')

        redirect_uri = f"{base_url}/v1/apps/mcp/callback"

        client_info = {}
        if oauth_meta.get('registration_endpoint'):
            try:
                client_info = await register_oauth_client(
                    oauth_meta['registration_endpoint'], redirect_uri, scopes=oauth_meta.get('scopes_supported')
                )
            except Exception as e:
                raise HTTPException(status_code=502, detail=f'OAuth client registration failed: {str(e)}')
        else:
            raise HTTPException(
                status_code=422,
                detail='MCP server requires OAuth but does not support dynamic client registration',
            )

        state = generate_state_token(app_id, uid)

        # Generate PKCE pair (required by MCP OAuth 2.1 spec)
        code_verifier, code_challenge = generate_pkce_pair()

        auth_url = build_authorization_url(
            oauth_meta['authorization_endpoint'],
            client_info['client_id'],
            redirect_uri,
            state,
            scopes=oauth_meta.get('scopes_supported'),
            code_challenge=code_challenge,
        )
        logger.info(f"[MCP OAuth] client_id={client_info['client_id']}, redirect_uri={redirect_uri}")
        logger.info(f"[MCP OAuth] auth_url={auth_url}")

        # Create app in pending state (no tools yet)
        app_dict = {
            'id': app_id,
            'name': app_name,
            'description': app_description,
            'image': logo_url,
            'uid': uid,
            'author': user.get('display_name', ''),
            'email': user.get('email', ''),
            'private': True,
            'approved': True,
            'status': 'pending_mcp_auth',
            'category': 'utilities-and-tools',
            'capabilities': ['chat'],
            'created_at': datetime.now(timezone.utc),
            'external_integration': {
                'mcp_server_url': server_url,
                'mcp_oauth_tokens': {
                    'client_id': client_info['client_id'],
                    'client_secret': client_info.get('client_secret'),
                    'token_endpoint': oauth_meta['token_endpoint'],
                    'redirect_uri': redirect_uri,
                    'code_verifier': code_verifier,
                },
            },
            'chat_tools': [],
        }
        await run_blocking(db_executor, add_app_to_db, app_dict)

        return {
            'app_id': app_id,
            'requires_oauth': True,
            'auth_url': auth_url,
        }

    else:
        # No OAuth — discover tools directly
        try:
            tools = await discover_mcp_tools(server_url)
        except Exception as e:
            raise HTTPException(status_code=502, detail=f'Failed to discover MCP tools: {str(e)}')

        if not tools:
            raise HTTPException(status_code=422, detail='No tools found on the MCP server')

        # Use the resolved URL from discovery (may differ from user input if /http or /sse was needed)
        resolved_url = tools[0].endpoint if tools else server_url

        app_dict = {
            'id': app_id,
            'name': app_name,
            'description': app_description,
            'image': logo_url,
            'uid': uid,
            'author': user.get('display_name', ''),
            'email': user.get('email', ''),
            'private': True,
            'approved': True,
            'status': 'approved',
            'category': 'utilities-and-tools',
            'capabilities': ['chat'],
            'created_at': datetime.now(timezone.utc),
            'external_integration': {
                'mcp_server_url': resolved_url,
            },
            'chat_tools': _serialize_chat_tools_for_firestore(tools),
        }
        await run_blocking(db_executor, add_app_to_db, app_dict)

        return {
            'app_id': app_id,
            'requires_oauth': False,
            'tools_count': len(tools),
            'tool_names': [t.name for t in tools],
        }


@router.get(
    '/v1/apps/mcp/callback', tags=['v1'], response_class=HTMLResponse
)  # response_model omitted: OAuth callback returns HTML
async def mcp_oauth_callback(code: str, state: str):
    """OAuth callback for MCP server authorization.

    Exchanges the authorization code for tokens, discovers tools, updates the app.
    Returns an HTML success/failure page.
    """
    try:
        app_id, uid = parse_state_token(state)
    except ValueError:
        return HTMLResponse('<html><body><h1>Invalid state parameter</h1></body></html>', status_code=400)

    app_data = await run_blocking(db_executor, get_app_by_id_db, app_id)
    if not app_data:
        return HTMLResponse('<html><body><h1>App not found</h1></body></html>', status_code=404)

    ext = app_data.get('external_integration', {})
    oauth_tokens = ext.get('mcp_oauth_tokens', {})
    server_url = ext.get('mcp_server_url', '')

    if not oauth_tokens or not oauth_tokens.get('token_endpoint'):
        return HTMLResponse('<html><body><h1>OAuth configuration missing</h1></body></html>', status_code=400)

    # Exchange code for tokens (include PKCE code_verifier)
    try:
        token_data = await exchange_oauth_code(
            oauth_tokens['token_endpoint'],
            code,
            oauth_tokens.get('redirect_uri', ''),
            oauth_tokens['client_id'],
            oauth_tokens.get('client_secret'),
            code_verifier=oauth_tokens.get('code_verifier'),
        )
    except Exception as e:
        return HTMLResponse(f'<html><body><h1>Token exchange failed</h1><p>{str(e)}</p></body></html>', status_code=502)

    # Update stored tokens
    oauth_tokens['access_token'] = token_data['access_token']
    oauth_tokens['refresh_token'] = token_data.get('refresh_token')
    if token_data.get('expires_in'):
        oauth_tokens['expires_at'] = time.time() + token_data['expires_in']

    # Discover tools with the new access token
    try:
        tools = await discover_mcp_tools(server_url, token_data['access_token'])
    except Exception as e:
        return HTMLResponse(f'<html><body><h1>Tool discovery failed</h1><p>{str(e)}</p></body></html>', status_code=502)

    # Use the resolved URL from the first tool (discover_mcp_tools stores the working URL)
    resolved_url = tools[0].endpoint if tools else server_url

    # Update app with tokens and tools
    update_dict = {
        'id': app_id,
        'status': 'approved',
        'external_integration': {
            'mcp_server_url': resolved_url,
            'mcp_oauth_tokens': oauth_tokens,
        },
        'chat_tools': _serialize_chat_tools_for_firestore(tools),
    }
    await run_blocking(db_executor, update_app_in_db, update_dict)
    await run_blocking(db_executor, delete_app_cache_by_id, app_id)

    # Auto-enable the app for the user
    await run_blocking(db_executor, enable_app, uid, app_id)

    tool_count = len(tools)
    tool_names = ', '.join(escape(t.name) for t in tools)

    return HTMLResponse(f"""
    <html>
    <head><meta name="viewport" content="width=device-width,initial-scale=1">
    <style>
        body {{ font-family: -apple-system, system-ui, sans-serif; background: #111; color: #fff;
               display: flex; align-items: center; justify-content: center; min-height: 100vh; margin: 0; }}
        .card {{ background: #1a1a1a; border-radius: 16px; padding: 40px; text-align: center; max-width: 400px; }}
        h1 {{ color: #4ade80; margin-bottom: 8px; }}
        p {{ color: #aaa; }}
        .count {{ font-size: 2em; font-weight: bold; color: #fff; }}
    </style></head>
    <body><div class="card">
        <h1>Connected!</h1>
        <p class="count">{tool_count} tools</p>
        <p>{tool_names}</p>
        <p style="margin-top:24px;color:#666;">You can close this window and return to the app.</p>
    </div></body></html>
    """)


@router.post('/v1/apps/{app_id}/mcp/refresh', tags=['v1'], response_model=McpRefreshToolsResponse)
async def refresh_mcp_tools(app_id: str, uid: str = Depends(auth.get_current_user_uid)):
    """Re-discover tools from an MCP server and update the app."""
    app_data = await run_blocking(db_executor, get_app_by_id_db, app_id)
    if not app_data:
        raise HTTPException(status_code=404, detail='App not found')
    if app_data.get('uid') != uid:
        raise HTTPException(status_code=403, detail='Not authorized')

    ext = app_data.get('external_integration', {})
    server_url = ext.get('mcp_server_url')
    if not server_url:
        raise HTTPException(status_code=422, detail='App is not an MCP server app')

    oauth_tokens = ext.get('mcp_oauth_tokens')
    access_token = oauth_tokens.get('access_token') if oauth_tokens else None

    try:
        tools = await discover_mcp_tools(server_url, access_token)
    except PermissionError:
        # Try token refresh
        if oauth_tokens and oauth_tokens.get('refresh_token'):
            new_tokens = await refresh_oauth_token(
                oauth_tokens['token_endpoint'],
                oauth_tokens['refresh_token'],
                oauth_tokens['client_id'],
                oauth_tokens.get('client_secret'),
            )
            oauth_tokens['access_token'] = new_tokens['access_token']
            if new_tokens.get('refresh_token'):
                oauth_tokens['refresh_token'] = new_tokens['refresh_token']

            tools = await discover_mcp_tools(server_url, new_tokens['access_token'])

            update_dict = {
                'id': app_id,
                'external_integration': {
                    'mcp_server_url': server_url,
                    'mcp_oauth_tokens': oauth_tokens,
                },
                'chat_tools': _serialize_chat_tools_for_firestore(tools),
            }
            await run_blocking(db_executor, update_app_in_db, update_dict)
            await run_blocking(db_executor, delete_app_cache_by_id, app_id)

            return {'tools_count': len(tools), 'tool_names': [t.name for t in tools]}
        raise HTTPException(status_code=401, detail='MCP server requires re-authorization')
    except Exception as e:
        raise HTTPException(status_code=502, detail=f'Failed to discover tools: {str(e)}')

    update_dict = {
        'id': app_id,
        'chat_tools': _serialize_chat_tools_for_firestore(tools),
    }
    await run_blocking(db_executor, update_app_in_db, update_dict)
    await run_blocking(db_executor, delete_app_cache_by_id, app_id)

    return {'tools_count': len(tools), 'tool_names': [t.name for t in tools]}


# ******************************************************
# **************** ENABLE/DISABLE APPS *****************
# ******************************************************


def _setup_completed_from_response(res: httpx.Response) -> bool:
    """Read `is_setup_completed` from a third-party setup_completed_url response.

    The body is developer-controlled, so it may be non-JSON or a JSON scalar/array
    rather than the documented object. Anything that is not an object carrying a
    truthy `is_setup_completed` means setup is not completed.
    """
    try:
        payload: object = res.json()
    except ValueError:
        return False
    return isinstance(payload, dict) and bool(payload.get('is_setup_completed', False))


def _disabled_app_install_detail(app: App, uid: str) -> str:
    """Explain a blocked install truthfully.

    The previous copy claimed a live connectivity problem and that the developer
    had been notified. Neither is checked here: `disabled` is a stored flag set
    up to months earlier, and the notification went out then. Owners chased
    their own healthy servers looking for a request that is never sent.
    """
    when = f' on {app.disabled_at[:10]}' if app.disabled_at else ''
    if app.uid == uid:
        cause = f' Last error: {app.disabled_error}.' if app.disabled_error else ''
        return (
            f'This app was automatically disabled{when} after its webhook endpoint failed for 72 hours.'
            f'{cause} Fix the endpoint, then press Re-enable on the app page.'
        )
    return f'This app was disabled{when} because its endpoint stopped responding. Its developer has to re-enable it.'


@router.post('/v1/apps/enable', response_model=AppMutationResponse)
async def enable_app_endpoint(app_id: str, uid: str = Depends(auth.get_current_user_uid)):
    app = await run_blocking(db_executor, get_available_app_by_id, app_id, uid)
    app = App(**app) if app else None
    if not app:
        raise HTTPException(status_code=404, detail='App not found')
    if app.disabled:
        raise HTTPException(status_code=400, detail=_disabled_app_install_detail(app, uid))
    if app.private is not None:
        if app.private and app.uid != uid and not await run_blocking(db_executor, is_tester, uid):
            raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    if app.works_externally() and app.external_integration.setup_completed_url:
        client = get_webhook_client()
        res = await client.get(app.external_integration.setup_completed_url + f'?uid={uid}')
        logger.info(f'enable_app_endpoint {res.status_code} {res.content}')
        if res.status_code != 200 or not _setup_completed_from_response(res):
            raise HTTPException(status_code=400, detail='App setup is not completed')

    # Check payment status
    if app.is_paid and not await run_blocking(db_executor, get_is_user_paid_app, app.id, uid):
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')

    await run_blocking(db_executor, enable_app, uid, app_id)
    if (
        (app.private is None or not app.private)
        and (app.uid is None or app.uid != uid)
        and not await run_blocking(db_executor, is_tester, uid)
    ):
        await run_blocking(db_executor, increase_app_installs_count, app_id)
    return {'status': 'ok'}


@router.post('/v1/apps/disable', response_model=AppMutationResponse)
def disable_app_endpoint(app_id: str, uid: str = Depends(auth.get_current_user_uid)):
    # Allow users to always disable apps they have installed, even if the app
    # was made private after installation (see issue #4886).
    if is_app_enabled(uid, app_id):
        disable_app(uid, app_id)
        app = get_available_app_by_id(app_id, uid)
        if app:
            app = App(**app)
            if (app.private is None or not app.private) and (app.uid is None or app.uid != uid) and not is_tester(uid):
                decrease_app_installs_count(app_id)
        return {'status': 'ok'}

    raise HTTPException(status_code=404, detail='App not found')


# ******************************************************
# ******************* TEAM ENDPOINTS *******************
# ******************************************************


@router.post('/v1/apps/tester', tags=['v1'], response_model=AppMutationResponse)
def add_new_tester(data: AddTesterRequest, secret_key: str = Header(...)):
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    if not data.uid:
        raise HTTPException(status_code=422, detail='uid is required')
    if not data.apps:
        raise HTTPException(status_code=422, detail='apps is required')
    payload = data.model_dump()
    payload['added_at'] = datetime.now(timezone.utc).isoformat()
    add_tester(payload)
    return {'status': 'ok'}


@router.post('/v1/apps/tester/access', tags=['v1'], response_model=AppMutationResponse)
def add_app_access_tester(data: TesterAccessRequest, secret_key: str = Header(...)):
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    if not data.uid:
        raise HTTPException(status_code=422, detail='uid is required')
    if not data.app_id:
        raise HTTPException(status_code=422, detail='app_id is required')
    add_app_access_for_tester(data.app_id, data.uid)
    return {'status': 'ok'}


@router.delete('/v1/apps/tester/access', tags=['v1'], response_model=AppMutationResponse)
def remove_app_access_tester(data: TesterAccessRequest, secret_key: str = Header(...)):
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    if not data.uid:
        raise HTTPException(status_code=422, detail='uid is required')
    if not data.app_id:
        raise HTTPException(status_code=422, detail='app_id is required')
    remove_app_access_for_tester(data.app_id, data.uid)
    return {'status': 'ok'}


@router.get('/v1/apps/tester/check', tags=['v1'], response_model=AppTesterCheckResponse)
def check_is_tester(uid: str = Depends(auth.get_current_user_uid)):
    if is_tester(uid):
        return {'is_tester': True}
    return {'is_tester': False}


@router.get('/v1/apps/public/unapproved', tags=['v1'], response_model=List[UnapprovedPublicAppResponse])
def get_unapproved_public_apps(secret_key: str = Header(...)):
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    apps = get_unapproved_public_apps_db()
    return apps


@router.patch('/v1/apps/{app_id}/popular', tags=['v1'], response_model=AppMutationResponse)
def set_app_popular(app_id: str, value: bool = Query(...), secret_key: str = Header(...)):
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    set_app_popular_db(app_id, value)
    delete_app_cache_by_id(app_id)
    invalidate_popular_apps_cache()
    return {'status': 'ok'}


@router.post('/v1/apps/{app_id}/approve', tags=['v1'], response_model=AppMutationResponse)
def approve_app(app_id: str, uid: str, secret_key: str = Header(...)):
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    change_app_approval_status(app_id, True)
    invalidate_approved_apps_cache()  # App is now public, invalidate cache
    delete_app_cache_by_id(app_id)
    app = get_available_app_by_id(app_id, uid)
    send_notification(
        uid,
        'App Approved 🎉',
        f'Your app {app["name"]} has been approved and is now available for everyone to use 🥳',
    )
    return {'status': 'ok'}


@router.post('/v1/apps/{app_id}/reject', tags=['v1'], response_model=AppMutationResponse)
def reject_app(app_id: str, uid: str, secret_key: str = Header(...)):
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    change_app_approval_status(app_id, False)
    invalidate_approved_apps_cache()  # App removed from public list, invalidate cache
    delete_app_cache_by_id(app_id)
    app = get_available_app_by_id(app_id, uid)
    # TODO: Add reason for rejection in payload and also redirect to the app page
    send_notification(
        uid,
        'App Rejected 😔',
        f'Your app {app["name"]} has been rejected. Please make the necessary changes and resubmit for approval.',
    )
    return {'status': 'ok'}


@router.post('/v1/app/thumbnails', tags=['v1'], response_model=AppThumbnailUploadResponse)
@max_part_size(APP_IMAGE_MAX_PART_SIZE)
async def upload_app_thumbnail_endpoint(file: UploadFile = File(...), uid: str = Depends(auth.get_current_user_uid)):
    """Upload a thumbnail image for an app.

    Args:
        file: The thumbnail image file
        app_id: ID of the app to add thumbnail for
        uid: User ID from auth

    Returns:
        Dict with thumbnail URL
    """
    # Save uploaded file temporarily
    thumbnail_id = str(ULID())
    os.makedirs('_temp/thumbnails', exist_ok=True)
    temp_path = f'_temp/thumbnails/{thumbnail_id}.jpg'

    try:
        contents = await file.read()
        await run_blocking(storage_executor, _write_file, temp_path, contents)

        # Upload to cloud storage
        url = await run_blocking(storage_executor, upload_app_thumbnail, temp_path, thumbnail_id)

        return {'thumbnail_url': url, 'thumbnail_id': thumbnail_id}

    finally:
        # Cleanup temp file
        if os.path.exists(temp_path):
            os.remove(temp_path)


@router.delete('/v1/personas/{persona_id}', tags=['v1'], response_model=AppMutationResponse)
def delete_persona(persona_id: str, secret_key: str = Header(...)):
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    personas = get_persona_by_id_db(persona_id)
    if not personas:
        raise HTTPException(status_code=404, detail='Persona not found')
    delete_persona_db(persona_id)
    return {'status': 'ok'}


@router.get('/v1/personas/{persona_id}', tags=['v1'], response_model=List[PersonaRecordResponse])
def get_personas(persona_id: str, secret_key: str = Header(...)):
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='You are not authorized to perform this action')
    persona = get_personas_by_username_db(persona_id)
    if not persona:
        raise HTTPException(status_code=404, detail='Persona not found')
    logger.info(persona)
    return persona


@router.post('/v1/apps/{app_id}/keys', tags=['v1'], response_model=AppApiKeyResponse)
def create_api_key_for_app(app_id: str, uid: str = Depends(auth.get_current_user_uid)):
    app = get_available_app_by_id(app_id, uid)
    if not app:
        raise HTTPException(status_code=404, detail='App not found')

    if app.get('uid') != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to create API keys for this app')

    key, hashed_key, label = generate_api_key()

    data = {'id': str(ULID()), 'hashed': hashed_key, 'label': label, 'created_at': datetime.now(timezone.utc)}
    create_api_key_db(app_id, data)

    # Return both the raw key (for one-time display to user) and the stored data
    return {'id': data['id'], 'secret': key, 'label': label, 'created_at': data['created_at']}  # with sk_


@router.get('/v1/apps/{app_id}/keys', tags=['v1'], response_model=List[AppApiKeyResponse])
def list_api_keys(app_id: str, uid: str = Depends(auth.get_current_user_uid)):
    app = get_available_app_by_id(app_id, uid)
    if not app:
        raise HTTPException(status_code=404, detail='App not found')

    if app.get('uid') != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to view API keys for this app')

    keys = list_api_keys_db(app_id)
    return keys


@router.delete('/v1/apps/{app_id}/keys/{key_id}', tags=['v1'], response_model=AppStatusMessageResponse)
def delete_api_key(app_id: str, key_id: str, uid: str = Depends(auth.get_current_user_uid)):
    app = get_available_app_by_id(app_id, uid)
    if not app:
        raise HTTPException(status_code=404, detail='App not found')

    if app.get('uid') != uid:
        raise HTTPException(status_code=403, detail='You are not authorized to delete API keys for this app')

    delete_api_key_db(app_id, key_id)

    return {'status': 'ok', 'message': 'API key deleted'}


# ******************************************************
# ******** CONVERSATION SUMMARY APP IDS ****************
# ******************************************************


@router.get('/v1/summary-app-ids', tags=['v1'], response_model=ConversationSummaryAppIdsResponse)
def get_summary_app_ids(secret_key: str = Header(...)):
    """Get all conversation summary app IDs from Redis"""
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='Forbidden')

    app_ids = get_conversation_summary_app_ids()
    logger.info(app_ids)
    return {'app_ids': app_ids or []}


@router.post('/v1/summary-app-ids/{app_id}', tags=['v1'], response_model=AppStatusMessageResponse)
def add_summary_app_id(app_id: str, secret_key: str = Header(...)):
    """Add an app ID to the conversation summary apps list"""
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='Forbidden')

    success = add_conversation_summary_app_id(app_id)
    if success:
        return {'status': 'ok', 'message': f'App {app_id} added to conversation summary apps'}
    else:
        return {'status': 'ok', 'message': f'App {app_id} already exists in conversation summary apps'}


@router.delete('/v1/summary-app-ids/{app_id}', tags=['v1'], response_model=AppStatusMessageResponse)
def delete_summary_app_id(app_id: str, secret_key: str = Header(...)):
    """Remove an app ID from the conversation summary apps list"""
    if secret_key != os.getenv('ADMIN_KEY'):
        raise HTTPException(status_code=403, detail='Forbidden')

    success = remove_conversation_summary_app_id(app_id)
    if success:
        return {'status': 'ok', 'message': f'App {app_id} removed from conversation summary apps'}
    else:
        raise HTTPException(status_code=404, detail=f'App {app_id} not found in conversation summary apps')