notifications.py

import asyncio
import hashlib
import json
import math
import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple, Union, cast
from firebase_admin import messaging, auth
import database.notifications as notification_db
from utils.executors import db_executor, postprocess_executor, run_blocking
from database.redis_db import (
    set_credit_limit_notification_sent,
    has_credit_limit_notification_been_sent,
    set_silent_user_notification_sent,
    has_silent_user_notification_been_sent,
)
from database.auth import get_user_from_uid
from utils.notification_text import to_plain_text
from .llm.notifications import (
    generate_notification_message,
    generate_credit_limit_notification,
    generate_silent_user_notification,
)
import logging

logger = logging.getLogger(__name__)


def _get_user(uid: str) -> Any:
    return auth.get_user(uid)  # type: ignore[reportUnknownMemberType]  # firebase_admin auth untyped


# iOS bundle ID for APNs
IOS_BUNDLE_ID = 'com.friend-app-with-wearable.ios12'

# Error codes that indicate a token is permanently invalid
PERMANENT_FAILURE_CODES = frozenset(
    [
        'UNREGISTERED',  # App uninstalled
        'INVALID_REGISTRATION_TOKEN',  # Token format invalid
        'NOT_FOUND',  # FCM/APNs token no longer maps to a valid registration
    ]
)


def _generate_tag(content: str) -> str:
    """Generate a 16-char hash tag for deduplication."""
    return hashlib.md5(content.encode()).hexdigest()[:16]


def _generate_notification_tag(user_id: str, title: str, body: str, data: Optional[Dict[str, Any]] = None) -> str:
    """Generate a tag for notification deduplication based on content."""
    content = f"{user_id}:{title}:{body}"
    if data:
        unique_id: str = str(data.get('action_item_id') or data.get('app_id') or data.get('type', ''))
        content += f":{unique_id}"
    return _generate_tag(content)


def _build_android_config(tag: str, priority: str = 'normal', is_data_only: bool = False) -> messaging.AndroidConfig:
    """Build Android configuration with deduplication."""
    config_kwargs: Dict[str, Any] = {
        'collapse_key': tag,
        'priority': priority,
    }
    # Only add notification config if not data-only (Android shows empty notification otherwise)
    if not is_data_only:
        config_kwargs['notification'] = messaging.AndroidNotification(tag=tag)
    return messaging.AndroidConfig(**config_kwargs)


def _build_apns_config(tag: str, is_background: bool = False) -> messaging.APNSConfig:
    """Build APNs configuration with deduplication."""
    headers = {'apns-collapse-id': tag}

    if is_background:
        headers.update(
            {
                'apns-push-type': 'background',
                'apns-priority': '5',
                'apns-topic': IOS_BUNDLE_ID,
            }
        )
        return messaging.APNSConfig(
            headers=headers,
            payload=messaging.APNSPayload(aps=messaging.Aps(content_available=True)),
        )

    return messaging.APNSConfig(headers=headers)


def _build_webpush_config(
    tag: str, title: Optional[str] = None, body: Optional[str] = None, link: Optional[str] = None
) -> messaging.WebpushConfig:
    """Build WebPush configuration for browser notifications.

    Note: WebpushNotification must explicitly include title/body because
    browsers use webpush.notification instead of the top-level notification
    when the webpush block is present.

    fcm_options.link must be an absolute HTTPS URL - relative paths will cause
    FCM to reject the entire message batch with 'WebpushFCMOptions.link must be a HTTPS URL'.
    """
    config_kwargs: Dict[str, Any] = {
        'headers': {
            'Topic': tag,  # For deduplication
            'Urgency': 'high',
        },
        'notification': messaging.WebpushNotification(
            title=title,
            body=body,
            icon='/logo.png',
        ),
    }

    # Only include fcm_options if link is a valid HTTPS URL
    if link and link.startswith('https://'):
        config_kwargs['fcm_options'] = messaging.WebpushFCMOptions(link=link)

    return messaging.WebpushConfig(**config_kwargs)


def _build_message(
    token: str,
    tag: str,
    notification: Optional[messaging.Notification] = None,
    data: Optional[Dict[str, Any]] = None,
    is_background: bool = False,
    priority: str = 'normal',
) -> messaging.Message:
    """Build a complete FCM message with proper platform configs."""
    # Extract title/body for webpush config (browsers need explicit values)
    title: Optional[str] = cast(Any, notification).title if notification else None
    body: Optional[str] = cast(Any, notification).body if notification else None
    # Extract navigate_to for webpush click-through link
    link: Optional[str] = data.get('navigate_to') if data else None

    return messaging.Message(
        token=token,
        notification=notification,
        data=data,
        android=_build_android_config(tag, priority, is_data_only=(notification is None)),
        apns=_build_apns_config(tag, is_background),
        webpush=_build_webpush_config(tag, title, body, link),
    )


def _send_messages(messages: List[messaging.Message]) -> Any:
    """Send one FCM batch through the synchronous Firebase Admin SDK."""
    return cast(Any, messaging.send_each(messages))  # type: ignore[reportUnknownMemberType]


def _collect_send_results(response: Any, tokens: List[str]) -> Tuple[int, List[str]]:
    """Return the successful-send count and permanently invalid tokens."""
    invalid_tokens: List[str] = []
    success_count = 0

    for idx, result in enumerate(response.responses):
        if result.success:
            success_count += 1
        elif result.exception:
            error_code = getattr(result.exception, 'code', None)
            if error_code in PERMANENT_FAILURE_CODES:
                invalid_tokens.append(tokens[idx])
                logger.error(f'Invalid token removed - Error: {error_code}')
            else:
                logger.error(f'FCM send failed: {result.exception}({error_code})')

    return success_count, invalid_tokens


def _send_to_user(
    user_id: str,
    tag: str,
    notification: Optional[messaging.Notification] = None,
    data: Optional[Dict[str, Any]] = None,
    is_background: bool = False,
    priority: str = 'normal',
    tokens: Optional[List[str]] = None,
) -> int:
    """Send a message to all user's devices using batch send. Returns count of successful sends."""
    if tokens is None:
        tokens = notification_db.get_all_tokens(user_id)
    if not tokens:
        logger.info(f"No tokens found for user {user_id}")
        return 0

    # Build messages for all tokens
    messages = [_build_message(token, tag, notification, data, is_background, priority) for token in tokens]

    try:
        response = _send_messages(messages)
        success_count, invalid_tokens = _collect_send_results(response, tokens)

        # Remove invalid tokens in bulk
        if invalid_tokens:
            notification_db.remove_bulk_tokens(invalid_tokens)

        logger.info(f'FCM batch send: {success_count}/{len(tokens)} successful')
        return success_count

    except Exception as e:
        logger.error(f'FCM batch send error: {e}')
        return 0


async def _send_to_user_async(
    user_id: str,
    tag: str,
    notification: Optional[messaging.Notification] = None,
    data: Optional[Dict[str, Any]] = None,
    is_background: bool = False,
    priority: str = 'normal',
    tokens: Optional[List[str]] = None,
) -> int:
    """Async boundary for the synchronous token store and Firebase Admin SDK."""
    if tokens is None:
        tokens = await run_blocking(db_executor, notification_db.get_all_tokens, user_id)
    if not tokens:
        logger.info(f"No tokens found for user {user_id}")
        return 0

    messages = [_build_message(token, tag, notification, data, is_background, priority) for token in tokens]

    try:
        response = await run_blocking(postprocess_executor, _send_messages, messages)
        success_count, invalid_tokens = _collect_send_results(response, tokens)

        if invalid_tokens:
            await run_blocking(db_executor, notification_db.remove_bulk_tokens, invalid_tokens)

        logger.info(f'FCM batch send: {success_count}/{len(tokens)} successful')
        return success_count
    except Exception as e:
        logger.error(f'FCM batch send error: {e}')
        return 0


def send_notification(
    user_id: str, title: str, body: str, data: Optional[Dict[str, Any]] = None, tokens: Optional[List[str]] = None
) -> None:
    """Send notification to all user's devices. Optionally pass pre-fetched tokens to avoid DB lookup."""
    logger.info(f'send_notification to user {user_id}')
    body = to_plain_text(body)
    tag = _generate_notification_tag(user_id, title, body, data)
    notification = messaging.Notification(title=title, body=body)
    _send_to_user(user_id, tag, notification=notification, data=data, tokens=tokens)


async def send_notification_async(
    user_id: str, title: str, body: str, data: Optional[Dict[str, Any]] = None, tokens: Optional[List[str]] = None
) -> None:
    """Async counterpart used by event-loop callers while preserving the sync public API."""
    logger.info(f'send_notification to user {user_id}')
    body = to_plain_text(body)
    tag = _generate_notification_tag(user_id, title, body, data)
    notification = messaging.Notification(title=title, body=body)
    await _send_to_user_async(user_id, tag, notification=notification, data=data, tokens=tokens)


async def send_subscription_paid_personalized_notification(user_id: str, data: Optional[Dict[str, Any]] = None) -> None:
    """Send a personalized notification to all user's devices when unlimited subscription is purchased"""
    # Get user name from Firebase Auth
    name: str = "there"
    try:
        user = await run_blocking(postprocess_executor, _get_user, user_id)
        name = user.display_name
        if not name and user.email:
            name = user.email.split('@')[0].capitalize()
        if not name:
            name = "there"
    except Exception as e:
        logger.error(f"Error getting user info from Firebase Auth: {e}")
        name = "there"

    # Generate welcome message for unlimited plan with user context
    title, body = await generate_notification_message(user_id, name, "unlimited")

    await send_notification_async(user_id, title, body, data)


async def send_credit_limit_notification(user_id: str) -> None:
    """Send a personalized credit limit notification if not sent recently"""
    # Check if notification was sent recently (within 6 hours). Offloaded: the Redis read is sync
    # and blocks the event loop in this async path.
    if await run_blocking(db_executor, has_credit_limit_notification_been_sent, user_id):
        logger.info(f"Credit limit notification already sent recently for user {user_id}")
        return

    name: str = "there"
    try:
        user = await run_blocking(postprocess_executor, _get_user, user_id)
        name = user.display_name
        if not name and user.email:
            name = user.email.split('@')[0].capitalize()
        if not name:
            name = "there"
    except Exception as e:
        logger.error(f"Error getting user info from Firebase Auth: {e}")
        name = "there"

    # Generate personalized credit limit message
    title, body = await generate_credit_limit_notification(user_id, name)

    # Send notification
    await send_notification_async(user_id, title, body)

    # Cache that notification was sent (6 hours TTL). Offloaded: the Redis write is sync and blocks
    # the event loop in this async path.
    await run_blocking(db_executor, set_credit_limit_notification_sent, user_id)
    logger.info(f"Credit limit notification sent to user {user_id}")


async def send_silent_user_notification(user_id: str) -> None:
    """Send a notification if a basic-plan user is silent for too long."""
    # Check if notification was sent recently (within 24 hours). Offloaded: the Redis read is sync
    # and blocks the event loop in this async path.
    if await run_blocking(db_executor, has_silent_user_notification_been_sent, user_id):
        logger.info(f"Silent user notification already sent recently for user {user_id}")
        return

    name: str = "there"
    try:
        user = await run_blocking(postprocess_executor, _get_user, user_id)
        name = user.display_name
        if not name and user.email:
            name = user.email.split('@')[0].capitalize()
        if not name:
            name = "there"
    except Exception as e:
        logger.error(f"Error getting user info from Firebase Auth: {e}")
        name = "there"

    # Generate personalized credit limit message
    title, body = generate_silent_user_notification(name)

    # Send notification
    await send_notification_async(user_id, title, body)

    # Cache that notification was sent (24 hours TTL). Offloaded: the Redis write is sync and blocks
    # the event loop in this async path.
    await run_blocking(db_executor, set_silent_user_notification_sent, user_id)
    logger.info(f"Silent user notification sent to user {user_id}")


def send_training_data_submitted_notification(user_id: str) -> None:
    """Send a notification when user submits their training data opt-in request."""
    # Get user name from Firebase Auth
    name: str = "there"
    try:
        user = _get_user(user_id)
        name = user.display_name
        if not name and user.email:
            name = user.email.split('@')[0].capitalize()
        if not name:
            name = "there"
    except Exception as e:
        logger.error(f"Error getting user info from Firebase Auth: {e}")
        name = "there"

    title = "omi"
    body = f"Hey {name}! Thanks for your interest in our training data program. We've received your request and our team will review it shortly. We'll notify you as soon as it's approved!"

    send_notification(user_id, title, body)
    logger.info(f"Training data submitted notification sent to user {user_id}")


async def send_bulk_notification(user_tokens: List[str], title: str, body: str) -> None:
    """Send notification to multiple users in batches."""
    try:
        batch_size = 500
        num_batches = math.ceil(len(user_tokens) / batch_size)
        body = to_plain_text(body)
        tag = _generate_tag(f"bulk:{title}:{body}")
        notification = messaging.Notification(title=title, body=body)

        def send_batch(batch_tokens: List[str]) -> Tuple[Any, List[str]]:
            messages = [_build_message(token, tag, notification=notification) for token in batch_tokens]
            response = _send_messages(messages)

            # Collect permanently invalid tokens
            invalid_tokens: List[str] = []
            for idx, result in enumerate(response.responses):
                if not result.success and result.exception:
                    error_code = getattr(result.exception, 'code', None)
                    if error_code in PERMANENT_FAILURE_CODES:
                        invalid_tokens.append(batch_tokens[idx])
                        logger.error(f"Invalid token found - Error: {error_code}")

            return response, invalid_tokens

        tasks = [
            run_blocking(postprocess_executor, send_batch, user_tokens[i * batch_size : (i + 1) * batch_size])
            for i in range(num_batches)
        ]
        results = await asyncio.gather(*tasks)

        # Remove invalid tokens
        invalid_tokens = [token for _, batch_invalid in results for token in batch_invalid]
        if invalid_tokens:
            logger.error(f"Removing {len(invalid_tokens)} invalid tokens")
            await run_blocking(db_executor, notification_db.remove_bulk_tokens, invalid_tokens)

    except Exception as e:
        logger.error(f"Error sending bulk notification: {e}")


def send_app_review_reply_notification(
    reviewer_uid: str, app_owner_uid: str, reply_body: str, app_id: str, app_name: str
):
    """Sends a notification to a user when their app review receives a reply."""
    app_owner = get_user_from_uid(app_owner_uid)
    owner_name = (app_owner or {}).get('display_name') or 'The developer'
    title = f'{owner_name} ({app_name})'
    body = reply_body
    data = {'app_id': app_id, 'type': 'app_review_reply', 'navigate_to': f'/apps/{app_id}'}
    send_notification(reviewer_uid, title, body, data)


def send_new_app_review_notification(
    app_owner_uid: str, reviewer_uid: str, app_id: str, app_name: str, review_body: str
):
    """Sends a notification to the app owner when a new review is submitted."""
    reviewer = get_user_from_uid(reviewer_uid)
    reviewer_name = (reviewer or {}).get('display_name') or 'A user'
    title = f'{reviewer_name} reviewed {app_name}'
    body = review_body
    data = {'app_id': app_id, 'type': 'new_app_review', 'navigate_to': f'/apps/{app_id}'}
    send_notification(app_owner_uid, title, body, data)


def send_action_item_data_message(user_id: str, action_item_id: str, description: str, due_at: str):
    """
    Sends a data-only FCM message for action item reminder scheduling.
    The app receives this in the background and schedules a local notification.
    """
    logger.info(f'send_action_item_data_message to user {user_id}')
    data = {
        'type': 'action_item_reminder',
        'action_item_id': action_item_id,
        'description': description,
        'due_at': due_at,
    }
    tag = _generate_tag(f"{user_id}:action_item_reminder:{action_item_id}")
    _send_to_user(user_id, tag, data=data, is_background=True, priority='high')


def _build_apple_reminders_sync_message(
    user_id: str, action_items: List[Dict[str, Any]]
) -> Optional[Tuple[str, Dict[str, str]]]:
    """Build the shared Apple Reminders payload and collapse tag."""
    if not action_items:
        return None

    items_payload: List[Dict[str, Any]] = []
    for item in action_items:
        due_at = item.get('due_at')
        due_at_str: str = ''
        if due_at:
            if hasattr(due_at, 'isoformat'):
                due_at_str = due_at.isoformat()
            else:
                due_at_str = str(due_at)

        items_payload.append(
            {
                'id': item['id'],
                'description': item['description'],
                'due_at': due_at_str,
            }
        )

    # FCM data values must be strings, so JSON-encode the items list
    # Include first item's fields at top level for backwards compatibility with old app versions
    # that expect action_item_id/description as top-level keys (single-item format).
    # Old apps will create only the first item; new apps read the full 'items' JSON array.
    first = items_payload[0]
    data = {
        'type': 'apple_reminders_sync',
        'items': json.dumps(items_payload),
        'action_item_id': first['id'],
        'description': first['description'],
        'due_at': first['due_at'],
    }

    # Use a unique tag per batch based on all item IDs to avoid collapsing different batches
    item_ids = ':'.join(item['id'] for item in action_items)
    tag = _generate_tag(f"{user_id}:apple_reminders_sync:{item_ids}")
    return tag, data


def send_apple_reminders_sync_push(user_id: str, action_items: List[Dict[str, Any]]) -> bool:
    """
    Sends a single silent push notification with a batch of action items to sync to Apple Reminders.
    This avoids iOS throttling that occurs when sending multiple rapid silent pushes.

    Args:
        user_id: The user's Firebase UID
        action_items: List of dicts, each with 'id', 'description', and optional 'due_at'

    Returns:
        bool: True if notification was sent successfully
    """
    message = _build_apple_reminders_sync_message(user_id, action_items)
    if message is None:
        return False

    logger.info(f'send_apple_reminders_sync_push to user {user_id}, {len(action_items)} items')
    tag, data = message
    success_count = _send_to_user(user_id, tag, data=data, is_background=True, priority='high')
    return success_count > 0


async def send_apple_reminders_sync_push_async(user_id: str, action_items: List[Dict[str, Any]]) -> bool:
    """Async Apple Reminders push boundary for event-loop callers."""
    message = _build_apple_reminders_sync_message(user_id, action_items)
    if message is None:
        return False

    logger.info(f'send_apple_reminders_sync_push to user {user_id}, {len(action_items)} items')
    tag, data = message
    success_count = await _send_to_user_async(user_id, tag, data=data, is_background=True, priority='high')
    return success_count > 0


def send_merge_completed_message(
    user_id: str, merged_conversation_id: str, removed_conversation_ids: List[str]
) -> None:
    """
    Sends a data-only FCM message when conversation merge completes.

    The app receives this and:
    - Foreground: Shows toast "Conversations merged successfully"
    - Background: Shows local notification

    Args:
        user_id: The user's Firebase UID
        merged_conversation_id: ID of the primary (merged) conversation
        removed_conversation_ids: List of secondary conversation IDs that were removed
    """
    logger.info(f'send_merge_completed_message to user {user_id}')
    data = {
        'type': 'merge_completed',
        'merged_conversation_id': merged_conversation_id,
        'removed_conversation_ids': ','.join(removed_conversation_ids),
    }
    tag = _generate_tag(f"{user_id}:merge_completed:{merged_conversation_id}")
    _send_to_user(user_id, tag, data=data, is_background=True, priority='high')


def send_important_conversation_message(user_id: str, conversation_id: str):
    """
    Sends a data-only FCM message when a long conversation (>30 min) completes.

    The app receives this and:
    - Shows a local notification: "You just had an important convo, click to share summary"
    - On tap: navigates to conversation detail with share sheet auto-open

    Args:
        user_id: The user's Firebase UID
        conversation_id: ID of the completed conversation
    """
    tokens = notification_db.get_all_tokens(user_id)
    if not tokens:
        logger.info(f"No notification tokens found for user {user_id} for important conversation notification")
        return

    # FCM data values must be strings
    data = {
        'type': 'important_conversation',
        'conversation_id': conversation_id,
        'navigate_to': f'/conversation/{conversation_id}?share=1',
    }

    tag = _generate_tag(f'{user_id}:important_conversation:{conversation_id}')
    _send_to_user(user_id, tag, data=data, is_background=True, priority='high')


def send_action_item_update_message(user_id: str, action_item_id: str, description: str, due_at: str):
    """
    Sends a data-only FCM message when an action item is updated.
    The app receives this and reschedules the local notification.
    """
    logger.info(f'send_action_item_update_message to user {user_id}')
    data = {
        'type': 'action_item_update',
        'action_item_id': action_item_id,
        'description': description,
        'due_at': due_at,
    }
    tag = _generate_tag(f"{user_id}:action_item_update:{action_item_id}")
    _send_to_user(user_id, tag, data=data, is_background=True, priority='high')


def send_action_item_deletion_message(user_id: str, action_item_id: str):
    """
    Sends a data-only FCM message when an action item is deleted.
    The app receives this and cancels the scheduled local notification.
    """
    logger.info(f'send_action_item_deletion_message to user {user_id}')
    data = {
        'type': 'action_item_delete',
        'action_item_id': action_item_id,
    }
    tag = _generate_tag(f"{user_id}:action_item_delete:{action_item_id}")
    _send_to_user(user_id, tag, data=data, is_background=True, priority='high')


def sync_action_item_reminder(
    user_id: str,
    action_item_id: str,
    description: str,
    completed: bool,
    due_at: Optional[Union[datetime, str]],
):
    """Reconcile the client-scheduled reminder after an action item is created or updated (#5085).

    The mobile client schedules a local reminder from the action-item update/data message and
    cancels it on the 'action_item_delete' message. The reminder must be cancelled when the task is
    completed or no longer has a due date, and (re)scheduled only for an open task that still has a
    due date. Reusing send_action_item_deletion_message is intentional: the client treats it as
    "cancel the scheduled local notification by id", not as a task deletion.
    """
    if completed or not due_at:
        send_action_item_deletion_message(user_id=user_id, action_item_id=action_item_id)
        return
    due_iso: str = due_at.isoformat() if isinstance(due_at, datetime) else due_at
    send_action_item_update_message(
        user_id=user_id, action_item_id=action_item_id, description=description or '', due_at=due_iso
    )


def send_action_items_batch_deletion_message(user_id: str, action_item_ids: List[str]):
    """
    Bulk equivalent of send_action_item_deletion_message — one FCM data
    message per chunk of ids (chunked to stay under FCM's 4KB data payload
    ceiling) instead of one message per id. The app splits the comma-joined
    ids and cancels each scheduled local notification client-side.
    """
    if not action_item_ids:
        return

    # Action item ids are UUID-shaped (~36 chars); 100 per chunk keeps the
    # serialized payload comfortably under FCM's 4KB limit.
    chunk_size = 100
    # Per-invocation nonce so concurrent bulk-delete calls that happen to
    # share a leading id don't collide on FCM tags (which would let the
    # second dispatch silently replace the first).
    nonce = uuid.uuid4().hex[:8]
    for start in range(0, len(action_item_ids), chunk_size):
        chunk = action_item_ids[start : start + chunk_size]
        data = {
            'type': 'action_item_batch_delete',
            'ids': ','.join(chunk),
        }
        tag = _generate_tag(f"{user_id}:action_item_batch_delete:{nonce}:{start}")
        _send_to_user(user_id, tag, data=data, is_background=True, priority='high')
    logger.info(f'send_action_items_batch_deletion_message to user {user_id} count={len(action_item_ids)}')


def send_action_item_created_notification(user_id: str, action_item_description: str):
    """
    Sends a notification when a new action item is created via the agentic chat.
    This provides confirmation that the task was successfully added.
    """
    # Truncate description if too long
    max_length = 60
    display_description = (
        action_item_description[:max_length] + '...'
        if len(action_item_description) > max_length
        else action_item_description
    )

    title = "Task Added"
    body = display_description

    send_notification(user_id, title, body)
    logger.info(f"Action item created notification sent to user {user_id}")


def send_action_item_completed_notification(user_id: str, action_item_description: str):
    """
    Sends a notification when a user completes an action item via the agentic chat.
    This provides positive feedback and confirmation of task completion.
    """
    # Truncate description if too long
    max_length = 60
    display_description = (
        action_item_description[:max_length] + '...'
        if len(action_item_description) > max_length
        else action_item_description
    )

    title = "Task Complete! 🎉"
    body = display_description

    send_notification(user_id, title, body)
    logger.info(f"Action item completed notification sent to user {user_id}")