tatlock/src/core/multi_tenancy.py

"""
Multi-tenancy helpers for Tatlock.

Provides utilities for user namespace management across:
- Qdrant (collection per user for memories)
- Redis (user-scoped keys for session context)

Adapted from library-desk patterns.
"""
import re


def sanitize_user_id(user_id: str) -> str:
    """
    Sanitize user ID for use in collection names, keys, and paths.

    Converts special characters to underscores and ensures alphanumeric safety.

    Args:
        user_id: Raw user identifier (email, username, etc.)

    Returns:
        Sanitized user ID safe for use in identifiers

    Examples:
        >>> sanitize_user_id("john@example.com")
        'john_at_example_com'
        >>> sanitize_user_id("user.name")
        'user_name'
        >>> sanitize_user_id("User Name")
        'user_name'
    """
    sanitized = user_id.lower()

    # Convert @ to _at_
    sanitized = sanitized.replace("@", "_at_")

    # Convert dots to underscores
    sanitized = sanitized.replace(".", "_")

    # Replace any non-alphanumeric characters with underscores
    sanitized = re.sub(r'[^a-z0-9_]', '_', sanitized)

    # Remove consecutive underscores
    sanitized = re.sub(r'_+', '_', sanitized)

    # Remove leading/trailing underscores
    sanitized = sanitized.strip('_')

    return sanitized


def get_memory_collection_name(user_id: str) -> str:
    """
    Get Qdrant collection name for user's memories.

    Pattern: memories_{sanitized_user_id}

    Args:
        user_id: User identifier

    Returns:
        Qdrant collection name

    Examples:
        >>> get_memory_collection_name("jpmschweitzer")
        'memories_jpmschweitzer'
        >>> get_memory_collection_name("john@example.com")
        'memories_john_at_example_com'
    """
    sanitized = sanitize_user_id(user_id)
    return f"memories_{sanitized}"


def get_session_key(user_id: str, conversation_id: str) -> str:
    """
    Get Redis key for session context.

    Pattern: session:{sanitized_user}:{conversation_id}

    Args:
        user_id: User identifier
        conversation_id: Conversation identifier

    Returns:
        Redis key for session context

    Examples:
        >>> get_session_key("jpmschweitzer", "conv_abc123")
        'session:jpmschweitzer:conv_abc123'
    """
    sanitized = sanitize_user_id(user_id)
    return f"session:{sanitized}:{conversation_id}"


def get_entities_key(user_id: str, conversation_id: str) -> str:
    """
    Get Redis key for recent entities in a conversation.

    Pattern: entities:{sanitized_user}:{conversation_id}

    Args:
        user_id: User identifier
        conversation_id: Conversation identifier

    Returns:
        Redis key for recent entities

    Examples:
        >>> get_entities_key("jpmschweitzer", "conv_abc123")
        'entities:jpmschweitzer:conv_abc123'
    """
    sanitized = sanitize_user_id(user_id)
    return f"entities:{sanitized}:{conversation_id}"


def validate_user_id(user_id: str) -> bool:
    """
    Validate that a user ID is acceptable.

    Checks:
    - Not empty
    - Not too long (max 100 chars)
    - Contains some alphanumeric characters

    Args:
        user_id: User identifier to validate

    Returns:
        True if valid, False otherwise

    Examples:
        >>> validate_user_id("jpmschweitzer")
        True
        >>> validate_user_id("")
        False
        >>> validate_user_id("a" * 101)
        False
    """
    if not user_id or len(user_id) > 100:
        return False

    # Must contain at least one alphanumeric character
    if not re.search(r'[a-zA-Z0-9]', user_id):
        return False

    return True