apps.py

import asyncio
import hashlib
import math
import os
import secrets
from collections import defaultdict
from datetime import datetime, timezone
from typing import List, Tuple, Dict, Any, Optional, Set, cast

import httpx
from fastapi import HTTPException
from pydantic import ValidationError
from database.cache import get_memory_cache, get_pubsub_manager
from database.redis_db import delete_generic_cache
from database.apps import (
    PUBLIC_APPROVED_APPS_CACHE_KEY,
    get_private_apps_db,
    get_public_unapproved_apps_db,
    get_public_approved_apps_db,
    get_app_by_id_db,
    get_app_usage_history_db,
    set_app_review_in_db,
    get_app_usage_count_db,
    get_app_memory_created_integration_usage_count_db,
    get_app_memory_prompt_usage_count_db,
    add_tester_db,
    add_app_access_for_tester_db,
    remove_app_access_for_tester_db,
    remove_tester_db,
    is_tester_db,
    can_tester_access_app_db,
    get_apps_for_tester_db,
    get_app_chat_message_sent_usage_count_db,
    update_app_in_db,
    get_audio_apps_count,
    get_persona_by_uid_db,
    update_persona_in_db,
    get_omi_personas_by_uid_db,
    get_api_key_by_hash_db,
    get_popular_apps_db,
)
from database.auth import get_user_name
from database.conversations import get_conversations
from database.memories import get_memories
from database._client import db as firestore_db
from utils.memory.memory_service import MemoryService
from database.redis_db import (
    get_enabled_apps,
    get_app_reviews,
    get_generic_cache,
    set_generic_cache,
    set_app_usage_history_cache,
    get_app_usage_history_cache,
    get_app_money_made_cache,
    set_app_money_made_cache,
    get_apps_installs_count,
    get_apps_reviews,
    get_app_cache_by_id,
    set_app_cache_by_id,
    set_app_review_cache,
    get_app_usage_count_cache,
    set_app_money_made_amount_cache,
    get_app_money_made_amount_cache,
    set_app_usage_count_cache,
    set_user_paid_app,
    get_user_paid_app,
    delete_app_cache_by_id,
    is_username_taken,
    get_user_app_subscription_customer_id,
    set_user_app_subscription_customer_id,
    can_update_persona,
    set_persona_update_timestamp,
)
from database.users import get_stripe_connect_account_id
from models.app import App, UsageHistoryItem, UsageHistoryType
from utils.conversations.factory import deserialize_conversations
from utils.conversations.render import conversations_to_string
from utils import stripe
from utils.llm.persona import condense_conversations, condense_memories, generate_persona_description, condense_tweets
from utils.llm.usage_tracker import track_usage, Features
from utils.executors import run_blocking, db_executor, llm_executor
from utils.social import get_twitter_timeline
import logging

logger = logging.getLogger(__name__)

_reviewers_env: Optional[str] = os.getenv('MARKETPLACE_APP_REVIEWERS')
MarketplaceAppReviewUIDs: List[str] = _reviewers_env.split(',') if _reviewers_env else []


def _records_with_ids(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Drop marketplace records with no id, logging how many were skipped.

    `_safe_build_app` already skips a record the `App` model rejects, but every list builder reads
    `app['id']` first — to batch the installs and reviews lookups, and again per app — on the raw
    dicts the Redis cache hands back. That is upstream of any model construction, so one legacy
    document without an id raises KeyError out of a builder that is cached and shared across users.
    """
    usable = [record for record in records if record.get('id')]
    skipped = len(records) - len(usable)
    if skipped:
        logger.warning("Skipping %d marketplace app record(s) without an id", skipped)
    return usable


def _safe_build_app(app_dict: dict[str, Any]) -> Optional[App]:
    """Build an App from a raw marketplace record, skipping (not raising on) a malformed one.

    The marketplace list builders are shared and Redis/process-cached across all users, so one
    legacy or malformed app document must not 500 the whole listing for everyone. Returns None
    for a record that fails validation, logging the app id and the offending field names only.
    """
    try:
        return App(**app_dict)
    except ValidationError as e:
        logger.warning(
            "Skipping malformed marketplace app %s: %s",
            app_dict.get('id'),
            [err['loc'][0] for err in e.errors()],
        )
        return None


def validate_app_endpoints_for_reenable(app_dict: Dict[str, Any], update_dict: Dict[str, Any], app_id: str) -> None:
    """Validate all configured endpoints before allowing a disabled app to be re-enabled.

    Raises HTTPException(400) if any endpoint is unreachable or unhealthy.
    """
    updated_ext_raw: object = update_dict.get('external_integration')
    updated_ext: Dict[str, Any] = cast(Dict[str, Any], updated_ext_raw) if isinstance(updated_ext_raw, dict) else {}
    existing_ext_raw: object = app_dict.get('external_integration')
    existing_ext: Dict[str, Any] = cast(Dict[str, Any], existing_ext_raw) if isinstance(existing_ext_raw, dict) else {}
    endpoints_to_check: List[Tuple[str, str, str, bool]] = []
    seen_urls: Set[str] = set()
    webhook_url: Any = updated_ext.get('webhook_url') or existing_ext.get('webhook_url', '')
    if webhook_url:
        endpoints_to_check.append(('webhook', str(webhook_url), 'POST', True))
        seen_urls.add(str(webhook_url))
    mcp_url: Any = updated_ext.get('mcp_server_url') or existing_ext.get('mcp_server_url', '')
    if mcp_url:
        endpoints_to_check.append(('MCP server', str(mcp_url), 'POST', False))
        seen_urls.add(str(mcp_url))
    chat_tools_raw: object = update_dict.get('chat_tools') or app_dict.get('chat_tools') or []
    chat_tools: List[Any] = list(cast(List[Any], chat_tools_raw)) if isinstance(chat_tools_raw, list) else []
    for tool in chat_tools:
        if isinstance(tool, dict):
            ep_raw: Any = cast(Dict[str, Any], tool).get('endpoint', '')
        else:
            ep_raw = getattr(tool, 'endpoint', '')
        if ep_raw and str(ep_raw) not in seen_urls:
            endpoints_to_check.append(('chat tool', str(ep_raw), 'HEAD', False))
            seen_urls.add(str(ep_raw))
    if not endpoints_to_check:
        raise HTTPException(
            status_code=400,
            detail='No configured endpoints found. Add a webhook URL, MCP server, or chat tool before re-enabling.',
        )
    for label, url, method, require_2xx in endpoints_to_check:
        try:
            # Must match delivery, which pins the destination IP and so cannot follow
            # redirects. Checking with redirects followed passed endpoints that then
            # failed on the very first real webhook.
            resp = httpx.request(method, url, json={}, timeout=10.0, follow_redirects=False)
            if 300 <= resp.status_code < 400:
                raise HTTPException(
                    status_code=400,
                    detail=(
                        f'{label.capitalize()} endpoint redirects ({resp.status_code}). Deliveries do not follow '
                        'redirects — point it at the final URL before re-enabling.'
                    ),
                )
            if require_2xx and (resp.status_code < 200 or resp.status_code >= 300):
                raise HTTPException(
                    status_code=400,
                    detail=f'{label.capitalize()} endpoint returned {resp.status_code}. Fix it before re-enabling.',
                )
        except httpx.TimeoutException:
            raise HTTPException(
                status_code=400, detail=f'{label.capitalize()} endpoint timed out. Fix it before re-enabling.'
            )
        except httpx.ConnectError:
            raise HTTPException(
                status_code=400, detail=f'Cannot connect to {label} endpoint. Fix it before re-enabling.'
            )
        except HTTPException:
            raise
        except Exception as e:
            logger.warning(f'{label.capitalize()} health check failed for {app_id}: {e}')
            raise HTTPException(
                status_code=400, detail=f'{label.capitalize()} health check failed. Fix it before re-enabling.'
            )


# ********************************
# ************ TESTER ************
# ********************************


def is_tester(uid: str) -> bool:
    return is_tester_db(uid)


def can_tester_access_app(uid: str, app_id: str) -> bool:
    return can_tester_access_app_db(app_id, uid)


def _invalidate_tester_cache(uid: str) -> None:
    """Invalidate tester-related caches after mutation."""
    cache = get_memory_cache()
    cache.delete(f"is_tester:{uid}")
    # Delete both tester=0 and tester=1 variants
    cache.delete(f"user_apps_slice:{uid}:0")
    cache.delete(f"user_apps_slice:{uid}:1")


def add_tester(data: Dict[str, Any]) -> None:
    add_tester_db(data)
    uid = data.get('uid')
    if uid:
        _invalidate_tester_cache(cast(str, uid))


def remove_tester(uid: str) -> None:
    remove_tester_db(uid)
    _invalidate_tester_cache(uid)


def add_app_access_for_tester(app_id: str, uid: str) -> None:
    add_app_access_for_tester_db(app_id, uid)
    _invalidate_tester_cache(uid)


def remove_app_access_for_tester(app_id: str, uid: str) -> None:
    remove_app_access_for_tester_db(app_id, uid)
    _invalidate_tester_cache(uid)


# ********************************


def _clamp_review_score(score: Any) -> float:
    # App reviews are a 0-5 scale. Clamp so a drifted or abusive out-of-range score cannot skew
    # rating_avg and the marketplace ranking (weighted_rating / compute_app_score) that reads it.
    # The read path already bounds score with Field(ge=0, le=5); this closes the same bound on the
    # write and aggregation paths, where the request model leaves score unbounded.
    try:
        return max(0.0, min(5.0, float(score)))
    except (TypeError, ValueError):
        return 0.0


def weighted_rating(app: App) -> float:
    C = 3.0  # Assume 3.0 is the mean rating across all apps
    m = 5  # Minimum number of ratings required to be considered
    R = app.rating_avg or 0
    v = app.rating_count or 0
    return (v / (v + m) * R) + (m / (v + m) * C)


def compute_app_score(app: App) -> float:
    """
    Compute app ranking score using the formula:
    score = ((rating_avg / 5) ** 2) * log(1 + rating_count) * sqrt(log(1 + installs))

    - Power of 2 on rating makes ratings below 3.0 fall steeply
    - sqrt on installs reduces dependence on install count

    Rating factor with power of 2:
      5.0 -> 1.0, 4.0 -> 0.64, 3.0 -> 0.36, 2.0 -> 0.16, 1.0 -> 0.04
    """
    rating_avg = app.rating_avg or 0
    rating_count = app.rating_count or 0
    # Clamp negative install counts (counter drift) so math.log(1 + installs) below never hits a domain
    # error; the source is also floored in redis_db.get_apps_installs_count.
    installs = max(0, app.installs or 0)

    rating_factor = (rating_avg / 5) ** 2  # Steep drop for low ratings
    score = rating_factor * math.log(1 + rating_count) * math.sqrt(math.log(1 + installs))
    return round(score, 4)


def invalidate_popular_apps_cache() -> None:
    """Invalidate the popular apps cache across all backend instances."""
    memory_cache = get_memory_cache()
    pubsub_manager = get_pubsub_manager()

    cache_key = 'get_popular_apps_data'

    # Clear local memory cache
    memory_cache.delete(cache_key)

    # Clear Redis cache
    delete_generic_cache(cache_key)

    # Notify all other instances
    pubsub_manager.publish_invalidation([cache_key])


def get_popular_apps() -> List[App]:
    cache_key = 'get_popular_apps_data'
    memory_cache = get_memory_cache()

    def fetch_and_process() -> List[App]:
        """Fetch from Redis/DB and process apps (called only once with singleflight)."""
        # Check Redis cache
        popular_apps: List[Dict[str, Any]]
        if cached_apps := get_generic_cache(cache_key):
            logger.info('get_popular_apps from Redis cache')
            popular_apps = cast(List[Dict[str, Any]], cached_apps)
        else:
            # Database query
            logger.info('get_popular_apps from db')
            popular_apps = get_popular_apps_db()
            # Reduce cache size by excluding large fields
            reduced_apps = [App.reduce_dict(app) for app in popular_apps]
            set_generic_cache(cache_key, reduced_apps, 60 * 30)  # 30 minutes cached
            popular_apps = reduced_apps

        usable_apps = _records_with_ids(popular_apps)

        # Process apps (add installs, reviews, ratings)
        app_ids = [app['id'] for app in usable_apps]
        apps_install = get_apps_installs_count(app_ids)
        apps_reviews = get_apps_reviews(app_ids)

        apps: List[App] = []
        for app in usable_apps:
            app_dict = app
            app_dict['installs'] = apps_install.get(app['id'], 0)
            reviews = apps_reviews.get(app['id'], {})
            sorted_reviews = reviews.values()
            rating_avg = (
                sum([_clamp_review_score(x['score']) for x in sorted_reviews]) / len(sorted_reviews)
                if reviews
                else None
            )
            app_dict['rating_avg'] = rating_avg
            app_dict['rating_count'] = len(sorted_reviews)
            built_app = _safe_build_app(app_dict)
            if built_app is not None:
                apps.append(built_app)
        apps = sorted(apps, key=lambda x: x.installs, reverse=True)
        return apps

    # Singleflight: only ONE request fetches, others wait
    return memory_cache.get_or_fetch(cache_key, fetch_and_process, ttl=30) or []


def get_available_apps(uid: str, include_reviews: bool = False) -> List[App]:
    cache_key = PUBLIC_APPROVED_APPS_CACHE_KEY
    memory_cache = get_memory_cache()

    # Cache tester flag per user (30s TTL) to avoid Firestore lookup every 1s (#5439 sub-task 3)
    tester = memory_cache.get_or_fetch(f"is_tester:{uid}", lambda: is_tester(uid), ttl=30)

    def fetch_public_approved() -> List[Dict[str, Any]]:
        """Fetch from Redis or DB (called only once with singleflight)."""
        if cached := get_generic_cache(cache_key):
            logger.info('get_public_approved_apps_data from Redis cache')
            return cast(List[Dict[str, Any]], cached)
        logger.info('get_public_approved_apps_data from db')
        data = get_public_approved_apps_db()
        # Reduce cache size by excluding large fields
        reduced_data = [App.reduce_dict(app) for app in data]
        set_generic_cache(cache_key, reduced_data, 60 * 10)  # 10 minutes cached
        return reduced_data

    # Singleflight: only ONE request fetches, others wait
    public_approved_data: List[Dict[str, Any]] = (
        cast(List[Dict[str, Any]], memory_cache.get_or_fetch(cache_key, fetch_public_approved, ttl=30)) or []
    )

    # Cache per-user app slice (private + unapproved + tester apps) with 30s TTL (#5439 sub-task 3)
    def fetch_user_apps_slice() -> Dict[str, Any]:
        return {
            'private_data': get_private_apps(uid),
            'public_unapproved_data': get_public_unapproved_apps(uid),
            'tester_apps': get_apps_for_tester_db(uid) if tester else [],
        }

    user_slice_raw = memory_cache.get_or_fetch(f"user_apps_slice:{uid}:{int(tester)}", fetch_user_apps_slice, ttl=30)
    user_slice: Dict[str, Any] = cast(Dict[str, Any], user_slice_raw) if user_slice_raw else {}
    private_data: List[Dict[str, Any]] = cast(List[Dict[str, Any]], user_slice.get('private_data', []))
    public_unapproved_data: List[Dict[str, Any]] = cast(
        List[Dict[str, Any]], user_slice.get('public_unapproved_data', [])
    )
    tester_apps: List[Dict[str, Any]] = cast(List[Dict[str, Any]], user_slice.get('tester_apps', []))

    user_enabled: Set[str] = set(get_enabled_apps(uid))
    all_apps: List[Dict[str, Any]] = _records_with_ids(
        private_data + public_approved_data + public_unapproved_data + tester_apps
    )
    apps: List[App] = []

    app_ids = [app['id'] for app in all_apps]
    apps_install = get_apps_installs_count(app_ids)
    apps_review = get_apps_reviews(app_ids) if include_reviews else {}

    for app in all_apps:
        if app.get('disabled'):
            continue
        # Copy dict to avoid mutating cached objects
        app_dict = dict(app)
        app_dict['enabled'] = app['id'] in user_enabled
        app_dict['rejected'] = app.get('approved') is False
        app_dict['installs'] = apps_install.get(app['id'], 0)
        if include_reviews:
            reviews = apps_review.get(app['id'], {})
            sorted_reviews = reviews.values()
            rating_avg = (
                sum([_clamp_review_score(x['score']) for x in sorted_reviews]) / len(sorted_reviews)
                if reviews
                else None
            )
            app_dict['reviews'] = [details for details in reviews.values() if details['review']]
            app_dict['user_review'] = reviews.get(uid)
            app_dict['rating_avg'] = rating_avg
            app_dict['rating_count'] = len(sorted_reviews)
        built_app = _safe_build_app(app_dict)
        if built_app is not None:
            apps.append(built_app)
    if include_reviews:
        apps.sort(key=weighted_rating, reverse=True)
    return apps


def get_available_app_model_by_id(app_id: str, uid: str | None) -> Optional[App]:
    """`get_available_app_by_id` as a validated App model.

    This is the same availability authority the set-preferred-app route uses
    (routers/users.py), for readers that must honor what that route admitted
    rather than re-deciding availability with a different check (#10074).
    """
    raw_app = get_available_app_by_id(app_id, uid)
    return _safe_build_app(dict(raw_app)) if raw_app else None


def get_available_app_by_id(app_id: str, uid: str | None) -> Dict[str, Any] | None:
    cached_app = get_app_cache_by_id(app_id)
    if cached_app:
        logger.info('get_app_cache_by_id from cache')
        if cached_app['private'] and cached_app.get('uid') != uid and not (uid and is_tester(uid)):
            return None
        return cached_app
    app = get_app_by_id_db(app_id)
    if not app:
        return None
    if app['private'] and app.get('uid') != uid and not (uid and is_tester(uid)):
        return None
    set_app_cache_by_id(app_id, app)
    return app


def get_available_app_by_id_with_reviews(app_id: str, uid: str | None) -> Dict[str, Any] | None:
    app = get_app_by_id_db(app_id)
    if not app:
        return None
    if app['private'] and app.get('uid') != uid and not (uid and is_tester(uid)):
        return None
    app['money_made'] = get_app_money_made_amount(app['id']) if not app['private'] else None
    app['usage_count'] = get_app_usage_count(app['id']) if not app['private'] else None
    reviews = get_app_reviews(app['id'])
    sorted_reviews = reviews.values()
    rating_avg = (
        sum([_clamp_review_score(x['score']) for x in sorted_reviews]) / len(sorted_reviews) if reviews else None
    )
    app['reviews'] = [details for details in reviews.values() if details['review']]
    app['rating_avg'] = rating_avg
    app['rating_count'] = len(sorted_reviews)
    app['user_review'] = reviews.get(uid) if uid else None

    # enabled
    user_enabled: Set[str] = set(get_enabled_apps(uid)) if uid else set()
    app['enabled'] = app['id'] in user_enabled

    # install
    apps_install = get_apps_installs_count([app['id']])
    app['installs'] = apps_install.get(app['id'], 0)
    return app


def get_public_unapproved_apps(uid: str) -> List[Dict[str, Any]]:
    data = get_public_unapproved_apps_db(uid)
    return data


def get_private_apps(uid: str) -> List[Dict[str, Any]]:
    data = get_private_apps_db(uid)
    return data


def invalidate_approved_apps_cache() -> None:
    """
    Invalidate the approved apps cache across all backend instances.

    This function:
    1. Invalidates memory cache on local instance
    2. Invalidates Redis cache
    3. Publishes invalidation message to all other instances via pub/sub
    """
    # Get cache instances
    memory_cache = get_memory_cache()
    pubsub_manager = get_pubsub_manager()

    # Invalidate both cache key variants (with and without reviews)
    cache_keys = [f'{PUBLIC_APPROVED_APPS_CACHE_KEY}:reviews={n}' for n in (0, 1)]

    # Clear local memory cache
    for key in cache_keys:
        memory_cache.delete(key)

    # Clear Redis cache
    delete_generic_cache(PUBLIC_APPROVED_APPS_CACHE_KEY)

    # Notify all other instances to clear their memory cache
    pubsub_manager.publish_invalidation(cache_keys)


def get_approved_available_apps(include_reviews: bool = False) -> list[App]:
    # Use separate cache keys for with/without reviews
    cache_key = f'{PUBLIC_APPROVED_APPS_CACHE_KEY}:reviews={int(include_reviews)}'
    redis_cache_key = PUBLIC_APPROVED_APPS_CACHE_KEY
    memory_cache = get_memory_cache()

    def fetch_and_process() -> List[App]:
        """Fetch from Redis/DB and process apps (called only once with singleflight)."""
        # Check Redis cache
        all_apps: List[Dict[str, Any]]
        if cached_apps := get_generic_cache(redis_cache_key):
            logger.info('get_public_approved_apps_data from Redis cache')
            all_apps = cast(List[Dict[str, Any]], cached_apps)
        else:
            # Database query
            logger.info('get_public_approved_apps_data from db')
            all_apps = get_public_approved_apps_db()
            # Reduce cache size by excluding large fields
            reduced_apps = [App.reduce_dict(app) for app in all_apps]
            set_generic_cache(redis_cache_key, reduced_apps, 60 * 10)  # 10 minutes cached
            all_apps = reduced_apps

        usable_apps = _records_with_ids(all_apps)

        # Process apps (add installs, reviews, etc.)
        app_ids = [app['id'] for app in usable_apps]
        apps_installs = get_apps_installs_count(app_ids)
        apps_reviews = get_apps_reviews(app_ids) if include_reviews else {}

        apps: List[App] = []
        for app in usable_apps:
            if app.get('disabled'):
                continue
            app_dict = app
            app_dict['installs'] = apps_installs.get(app['id'], 0)
            if include_reviews:
                reviews = apps_reviews.get(app['id'], {})
                sorted_reviews = reviews.values()
                rating_avg = (
                    sum([_clamp_review_score(x['score']) for x in sorted_reviews]) / len(sorted_reviews)
                    if reviews
                    else None
                )
                app_dict['reviews'] = []
                app_dict['rating_avg'] = rating_avg
                app_dict['rating_count'] = len(sorted_reviews)
            built_app = _safe_build_app(app_dict)
            if built_app is not None:
                apps.append(built_app)
        if include_reviews:
            apps.sort(key=weighted_rating, reverse=True)
        return apps

    # Singleflight: only ONE request fetches, others wait
    return memory_cache.get_or_fetch(cache_key, fetch_and_process, ttl=30) or []


def set_app_review(app_id: str, uid: str, review: Dict[str, Any]) -> Dict[str, str]:
    if 'score' in review:
        review['score'] = _clamp_review_score(review['score'])
    set_app_review_in_db(app_id, uid, review)
    set_app_review_cache(app_id, uid, review)
    return {'status': 'ok'}


def get_app_usage_count(app_id: str) -> int:
    cached_count = get_app_usage_count_cache(app_id)
    if cached_count:
        return cached_count
    usage = get_app_usage_count_db(app_id)
    set_app_usage_count_cache(app_id, usage)
    return usage


def get_app_money_made_amount(app_id: str) -> float:
    cached_money = get_app_money_made_amount_cache(app_id)
    if cached_money:
        return cached_money
    type1_usage = get_app_memory_created_integration_usage_count_db(app_id)
    type2_usage = get_app_memory_prompt_usage_count_db(app_id)
    type3_usage = get_app_chat_message_sent_usage_count_db(app_id)

    # tbd based on current prod stats
    t1multiplier = 0.02
    t2multiplier = 0.01
    t3multiplier = 0.005

    amount = round((type1_usage * t1multiplier) + (type2_usage * t2multiplier) + (type3_usage * t3multiplier), 2)
    set_app_money_made_amount_cache(app_id, amount)
    return amount


def _safe_usage_history_items(usage: List[Dict[str, Any]], app_id: str) -> List[UsageHistoryItem]:
    """Build UsageHistoryItem records from raw usage docs, skipping (not raising on) a malformed one.

    get_app_usage_history / get_app_money_made are Redis/process-cached and shared, so one legacy or
    malformed usage document (a bad type enum, a missing timestamp) must not 500 the whole enrichment.
    """
    items: List[UsageHistoryItem] = []
    for x in usage:
        try:
            items.append(UsageHistoryItem(**x))
        except ValidationError as e:
            logger.warning(
                "Skipping malformed usage history item for app %s: %s",
                app_id,
                [err['loc'][0] for err in e.errors()],
            )
    return items


def get_app_usage_history(app_id: str) -> List[Dict[str, Any]]:
    cached_usage = get_app_usage_history_cache(app_id)
    if cached_usage:
        return cached_usage
    usage = get_app_usage_history_db(app_id)
    usage = _safe_usage_history_items(usage, app_id)
    # return usage by date grouped count
    by_date: 'defaultdict[Any, int]' = defaultdict(int)
    for item in usage:
        date = item.timestamp.date()
        if date > datetime(2024, 11, 1, tzinfo=timezone.utc).date():
            by_date[date] += 1

    data: List[Dict[str, Any]] = [{'date': k, 'count': v} for k, v in by_date.items()]
    data = sorted(data, key=lambda x: x['date'])
    set_app_usage_history_cache(app_id, data)
    return data


def get_app_money_made(app_id: str) -> dict[str, int | float]:
    cached_money = get_app_money_made_cache(app_id)
    if cached_money:
        return cached_money
    usage = get_app_usage_history_db(app_id)
    usage = _safe_usage_history_items(usage, app_id)
    type1 = len(list(filter(lambda x: x.type == UsageHistoryType.memory_created_external_integration, usage)))
    type2 = len(list(filter(lambda x: x.type == UsageHistoryType.memory_created_prompt, usage)))
    type3 = len(list(filter(lambda x: x.type == UsageHistoryType.chat_message_sent, usage)))
    _type4 = len(list(filter(lambda x: x.type == UsageHistoryType.transcript_processed_external_integration, usage)))

    # tbd based on current prod stats
    t1multiplier = 0.02
    t2multiplier = 0.01
    t3multiplier = 0.005
    _t4multiplier = 0.00001  # This is for transcript processed triggered for every segment, so it should be very low

    money = {
        'money': round((type1 * t1multiplier) + (type2 * t2multiplier) + (type3 * t3multiplier), 2),
        'type1': type1,
        'type2': type2,
        'type3': type3,
    }

    set_app_money_made_cache(app_id, money)

    return money


def upsert_app_payment_link(
    app_id: str, is_paid_app: bool, price: Any, payment_plan: str, uid: str, previous_price: float | None = None
):
    if not is_paid_app:
        logger.info(f"App is not a paid app, app_id: {app_id}")
        return None

    if payment_plan not in ['monthly_recurring']:
        logger.error(f"App payment plan is invalid, app_id: {app_id}")
        return None

    app_data = get_app_by_id_db(app_id)
    if not app_data:
        logger.warning(f"App is not found, app_id: {app_id}")
        return None

    app = App(**app_data)

    if previous_price and previous_price == price:
        logger.info(f"App price is existing, app_id: {app_id}")
        return app

    # A paid app needs a positive numeric price before we can build a Stripe link. update_app passes the
    # raw request price straight through, so a null price (is_paid toggled on without a price) or a
    # non-numeric value would reach int(price * 100) below and raise, 500ing the update. Treat any
    # non-positive or non-numeric price like the existing price==0 case: skip link creation, no crash.
    if not isinstance(price, (int, float)) or isinstance(price, bool) or price <= 0:
        logger.error(f"App price is missing or not a positive number, app_id: {app_id}")
        return app

    # create recurring payment link
    if payment_plan == 'monthly_recurring':
        stripe_acc_id: str = get_stripe_connect_account_id(uid) or ''

        # product
        if not app.payment_product_id:
            payment_product = stripe.create_product(f"{app.name} Monthly Plan", app.description, app.image)
            app.payment_product_id = payment_product.id

        # price
        payment_price = stripe.create_app_monthly_recurring_price(app.payment_product_id, int(round(price * 100)))
        app.payment_price_id = payment_price.id

        # payment link
        payment_link = stripe.create_app_payment_link(app.payment_price_id, app.id, stripe_acc_id)
        app.payment_link_id = payment_link.id
        app.payment_link = payment_link.url

    # updates
    update_app_in_db(app.model_dump())
    return app


def get_is_user_paid_app(app_id: str, uid: str):
    if uid in MarketplaceAppReviewUIDs:
        return True
    return get_user_paid_app(app_id, uid) is not None


def is_permit_payment_plan_get(uid: str):
    if uid in MarketplaceAppReviewUIDs:
        return False

    return True


def paid_app(app_id: str, uid: str):
    expired_seconds = 60 * 60 * 24 * 30  # 30 days
    set_user_paid_app(app_id, uid, expired_seconds)


def set_user_app_sub_customer_id(app_id: str, uid: str, customer_id: str):
    set_user_app_subscription_customer_id(app_id, uid, customer_id)


def find_app_subscription(app_id: str, uid: str, status_filter: str = 'all') -> Dict[str, Any] | None:
    """
    Find a user's subscription for a specific app using cached customer ID or metadata search.

    Args:
        app_id: The app ID to search for
        uid: The user ID
        status_filter: Stripe subscription status filter ('all', 'active', etc.)

    Returns:
        Dictionary representation of the subscription or None if not found
    """
    try:
        cached_customer_id = get_user_app_subscription_customer_id(app_id, uid)
        latest_subscription = None

        if cached_customer_id:
            latest_subscription = stripe.find_app_subscription_by_customer_id(
                cached_customer_id, app_id, uid, status_filter
            )

            if latest_subscription is None:
                cached_customer_id = None

        if not latest_subscription and not cached_customer_id:
            latest_subscription = stripe.find_app_subscription_by_metadata(app_id, uid, status_filter)

            # Cache the customer ID for future lookups
            if latest_subscription and latest_subscription.get('customer'):
                set_user_app_subscription_customer_id(app_id, uid, str(latest_subscription.get('customer')))

        return latest_subscription
    except Exception as e:
        logger.error(f"Error finding app subscription: {e}")
        return None


def is_audio_bytes_app_enabled(uid: str):
    enabled_apps = get_enabled_apps(uid)
    # https://firebase.google.com/docs/firestore/query-data/queries#in_and_array-contains-any
    limit = 30
    enabled_apps = list(set(enabled_apps))
    for i in range(0, len(enabled_apps), limit):
        audio_apps_count = get_audio_apps_count(enabled_apps[i : i + limit])
        if audio_apps_count > 0:
            return True
    return False


def get_persona_by_uid(uid: str):
    persona = get_persona_by_uid_db(uid)
    if persona:
        return persona
    return None


def get_omi_personas_by_uid(uid: str):
    personas = get_omi_personas_by_uid_db(uid)
    if personas:
        return personas
    return None


async def generate_persona_prompt(uid: str, persona: Dict[str, Any]):
    """Generate a persona prompt based on user memories and conversations."""

    # Get latest memories and user info — exclude locked content
    all_memories = await run_blocking(db_executor, get_memories, uid, limit=250)
    memories = [m for m in all_memories if not m.get('is_locked')]
    user_name = await run_blocking(db_executor, get_user_name, uid)

    # Get and condense recent conversations — exclude locked content
    all_conversations = await run_blocking(db_executor, get_conversations, uid, limit=10)
    conversations = deserialize_conversations([c for c in all_conversations if not c.get('is_locked')])
    conversation_history = conversations_to_string(conversations)
    with track_usage(uid, Features.PERSONA):
        conversation_history = await run_blocking(llm_executor, condense_conversations, [conversation_history])

    tweets = None
    if "twitter" in persona['connected_accounts']:
        logger.info("twitter is in connected accounts")
        # Get latest tweets
        timeline = await get_twitter_timeline(persona['twitter']['username'])
        tweets = [{'tweet': tweet.text, 'posted_at': tweet.created_at} for tweet in timeline.timeline]

    # Condense memories
    with track_usage(uid, Features.PERSONA):
        memories_text = await run_blocking(
            llm_executor, condense_memories, [memory['content'] for memory in memories], user_name or ""
        )

    # Generate updated chat prompt
    persona_prompt = f"""
    You are {user_name} AI. Your objective is to personify {user_name} as accurately as possible for 1:1 cloning.

    You have:
    - **Condensed Facts:** Core identity, personality, values, behavioral patterns, preferences, and continuity for ongoing projects or relationships.
    - **Condensed Context from Recent Conversations:** Key themes, communication nuances, emotional undertones, decision-making patterns, and conversational flow.
    """

    # Add condensed tweets if available
    if tweets:
        persona_prompt += (
            f"- **Condensed Tweets:** Recent tweets from {user_name} for additional context and personality insights.\n"
        )

    persona_prompt += f"""
    **Style:**
    - Highly interactive and opinionated.
    - Strong, slightly polarizing opinions based on {user_name}'s identity.
    - Reference personal facts naturally.
    - 2-3 sentence responses, never more than 20 words.
    - Maintain conversational flow with coherent, context-aware replies.

    **Guidelines:**
    1. Seamlessly integrate facts and recent contexts for continuity, recency, and relevance.
    2. Use the condensed facts as foundational knowledge and ongoing continuity.
    3. Leverage recent conversations to maintain situational awareness and contextual accuracy.
    4. Emulate {user_name}'s humor, emotional responses, vocabulary, and tone.
    5. Respond with the appropriate level of detail and complexity as {user_name} would.
    6. Exhibit the same thought processes, decision-making patterns, and conversational dynamics.
    """

    # Add a guideline about tweets if they exist
    if tweets:
        persona_prompt += "7. Utilize tweets to enhance authenticity, incorporating common expressions, opinions, and phrasing from {user_name}'s social media presence.\n"

    persona_prompt += f"""
    **Rules:**
    - Never mention being AI.
    - No ** or formal language.
    - No unnecessary text or explanations.
    - Maintain the illusion of continuity and authenticity at all times.
    - Every response must be indistinguishable from how {user_name} would naturally respond.

    You have all the necessary condensed facts and contextual knowledge. Begin personifying {user_name} now.

    Personal Facts and Context:
    {memories_text}

    Recent Conversations:
    {conversation_history}

    Recent Tweets:
    {tweets}

    Use these facts, conversations and tweets to shape your personality. Responses should help the conversation go deeper. You have strong opinions, slightly controversial."""
    return persona_prompt


def generate_persona_desc(uid: str, persona_name: str):
    """Generate a persona description based on user memories."""
    memories = get_memories(uid, limit=250)

    with track_usage(uid, Features.PERSONA):
        persona_description = generate_persona_description(memories, persona_name)
    return persona_description


def update_personas_async(uid: str):
    if not can_update_persona(uid):
        logger.info(f"[PERSONAS] Rate limited - uid={uid} already updated today")
        return

    logger.info(f"[PERSONAS] Starting persona updates in background thread for uid={uid}")
    personas = get_omi_personas_by_uid_db(uid)
    if personas:
        set_persona_update_timestamp(uid)

        async def _batch():
            await asyncio.gather(*[update_persona_prompt(persona) for persona in personas])

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            loop.run_until_complete(_batch())
        except Exception as e:
            logger.error(f"Error in persona batch update for uid={uid}: {str(e)}")
        finally:
            loop.close()
        logger.info(f"[PERSONAS] Finished persona updates in background thread for uid={uid}")
    else:
        logger.info(f"[PERSONAS] No personas found for uid={uid}")


async def update_persona_prompt(persona: Dict[str, Any]):
    """Update a persona's chat prompt with latest memories and conversations."""
    uid = persona['uid']
    universal_memories = await run_blocking(
        db_executor,
        MemoryService(db_client=firestore_db).read,
        uid,
        limit=250,
        offset=0,
    )
    memories = [memory.dict() for memory in universal_memories if memory.visibility == 'public']
    user_name = await run_blocking(db_executor, get_user_name, uid)

    # Get and condense recent conversations
    all_conversations = await run_blocking(db_executor, get_conversations, uid, limit=10)
    conversations = deserialize_conversations(all_conversations)
    conversation_history = conversations_to_string(conversations)
    with track_usage(uid, Features.PERSONA):
        conversation_history = await run_blocking(llm_executor, condense_conversations, [conversation_history])

    condensed_tweets = None
    # Condense tweets
    if "twitter" in persona['connected_accounts'] and 'twitter' in persona:
        # Get latest tweets
        timeline = await get_twitter_timeline(persona['twitter']['username'])
        tweets = [tweet.text for tweet in timeline.timeline]
        with track_usage(uid, Features.PERSONA):
            condensed_tweets = await run_blocking(llm_executor, condense_tweets, tweets, persona['name'])

    # Condense memories
    with track_usage(uid, Features.PERSONA):
        memories_text = await run_blocking(
            llm_executor, condense_memories, [memory['content'] for memory in memories], user_name or ""
        )

    # Generate updated chat prompt
    persona_prompt = f"""
You are {user_name} AI. Your objective is to personify {user_name} as accurately as possible for 1:1 cloning.

You have:
- **Condensed Facts:** Core identity, personality, values, behavioral patterns, preferences, and continuity for ongoing projects or relationships.
- **Condensed Context from Recent Conversations:** Key themes, communication nuances, emotional undertones, decision-making patterns, and conversational flow.
"""

    # Add condensed tweets if available
    if condensed_tweets:
        persona_prompt += (
            f"- **Condensed Tweets:** Recent tweets from {user_name} for additional context and personality insights.\n"
        )

    persona_prompt += f"""
**Style:**
- Highly interactive and opinionated.
- Strong, slightly polarizing opinions based on {user_name}'s identity.
- Reference personal facts naturally.
- 2-3 sentence responses, never more than 20 words.
- Maintain conversational flow with coherent, context-aware replies.

**Guidelines:**
1. Seamlessly integrate facts and recent contexts for continuity, recency, and relevance.
2. Use the condensed facts as foundational knowledge and ongoing continuity.
3. Leverage recent conversations to maintain situational awareness and contextual accuracy.
4. Emulate {user_name}'s humor, emotional responses, vocabulary, and tone.
5. Respond with the appropriate level of detail and complexity as {user_name} would.
6. Exhibit the same thought processes, decision-making patterns, and conversational dynamics.
"""

    # Add a guideline about tweets if they exist
    if condensed_tweets:
        persona_prompt += "7. Utilize condensed tweets to enhance authenticity, incorporating common expressions, opinions, and phrasing from {user_name}'s social media presence.\n"

    persona_prompt += f"""
**Rules:**
- Never mention being AI.
- No ** or formal language.
- No unnecessary text or explanations.
- Maintain the illusion of continuity and authenticity at all times.
- Every response must be indistinguishable from how {user_name} would naturally respond.

You have all the necessary condensed facts and contextual knowledge. Begin personifying {user_name} now.

Personal Facts and Context:
{memories_text}

Recent Conversations:
{conversation_history}

Recent Tweets:
{condensed_tweets}

Use these facts, conversations and tweets to shape your personality. Responses should help the conversation go deeper. You have strong opinions, slightly controversial."""

    persona['persona_prompt'] = persona_prompt
    persona['updated_at'] = datetime.now(timezone.utc)

    await run_blocking(db_executor, update_persona_in_db, persona)
    await run_blocking(db_executor, delete_app_cache_by_id, persona['id'])


def increment_username(username: str) -> str:
    if is_username_taken(username):
        i = 1
        while is_username_taken(f"{username}{i}"):
            i += 1
        return f"{username}{i}"
    else:
        return username


def generate_api_key() -> Tuple[str, str, str]:
    raw_key = secrets.token_hex(16)  # 16 bytes = 32 hex chars
    hashed_key = hashlib.sha256(raw_key.encode()).hexdigest()
    formatted_label = f"sk_{raw_key[:4]}...{raw_key[-4:]}"
    return f'sk_{raw_key}', hashed_key, formatted_label


def verify_api_key(app_id: str, api_key: str) -> bool:
    if api_key.startswith("sk_"):
        api_key = api_key[3:]
    hashed_key = hashlib.sha256(api_key.encode()).hexdigest()
    stored_key = get_api_key_by_hash_db(app_id, hashed_key)
    return stored_key is not None


def app_has_action(app: Optional[Dict[str, Any]], action_name: str) -> bool:
    """Check if an app has a specific action capability."""
    if not app:
        return False

    if not app.get('external_integration'):
        return False

    ext_int_raw = app['external_integration']
    ext_int: Dict[str, Any] = cast(Dict[str, Any], ext_int_raw) if isinstance(ext_int_raw, dict) else {}
    actions_raw = ext_int.get('actions', [])
    actions: List[Dict[str, Any]] = cast(List[Dict[str, Any]], actions_raw) if isinstance(actions_raw, list) else []
    for action in actions:
        if action.get('action') == action_name:
            return True

    return False


def app_can_create_memories(app: Optional[Dict[str, Any]]) -> bool:
    """Check if an app can create memories (facts)."""
    return app_has_action(app, 'create_memories') or app_has_action(app, 'create_facts')


def app_can_read_memories(app: Optional[Dict[str, Any]]) -> bool:
    """Check if an app can read memories (facts)."""
    return app_has_action(app, 'read_memories') or app_has_action(app, 'read_facts')


def app_can_read_conversations(app: Optional[Dict[str, Any]]) -> bool:
    """Check if an app can read conversations."""
    return app_has_action(app, 'read_conversations')


def app_can_create_conversation(app: Optional[Dict[str, Any]]) -> bool:
    """Check if an app can create a conversation."""
    return app_has_action(app, 'create_conversation')


def is_user_app_enabled(uid: str, app_id: str) -> bool:
    """Check if a specific app is enabled for the user based on Redis cache."""
    user_enabled_apps: Set[str] = set(get_enabled_apps(uid))
    return app_id in user_enabled_apps


# ********************************
# ********* v2 APPS UTILS ********
# ********************************


def normalize_app_numeric_fields(app_dict: Dict[str, Any]) -> Dict[str, Any]:
    """Ensure numeric fields that clients expect as float are emitted as float."""

    def _to_float(value: Any) -> Optional[float]:
        try:
            return float(value) if value is not None else None
        except (ValueError, TypeError):
            return value

    # Normalize fields that should be floats
    for field in ['rating_avg', 'money_made', 'price']:
        if field in app_dict and app_dict[field] is not None:
            app_dict[field] = _to_float(app_dict[field])

    return app_dict


def sort_apps_by_installs_only(apps: List[App]) -> List[App]:
    """Sort apps by install count only (no score calculation)."""
    return sorted(apps, key=lambda a: a.installs, reverse=True)


def sort_apps_by_installs(apps: List[App]) -> List[App]:
    """Sort apps by computed score in descending order.

    Score formula: ((rating_avg / 5) ** 2) * log(1 + rating_count) * sqrt(log(1 + installs))
    This balances review quality, review quantity, and popularity.
    """
    # Compute and assign scores to each app
    for app in apps:
        app.score = compute_app_score(app)

    return sorted(apps, key=lambda a: a.score or 0, reverse=True)


def paginate_apps(apps: List[App], offset: int, limit: int) -> List[App]:
    """Apply pagination to apps list."""
    return apps[offset : offset + limit]


def build_pagination_metadata(total: int, offset: int, limit: int, category: Optional[str] = None) -> Dict[str, Any]:
    """Build pagination metadata for API response."""
    has_next = (offset + limit) < total
    has_previous = offset > 0

    metadata: Dict[str, Any] = {
        'total': total,
        'count': max(0, min(limit, total - offset)),
        'offset': offset,
        'limit': limit,
        'hasNext': has_next,
        'hasPrevious': has_previous,
    }

    # Add navigation links if category is specified
    if category:
        base_url = f"/v2/apps?category={category}"
        metadata['links'] = {
            'next': f"{base_url}&offset={offset + limit}&limit={limit}" if has_next else None,
            'previous': f"{base_url}&offset={max(offset - limit, 0)}&limit={limit}" if has_previous else None,
        }

    return metadata


def get_capabilities_list() -> List[Dict[str, Any]]:
    """Get the list of app capabilities for grouping."""
    return [
        {'title': 'Featured', 'id': 'popular'},
        {'title': 'Integrations', 'id': 'external_integration'},
        {'title': 'Chat Assistants', 'id': 'chat'},
        {'title': 'Summary Apps', 'id': 'memories'},
        {'title': 'Realtime Notifications', 'id': 'proactive_notification'},
        {'title': 'Tasks', 'id': 'tasks'},
    ]


def _app_has_auth_steps(app: App) -> bool:
    """Check if app has external_integration with auth_steps."""
    has_external_integration = 'external_integration' in (app.capabilities or set())
    if not has_external_integration:
        return False
    ext_int = app.external_integration
    if ext_int is None:
        return False
    auth_steps: List[Any] = list(cast(List[Any], getattr(ext_int, 'auth_steps', None) or []))
    return len(auth_steps) > 0


def _is_notification_app(app: App) -> bool:
    """Check if app is a notification/simple integration app.

    Returns True for:
    - Apps with proactive_notification capability
    - Simple integrations (external_integration without auth_steps, chat, or memories)
    """
    app_capabilities = app.capabilities or set()
    if 'proactive_notification' in app_capabilities:
        return True
    has_external_integration = 'external_integration' in app_capabilities
    has_auth_steps = _app_has_auth_steps(app)
    return (
        has_external_integration
        and not has_auth_steps
        and 'chat' not in app_capabilities
        and 'memories' not in app_capabilities
    )


def _get_app_capability(app: App) -> str | None:
    """Determine which capability section an app belongs to.

    Returns the capability id or None if the app doesn't match any section.
    Priority order: external_integration (with auth) > chat > memories > proactive_notification
    """
    app_capabilities = app.capabilities or set()
    has_external_integration = 'external_integration' in app_capabilities
    has_auth_steps = _app_has_auth_steps(app)

    # Notification apps (including simple integrations) go to proactive_notification
    if _is_notification_app(app):
        return 'proactive_notification'

    # External integration with auth_steps
    if has_external_integration and has_auth_steps:
        return 'external_integration'

    # Chat apps (excluding those with external_integration+auth_steps)
    if 'chat' in app_capabilities:
        if not has_external_integration or not has_auth_steps:
            return 'chat'

    # Memories apps (excluding those with chat or external_integration+auth_steps)
    if 'memories' in app_capabilities and 'chat' not in app_capabilities:
        if not has_external_integration or not has_auth_steps:
            return 'memories'

    return None


def group_apps_by_capability(apps: List[App], capabilities: List[Dict[str, Any]]) -> Dict[str, List[App]]:
    """Group apps by capability with enhanced filtering rules.

    Groups:
    - popular: Apps marked as is_popular (sorted by installs)
    - proactive_notification: Apps with proactive_notification OR simple integrations
    - external_integration: Apps with external_integration AND auth_steps
    - chat: Apps with chat (excluding those with external_integration+auth_steps)
    - memories: Apps with memories but no chat (excluding those with external_integration+auth_steps)

    Popular apps are excluded from other sections.
    Notification/simple integration apps are excluded from other sections.
    """
    grouped: defaultdict[str, List[App]] = defaultdict(list)

    # First pass: collect popular apps
    popular_apps = [app for app in apps if getattr(app, 'is_popular', False)]
    popular_app_ids = {app.id for app in popular_apps}
    if popular_apps:
        grouped['popular'] = sort_apps_by_installs_only(popular_apps)

    # Second pass: collect notification apps (exclusive)
    notification_app_ids: Set[str] = set()
    for app in apps:
        if _is_notification_app(app):
            grouped['proactive_notification'].append(app)
            notification_app_ids.add(app.id)

    # Sort notification apps
    if grouped['proactive_notification']:
        grouped['proactive_notification'] = sort_apps_by_installs(grouped['proactive_notification'])

    # Group remaining apps by capability
    for app in apps:
        # Skip popular apps in other sections
        if app.id in popular_app_ids:
            continue

        # Skip notification apps (already processed)
        if app.id in notification_app_ids:
            continue

        capability = _get_app_capability(app)
        if capability and capability != 'proactive_notification':
            grouped[capability].append(app)

    # Sort each capability group by score (except popular which is already sorted by installs)
    for cap_id in grouped:
        if cap_id not in ('popular', 'proactive_notification'):
            grouped[cap_id] = sort_apps_by_installs(grouped[cap_id])

    return grouped


def filter_apps_by_capability(apps: List[App], capability: str) -> List[App]:
    """Filter apps by capability with enhanced filtering rules.

    Note: Unlike group_apps_by_capability (used for main apps page), this does NOT exclude
    popular apps - they should appear on individual capability pages if they match.
    """
    if capability == 'popular':
        return [app for app in apps if getattr(app, 'is_popular', False)]

    filtered_apps: List[App] = []
    for app in apps:
        # Skip notification apps in non-notification sections
        if capability != 'proactive_notification' and _is_notification_app(app):
            continue

        app_capability = _get_app_capability(app)
        if app_capability == capability:
            filtered_apps.append(app)

    return filtered_apps


def build_capability_groups_response(
    grouped_apps: Dict[str, List[App]], capabilities: List[Dict[str, Any]], offset: int, limit: int
) -> List[Dict[str, Any]]:
    """Build the groups response for v2/apps endpoint grouped by capability."""
    id_to_title: Dict[Any, Any] = {c['id']: c['title'] for c in capabilities}

    ordered_keys: List[Any] = [c['id'] for c in capabilities]
    for key in grouped_apps:
        if key not in ordered_keys:
            ordered_keys.append(key)

    groups: List[Dict[str, Any]] = []
    for capability_id in ordered_keys:
        apps = grouped_apps.get(capability_id, [])
        if not apps:
            continue

        total = len(apps)
        page = paginate_apps(apps, offset, limit)

        groups.append(
            {
                'capability': {
                    'id': capability_id,
                    'title': id_to_title.get(capability_id, capability_id.title().replace('_', ' ')),
                },
                'data': [normalize_app_numeric_fields(app.to_reduced_dict()) for app in page],
                'pagination': build_pagination_metadata(total, offset, limit, capability_id),
            }
        )

    return groups


# Base category mapping (used for non-chat capabilities)
_BASE_CATEGORY_MAPPING: Dict[str, str] = {
    # Productivity & Tools
    'personality-emulation': 'productivity-tools',
    'education-and-learning': 'productivity-tools',
    'productivity-and-organization': 'productivity-tools',
    'utilities-and-tools': 'productivity-tools',
    'financial': 'productivity-tools',
    'shopping-and-commerce': 'productivity-tools',
    'news-and-information': 'productivity-tools',
    # Personal & Wellness
    'conversation-analysis': 'personal-wellness',
    'communication-improvement': 'personal-wellness',
    'emotional-and-mental-support': 'personal-wellness',
    'health-and-wellness': 'personal-wellness',
    'safety-and-security': 'personal-wellness',
    'other': 'personal-wellness',
    # Social & Entertainment
    'social-and-relationships': 'social-entertainment',
    'entertainment-and-fun': 'social-entertainment',
    'travel-and-exploration': 'social-entertainment',
}

# Chat-specific overrides (remaps categories to chat-specific master categories)
_CHAT_CATEGORY_OVERRIDES: Dict[str, str] = {
    # Personality Clone (unique to chat)
    'personality-emulation': 'personality-clone',
    # Productivity & Lifestyle (replaces productivity-tools and personal-wellness)
    'education-and-learning': 'productivity-lifestyle',
    'productivity-and-organization': 'productivity-lifestyle',
    'utilities-and-tools': 'productivity-lifestyle',
    'financial': 'productivity-lifestyle',
    'shopping-and-commerce': 'productivity-lifestyle',
    'news-and-information': 'productivity-lifestyle',
    'conversation-analysis': 'productivity-lifestyle',
    'communication-improvement': 'productivity-lifestyle',
    'emotional-and-mental-support': 'productivity-lifestyle',
    'health-and-wellness': 'productivity-lifestyle',
    'safety-and-security': 'productivity-lifestyle',
    'other': 'productivity-lifestyle',
}


def get_master_category_mapping(capability_id: str) -> Dict[str, str]:
    """Get master category mapping for a capability."""
    if capability_id == 'chat':
        return {**_BASE_CATEGORY_MAPPING, **_CHAT_CATEGORY_OVERRIDES}
    return _BASE_CATEGORY_MAPPING


def get_master_categories_list(capability_id: str) -> List[Dict[str, str]]:
    """Get master categories list for a capability."""
    if capability_id == 'chat':
        return [
            {'title': 'Personality Clones', 'id': 'personality-clone'},
            {'title': 'Productivity & Lifestyle', 'id': 'productivity-lifestyle'},
            {'title': 'Social & Entertainment', 'id': 'social-entertainment'},
        ]

    # Default categories for other capabilities
    return [
        {'title': 'Productivity & Tools', 'id': 'productivity-tools'},
        {'title': 'Personal & Lifestyle', 'id': 'personal-wellness'},
        {'title': 'Social & Entertainment', 'id': 'social-entertainment'},
    ]


def group_capability_apps_by_category(apps: List[App], capability_id: str) -> Dict[str, List[App]]:
    """Group apps within a capability by master category."""
    category_mapping = get_master_category_mapping(capability_id)
    default_category = 'productivity-lifestyle' if capability_id == 'chat' else 'personal-wellness'

    grouped: defaultdict[str, List[App]] = defaultdict(list)
    for app in apps:
        original_category_id = app.category if app.category else 'other'
        master_category_id = category_mapping.get(original_category_id, default_category)
        grouped[master_category_id].append(app)

    # Sort each master category by score
    for master_category_id in grouped:
        grouped[master_category_id] = sort_apps_by_installs(grouped[master_category_id])

    return grouped


def build_capability_category_groups_response(
    grouped_apps: Dict[str, List[App]], capability_id: str
) -> List[Dict[str, Any]]:
    """Build response for capability apps grouped by category."""
    master_categories = get_master_categories_list(capability_id)
    id_to_title: Dict[str, str] = {c['id']: c['title'] for c in master_categories}

    ordered_keys: List[str] = [c['id'] for c in master_categories]
    for key in grouped_apps:
        if key not in ordered_keys:
            ordered_keys.append(key)

    groups: List[Dict[str, Any]] = []
    for category_id in ordered_keys:
        apps = grouped_apps.get(category_id, [])
        if not apps:
            continue

        # Capitalize unknown category titles
        title = id_to_title.get(category_id)
        if not title:
            title = ' '.join(word.capitalize() for word in category_id.replace('-', ' ').split())

        groups.append(
            {
                'category': {
                    'id': category_id,
                    'title': title,
                },
                'data': [normalize_app_numeric_fields(app.to_reduced_dict()) for app in apps],
                'count': len(apps),
            }
        )

    return groups


# ********************************
# *** CHAT TOOLS MANIFEST FETCH **
# ********************************


def fetch_app_chat_tools_from_manifest(
    manifest_url: str, timeout: int = 10, force_refresh: bool = False
) -> Dict[str, Any] | None:
    """
    Fetch chat tools definitions from an app's manifest endpoint.

    The manifest endpoint should return a JSON object with a 'tools' array containing
    tool definitions with: name, description, endpoint, method, parameters, auth_required, status_message.

    Implements caching with 2-hour TTL to reduce external requests.

    Args:
        manifest_url: Full URL to the manifest endpoint (e.g., https://my-app.com/.well-known/omi-tools.json)
        timeout: Request timeout in seconds
        force_refresh: If True, bypass cache and fetch fresh data

    Returns:
        Dict with 'tools' (list) and 'proactive_messages_enabled' (bool), or None if fetch fails

    Example manifest response:
    {
        "tools": [
            {
                "name": "add_to_playlist",
                "description": "Add a song to a Spotify playlist",
                "endpoint": "/tools/add_to_playlist",
                "method": "POST",
                "parameters": {
                    "properties": {
                        "song_name": {"type": "string", "description": "Name of the song"},
                        "artist_name": {"type": "string", "description": "Artist name"}
                    },
                    "required": ["song_name"]
                },
                "auth_required": true,
                "status_message": "Adding to playlist..."
            }
        ],
        "proactive_messages": {
            "enabled": true
        }
    }
    """
    if not manifest_url:
        return None

    # Check cache first (unless force refresh)
    cache_key = f'manifest:{manifest_url}'
    if not force_refresh:
        cached_result = get_generic_cache(cache_key)
        if cached_result:
            logger.info(f"✅ Using cached manifest for: {manifest_url}")
            return cached_result

    try:
        logger.info(f"📥 Fetching chat tools manifest from: {manifest_url}")

        response = httpx.get(
            manifest_url,
            timeout=float(timeout),
            headers={'Accept': 'application/json', 'User-Agent': 'Omi-App-Store/1.0'},
        )

        if response.status_code != 200:
            logger.error(f"⚠️ Manifest fetch failed with status {response.status_code}: {manifest_url}")
            return None

        data_raw: object = response.json()

        # Validate response structure
        if not isinstance(data_raw, dict):
            logger.error(f"⚠️ Invalid manifest format (not a dict): {manifest_url}")
            return None

        data: Dict[str, Any] = cast(Dict[str, Any], data_raw)
        tools_raw: object = data.get('tools', [])

        if not isinstance(tools_raw, list):
            logger.error(f"⚠️ Invalid manifest format ('tools' is not a list): {manifest_url}")
            return None

        tools: List[Any] = cast(List[Any], tools_raw)

        # Validate and normalize each tool
        validated_tools: List[Dict[str, Any]] = []
        for tool in tools:
            validated_tool = _validate_tool_definition(tool)
            if validated_tool:
                validated_tools.append(validated_tool)
            else:
                typed_t: Dict[str, Any] = cast(Dict[str, Any], tool) if isinstance(tool, dict) else {}
                tool_name: str = str(typed_t.get('name') or 'unknown')
                logger.error(f"⚠️ Skipping invalid tool in manifest: {tool_name}")

        # Parse chat_messages configuration
        chat_messages_raw: object = data.get('chat_messages', {})
        chat_messages_config: Dict[str, Any] = {}
        if isinstance(chat_messages_raw, dict):
            chat_messages: Dict[str, Any] = cast(Dict[str, Any], chat_messages_raw)
            if chat_messages.get('enabled', False):
                chat_messages_config = {
                    'enabled': True,
                    'target': chat_messages.get('target', 'app'),  # 'main' or 'app', default 'app'
                    'notify': chat_messages.get('notify', True),  # send push notification, default True
                }

        logger.info(
            f"✅ Fetched {len(validated_tools)} chat tools from manifest (chat_messages: {chat_messages_config})"
        )
        result: Dict[str, Any] = {
            'tools': validated_tools if validated_tools else None,
            'chat_messages': chat_messages_config if chat_messages_config else None,
        }

        # Cache for 2 hours (7200 seconds)
        if validated_tools:
            set_generic_cache(cache_key, result, 60 * 60 * 2)

        return result

    except httpx.TimeoutException:
        logger.warning(f"⚠️ Manifest fetch timed out: {manifest_url}")
        return None
    except httpx.RequestError as e:
        logger.error(f"⚠️ Manifest fetch request error: {e}")
        return None
    except ValueError as e:
        logger.error(f"⚠️ Invalid JSON in manifest response: {e}")
        return None
    except Exception as e:
        logger.error(f"⚠️ Unexpected error fetching manifest: {e}")
        return None


def _validate_tool_definition(tool: Dict[str, Any]) -> Dict[str, Any] | None:
    """
    Validate and normalize a single tool definition from the manifest.

    Required fields: name, description, endpoint
    Optional fields: method, parameters, auth_required, status_message

    Returns normalized tool dict or None if invalid.
    """
    typed_tool: Dict[str, Any] = tool

    # Check required fields
    name = typed_tool.get('name')
    description = typed_tool.get('description')
    endpoint = typed_tool.get('endpoint')

    if not name or not isinstance(name, str):
        logger.warning(f"⚠️ Tool missing required 'name' field")
        return None

    if not description or not isinstance(description, str):
        logger.warning(f"⚠️ Tool '{name}' missing required 'description' field")
        return None

    if not endpoint or not isinstance(endpoint, str):
        logger.warning(f"⚠️ Tool '{name}' missing required 'endpoint' field")
        return None

    # Build normalized tool definition
    validated: Dict[str, Any] = {
        'name': name.strip(),
        'description': description.strip(),
        'endpoint': endpoint.strip(),
        'method': (typed_tool.get('method') or 'POST').upper(),
        'auth_required': typed_tool.get('auth_required', True),
    }

    # Optional: status_message
    if typed_tool.get('status_message'):
        validated['status_message'] = str(typed_tool['status_message']).strip()

    # Optional: parameters (JSON schema format)
    parameters = typed_tool.get('parameters')
    if parameters and isinstance(parameters, dict):
        # Validate parameters schema structure
        typed_parameters: Dict[str, Any] = cast(Dict[str, Any], parameters)
        properties_raw = typed_parameters.get('properties')
        if 'properties' in typed_parameters and isinstance(properties_raw, dict):
            validated['parameters'] = {
                'properties': cast(Dict[str, Any], properties_raw),
                'required': (
                    cast(List[Any], typed_parameters.get('required'))
                    if isinstance(typed_parameters.get('required'), list)
                    else []
                ),
            }

    return validated


def app_can_read_tasks(app: Optional[Dict[str, Any]]) -> bool:
    """Check if an app can read tasks."""
    return app_has_action(app, 'read_tasks')