Dolibarr MCP Server

"""Tool dispatch handlers for Dolibarr MCP Server.

This module implements the dynamic dispatcher that routes tool calls
to the appropriate client methods using the TOOL_REGISTRY.

Features:
- Dynamic tool dispatch via lookup table
- DragonflyDB caching with TTL strategies
- TOON format output (default) with JSON fallback
"""

import logging
from typing import Any, Dict, List, Optional

from .tools import TOOL_REGISTRY
from .responses import (
    success_response,
    error_response,
    paginated_response,
    list_response,
)
from ..schemas.fields import LINE_FIELDS
from ..cache.strategies import (
    should_cache,
    get_ttl_for_entity,
    get_invalidation_targets,
)
from ..formats.formatter import (
    format_response,
    OutputFormat,
    get_format_from_request,
)

logger = logging.getLogger(__name__)


def _escape_sqlfilter(value: str) -> str:
    """Escape single quotes for SQL filters to prevent injection."""
    return value.replace("'", "''")


def _filter_fields(data: Any, fields: List[str]) -> Any:
    """Filter response to include only specified fields.

    Args:
        data: Response data (dict or list of dicts)
        fields: List of field names to include

    Returns:
        Filtered data with only specified fields
    """
    if isinstance(data, list):
        return [_filter_fields(item, fields) for item in data]
    if isinstance(data, dict):
        result = {k: v for k, v in data.items() if k in fields}
        # Handle nested lines with LINE_FIELDS
        if "lines" in data and "lines" in fields:
            result["lines"] = [
                _filter_fields(line, LINE_FIELDS)
                for line in data.get("lines", [])
            ]
        return result
    return data


async def _handle_search(
    client: Any,
    tool_def: Dict[str, Any],
    args: Dict[str, Any]
) -> Dict[str, Any]:
    """Handle search tool operations with SQL filter building.

    Args:
        client: DolibarrClient instance
        tool_def: Tool definition from registry
        args: Tool arguments

    Returns:
        Formatted response with search results
    """
    handler_type = tool_def.get("search_handler")
    method_name = tool_def["method"]
    method = getattr(client, method_name)
    limit = args.get("limit", 20)
    fields = tool_def.get("fields")

    if handler_type == "ref_prefix":
        # Search products by reference prefix
        ref = _escape_sqlfilter(args["ref_prefix"])
        sqlfilter = f"(t.ref:like:'{ref}%')"
        result = await method(sqlfilter, limit)

    elif handler_type == "label":
        # Search products by label
        label = _escape_sqlfilter(args["query"])
        sqlfilter = f"(t.label:like:'%{label}%')"
        result = await method(sqlfilter, limit)

    elif handler_type == "customer":
        # Search customers by name or alias
        q = _escape_sqlfilter(args["query"])
        sqlfilter = f"((t.nom:like:'%{q}%') OR (t.name_alias:like:'%{q}%'))"
        result = await method(sqlfilter, limit)

    elif handler_type == "project":
        # Search projects by ref or title
        q = _escape_sqlfilter(args["query"])
        sqlfilter = f"((t.ref:like:'%{q}%') OR (t.title:like:'%{q}%'))"
        result = await method(sqlfilter, limit)

    elif handler_type == "proposal":
        # Search proposals by ref only (customer name search requires JOIN not supported by API)
        q = _escape_sqlfilter(args["query"])
        sqlfilter = f"(t.ref:like:'%{q}%')"
        result = await method(sqlfilter, limit)

    elif handler_type == "resolve_ref":
        # Resolve exact product reference
        ref = args["ref"]
        products = await method(f"(t.ref:like:'{_escape_sqlfilter(ref)}')", 2)

        if not products:
            return success_response(
                {"status": "not_found", "ref": ref},
                {"action": "resolve_ref"}
            )

        if len(products) == 1:
            filtered = _filter_fields(products[0], fields) if fields else products[0]
            return success_response(
                {"status": "ok", "product": filtered},
                {"action": "resolve_ref"}
            )

        # Check for exact match
        exact = [p for p in products if p.get("ref") == ref]
        if len(exact) == 1:
            filtered = _filter_fields(exact[0], fields) if fields else exact[0]
            return success_response(
                {"status": "ok", "product": filtered},
                {"action": "resolve_ref"}
            )

        # Ambiguous - multiple matches
        filtered = _filter_fields(products, fields) if fields else products
        return success_response(
            {"status": "ambiguous", "products": filtered},
            {"action": "resolve_ref"}
        )

    else:
        raise ValueError(f"Unknown search handler: {handler_type}")

    # Filter and return results
    if fields:
        result = _filter_fields(result, fields)

    return list_response(result, limit)


async def dispatch_tool(
    client: Any,
    name: str,
    args: Dict[str, Any]
) -> Dict[str, Any]:
    """Dispatch tool call to appropriate client method.

    This replaces the original 59 if-statements with a single
    lookup table approach.

    Args:
        client: DolibarrClient instance
        name: Tool name
        args: Tool arguments

    Returns:
        Formatted response (success or error)
    """
    # Get tool definition from registry
    tool_def = TOOL_REGISTRY.get(name)
    if not tool_def:
        return error_response(
            "UNKNOWN_TOOL",
            f"Tool '{name}' not found",
            status=404,
            retriable=False,
            details={"tool_name": name}
        )

    # Handle search tools specially
    if tool_def.get("search_handler"):
        return await _handle_search(client, tool_def, args)

    # Get the client method
    method_name = tool_def["method"]
    method = getattr(client, method_name, None)
    if not method:
        return error_response(
            "TOOL_EXECUTION_ERROR",
            f"Client method '{method_name}' not found",
            status=500,
            details={"method": method_name}
        )

    # Extract ID parameters if specified
    args_copy = args.copy()
    id_param = tool_def.get("id_param")
    line_param = tool_def.get("line_param")

    positional_args = []
    if id_param and id_param in args_copy:
        positional_args.append(args_copy.pop(id_param))
    if line_param and line_param in args_copy:
        positional_args.append(args_copy.pop(line_param))

    # Call the method
    if positional_args:
        result = await method(*positional_args, **args_copy)
    else:
        result = await method(**args_copy)

    # Apply field filtering if specified
    fields = tool_def.get("fields")
    if fields and result:
        result = _filter_fields(result, fields)

    # Wrap response with pagination if needed
    if tool_def.get("paginated") and isinstance(result, list):
        limit = args.get("limit", 100)
        return list_response(result, limit)

    return success_response(result)


async def dispatch_tool_cached(
    client: Any,
    name: str,
    args: Dict[str, Any],
    cache: Optional[Any] = None,
) -> Dict[str, Any]:
    """Dispatch tool call with caching support.

    Checks cache before executing, stores results on success.
    Invalidates related caches for mutation operations.

    Args:
        client: DolibarrClient instance
        name: Tool name
        args: Tool arguments
        cache: Optional DragonflyCache instance

    Returns:
        Formatted response (success or error)
    """
    # Check if we should try cache
    if cache and should_cache(name):
        cache_key = cache.make_tool_key(name, args)
        cached = await cache.get(cache_key)
        if cached is not None:
            logger.debug(f"Cache HIT for {name}")
            # Add cache metadata to response
            if isinstance(cached, dict) and "metadata" in cached:
                cached["metadata"]["cached"] = True
            return cached

    # Execute the tool
    response = await dispatch_tool(client, name, args)

    # Cache successful read results
    if cache and response.get("success") and should_cache(name):
        ttl = get_ttl_for_entity(name)
        await cache.set(cache_key, response, ttl)
        logger.debug(f"Cache SET for {name} (TTL: {ttl}s)")

    # Invalidate related caches for mutations
    if cache and response.get("success"):
        targets = get_invalidation_targets(name)
        if targets:
            for target in targets:
                await cache.invalidate_pattern(f"tool:{target}:*")
            logger.debug(f"Cache INVALIDATE for {name}: {targets}")

    return response


async def dispatch_tool_formatted(
    client: Any,
    name: str,
    args: Dict[str, Any],
    cache: Optional[Any] = None,
    output_format: Optional[OutputFormat] = None,
) -> str:
    """Dispatch tool call and return formatted string output.

    Uses TOON format by default for optimal token efficiency.

    Args:
        client: DolibarrClient instance
        name: Tool name
        args: Tool arguments
        cache: Optional DragonflyCache instance
        output_format: Output format (default: TOON)

    Returns:
        Formatted string (TOON or JSON)
    """
    # Extract format from args if not specified
    if output_format is None:
        output_format = get_format_from_request(args)

    # Remove format args before dispatch
    args_clean = {k: v for k, v in args.items()
                  if k not in ('format', 'output_format')}

    # Get response (cached or fresh)
    if cache:
        response = await dispatch_tool_cached(client, name, args_clean, cache)
    else:
        response = await dispatch_tool(client, name, args_clean)

    # Format the response
    return format_response(response, output_format)


async def dispatch_tool_legacy(
    client: Any,
    name: str,
    args: Dict[str, Any]
) -> Any:
    """Dispatch tool call and return raw result (legacy format).

    This provides backward compatibility with the original response format
    that returns raw data without the success/data/metadata wrapper.

    Args:
        client: DolibarrClient instance
        name: Tool name
        args: Tool arguments

    Returns:
        Raw result data (not wrapped in success response)
    """
    response = await dispatch_tool(client, name, args)

    # For backward compatibility, return raw data on success
    if response.get("success"):
        return response.get("data")

    # For errors, return the error structure
    return response.get("error", response)