leocrm/app/workflows/step_handlers.py

"""Workflow step handlers — pluggable executors for each step type.

Each handler receives the step config, the workflow instance context,
and the DB session. It returns a StepResult indicating whether to
advance, wait, branch, or abort.

G-COND, G-WAIT, G-HTTP, G-MAIL, G-CAL, G-DMS, G-SEARCH, G-AGENT, G-CRM.
"""

from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.models.workflow import WorkflowInstance

logger = logging.getLogger(__name__)


class StepResult:
    """Result of a step execution."""

    def __init__(
        self,
        *,
        advance: bool = True,
        next_index: int | None = None,
        wait_until: datetime | None = None,
        wait_reason: str | None = None,
        output: dict[str, Any] | None = None,
        error: str | None = None,
        abort: bool = False,
    ):
        self.advance = advance
        self.next_index = next_index
        self.wait_until = wait_until
        self.wait_reason = wait_reason
        self.output = output or {}
        self.error = error
        self.abort = abort


# Callable[..., Awaitable[StepResult]] — classic alias for py3.11 compat
StepHandler = Any


# ─── Registry ────────────────────────────────────────────────────────────────

_HANDLERS: dict[str, StepHandler] = {}


def register_step_type(step_type: str):
    """Decorator to register a step handler."""
    def decorator(func: StepHandler) -> StepHandler:
        _HANDLERS[step_type] = func
        return func
    return decorator


def unregister_step_type(step_type: str) -> None:
    """Remove a step handler (plugin deactivation lifecycle symmetry)."""
    _HANDLERS.pop(step_type, None)


def get_step_handler(step_type: str) -> StepHandler | None:
    return _HANDLERS.get(step_type)


def get_available_step_types() -> list[str]:
    return sorted(_HANDLERS.keys())


# ─── Built-in Step Handlers ──────────────────────────────────────────────────

@register_step_type("wait")
async def _handle_wait(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    instance: WorkflowInstance,
    step: dict[str, Any],
) -> StepResult:
    """Wait/Delay step — sets resume_at and pauses the workflow.

    Config:
        duration_seconds: int — how long to wait
        or
        resume_at: ISO datetime — absolute resume time
    """
    config = step.get("config", {})
    duration = config.get("duration_seconds")
    resume_at_str = config.get("resume_at")

    if resume_at_str:
        wait_until = datetime.fromisoformat(resume_at_str)
    elif duration:
        wait_until = datetime.now(UTC) + timedelta(seconds=int(duration))
    else:
        return StepResult(error="wait step requires duration_seconds or resume_at", abort=True)

    return StepResult(advance=False, wait_until=wait_until, wait_reason="wait")


@register_step_type("http")
async def _handle_http(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    instance: WorkflowInstance,
    step: dict[str, Any],
) -> StepResult:
    """HTTP Request step — sends an HTTP request and maps the response.

    Config:
        method: GET/POST/PUT/PATCH/DELETE
        url: str
        headers: dict
        body: str (JSON or form)
        timeout_seconds: int (default 30)
        response_mapping: dict (maps response fields to context vars)

    SSRF protection: blocks private/internal IPs, only allows http/https.
    """
    import httpx

    config = step.get("config", {})
    method = config.get("method", "GET").upper()
    url = config.get("url", "")
    headers = config.get("headers", {})
    body = config.get("body")
    timeout = config.get("timeout_seconds", 30)
    response_mapping = config.get("response_mapping", {})

    if not url:
        return StepResult(error="http step requires url", abort=True)

    # SSRF protection
    if not _is_url_safe(url):
        return StepResult(error=f"URL blocked by SSRF protection: {url}", abort=True)

    try:
        async with httpx.AsyncClient(timeout=timeout, follow_redirects=False) as client:
            resp = await client.request(method, url, headers=headers, content=body)

        output = {
            "status_code": resp.status_code,
            "response_body": resp.text[:10000],  # Limit response size
            "response_headers": dict(resp.headers),
        }

        # Map response fields to context
        for ctx_key, resp_path in response_mapping.items():
            if resp_path == "status_code":
                instance.context[ctx_key] = resp.status_code
            elif resp_path == "body":
                instance.context[ctx_key] = resp.text[:10000]

        if resp.status_code >= 400:
            return StepResult(error=f"HTTP {resp.status_code}", output=output)

        return StepResult(output=output)

    except Exception as e:
        return StepResult(error=f"HTTP request failed: {e}", abort=True)


def _is_ip_unsafe(ip: Any) -> bool:
    """Check whether an IP address is private/internal/unsafe (F15)."""
    return (
        ip.is_private
        or ip.is_loopback
        or ip.is_link_local
        or ip.is_reserved
        or ip.is_multicast
        or ip.is_unspecified
    )


def _is_url_safe(url: str) -> bool:
    """SSRF protection — block private/internal targets (F15/Astra).

    Previous behaviour only blocked IP literals and a fixed hostname list —
    internal service names (``postgres``, ``redis``, ...) and external
    domains resolving to private addresses passed unchecked.

    Now resolves the hostname via DNS and requires ALL resolved IPs to be
    public. DNS resolution failure is fail-closed (blocked).

    Note: this performs a blocking ``socket.getaddrinfo`` call — acceptable
    for workflow background steps. Full DNS-rebinding protection (pinning
    the connection to the validated IP) is a follow-up; the check directly
    before the request already narrows the window.
    """
    import ipaddress
    import socket
    import urllib.parse

    try:
        parsed = urllib.parse.urlparse(url)
    except Exception:
        return False

    if parsed.scheme not in ("http", "https"):
        return False

    hostname = parsed.hostname
    if not hostname:
        return False

    # Fast path: known internal hostnames (no DNS needed)
    blocked_hosts = {"localhost", "metadata.google.internal"}
    if hostname.lower() in blocked_hosts:
        return False

    # IP literal — validate directly
    try:
        ip = ipaddress.ip_address(hostname)
        return not _is_ip_unsafe(ip)
    except ValueError:
        pass  # Not an IP literal — resolve via DNS

    # F15: DNS resolution — ALL resolved IPs must be public. This blocks
    # internal service names (postgres, redis, ...) and external domains
    # that resolve to private/link-local addresses.
    try:
        infos = socket.getaddrinfo(
            hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM
        )
    except (socket.gaierror, OSError):
        # DNS failure → fail-closed: an unresolvable target is not verifiable
        return False

    if not infos:
        return False

    for _family, _type, _proto, _canonname, sockaddr in infos:
        try:
            ip = ipaddress.ip_address(sockaddr[0])
        except ValueError:
            return False
        if _is_ip_unsafe(ip):
            return False
    return True


@register_step_type("mail")
async def _handle_mail(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    instance: WorkflowInstance,
    step: dict[str, Any],
) -> StepResult:
    """Mail Send step — sends an email via the mail plugin.

    Config:
        to: str (recipient email)
        subject: str
        body: str
        account_id: str (optional, uses default if not set)
    """
    config = step.get("config", {})
    to = config.get("to", "")
    subject = config.get("subject", "")
    body = config.get("body", "")

    if not to or not subject:
        return StepResult(error="mail step requires to and subject", abort=True)

    try:
        from app.plugins.builtins.contracts import get_contract
        contract = get_contract("mail")
        if contract is None:
            return StepResult(error="mail plugin not available", abort=True)
        send_fn = contract.get_function("send_email")
        if send_fn is None:
            return StepResult(error="mail send_email not exposed via contract yet", abort=True)

        result = await send_fn(
            db=db,
            tenant_id=tenant_id,
            to=to,
            subject=subject,
            body=body,
            account_id=config.get("account_id"),
        )
        return StepResult(output={"mail_result": result} if result else {})
    except Exception as e:
        logger.warning("mail step failed (plugin may not be active): %s", e)
        return StepResult(error=f"mail send failed: {e}")


@register_step_type("calendar")
async def _handle_calendar(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    instance: WorkflowInstance,
    step: dict[str, Any],
) -> StepResult:
    """Calendar step — creates/updates/deletes calendar events.

    Config:
        action: create|update|delete
        title: str (for create/update)
        start: ISO datetime (for create/update)
        end: ISO datetime (for create/update)
        event_id: str (for update/delete)
    """
    config = step.get("config", {})
    action = config.get("action", "create")

    try:
        from app.plugins.builtins.contracts import get_contract
        contract = get_contract("calendar")
        if contract is None:
            return StepResult(error="calendar plugin not available", abort=True)
        if action == "create":
            fn = contract.get_function("create_event")
            if fn is None:
                return StepResult(error="calendar create_event not exposed via contract yet", abort=True)
            result = await fn(
                db=db,
                tenant_id=tenant_id,
                title=config.get("title", ""),
                start=config.get("start"),
                end=config.get("end"),
            )
            return StepResult(output={"event": result} if result else {})
        elif action == "delete":
            fn = contract.get_function("delete_event")
            if fn is None:
                return StepResult(error="calendar delete_event not exposed via contract yet", abort=True)
            await fn(db=db, tenant_id=tenant_id, event_id=config.get("event_id", ""))
            return StepResult()
        else:
            return StepResult(error=f"unknown calendar action: {action}", abort=True)
    except Exception as e:
        logger.warning("calendar step failed: %s", e)
        return StepResult(error=f"calendar action failed: {e}")


@register_step_type("dms")
async def _handle_dms(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    instance: WorkflowInstance,
    step: dict[str, Any],
) -> StepResult:
    """DMS step — interacts with the document management system.

    Config:
        action: search|download|metadata
        query: str (for search)
        file_id: str (for download/metadata)
    """
    config = step.get("config", {})
    action = config.get("action", "search")

    try:
        from app.plugins.builtins.contracts import get_contract
        contract = get_contract("dms")
        if contract is None:
            return StepResult(error="dms plugin not available", abort=True)

        if action == "search":
            fn = contract.get_function("search_files")
            if fn is None:
                return StepResult(error="dms search_files not exposed via contract yet", abort=True)
            results = await fn(db=db, tenant_id=tenant_id, query=config.get("query", ""))
            return StepResult(output={"files": results} if results else {})
        elif action == "metadata":
            fn = contract.get_function("get_file_metadata")
            if fn is None:
                return StepResult(error="dms get_file_metadata not exposed via contract yet", abort=True)
            metadata = await fn(db=db, tenant_id=tenant_id, file_id=config.get("file_id", ""))
            return StepResult(output={"metadata": metadata} if metadata else {})
        else:
            return StepResult(error=f"unknown dms action: {action}", abort=True)
    except Exception as e:
        logger.warning("dms step failed: %s", e)
        return StepResult(error=f"dms action failed: {e}")


@register_step_type("search")
async def _handle_search(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    instance: WorkflowInstance,
    step: dict[str, Any],
) -> StepResult:
    """Search step — runs a unified search query.

    Config:
        query: str
        entity_type: str (optional filter)
        limit: int (default 20)
    """
    config = step.get("config", {})
    query = config.get("query", "")
    entity_type = config.get("entity_type")
    limit = config.get("limit", 20)

    if not query:
        return StepResult(error="search step requires query", abort=True)

    try:
        from app.plugins.builtins.contracts import get_contract
        contract = get_contract("unified_search")
        if contract is None:
            return StepResult(error="search plugin not available", abort=True)
        results = await contract.simple_search(
            db=db,
            query=query,
            tenant_id=tenant_id,
            entity_types=[entity_type] if entity_type else None,
            limit=int(limit),
        )
        # Store results in context for later steps
        instance.context["search_results"] = results
        return StepResult(output={"results": results} if results else {})
    except Exception as e:
        logger.warning("search step failed: %s", e)
        return StepResult(error=f"search failed: {e}")


@register_step_type("agent")
async def _handle_agent(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    instance: WorkflowInstance,
    step: dict[str, Any],
) -> StepResult:
    """Agent step — invokes an autonomous AI agent.

    Config:
        agent_id: str
        input: dict (passed as initial messages)
        wait_for_completion: bool (default True)
    """
    config = step.get("config", {})
    agent_id = config.get("agent_id", "")
    user_input = config.get("input", {})
    wait = config.get("wait_for_completion", True)

    if not agent_id:
        return StepResult(error="agent step requires agent_id", abort=True)

    try:
        from app.plugins.builtins.contracts import get_contract
        contract = get_contract("automation")
        if contract is None:
            return StepResult(error="automation plugin not available", abort=True)
        fn = contract.get_function("run_agent")
        if fn is None:
            return StepResult(error="automation run_agent not exposed via contract yet", abort=True)
        # run_agent is an ARQ job function: it opens its own DB session and runs
        # the agent loop to completion, so ``wait`` always ends up true here.
        logger.debug("agent step wait_for_completion=%s (step always waits)", wait)
        result = await fn(
            {},
            str(agent_id),
            trigger_type="workflow",
            trigger_data={"input": user_input},
        )
        instance.context["agent_result"] = result
        return StepResult(output={"agent_result": result} if result else {})
    except Exception as e:
        logger.warning("agent step failed: %s", e)
        return StepResult(error=f"agent execution failed: {e}")


@register_step_type("crm")
async def _handle_crm(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    instance: WorkflowInstance,
    step: dict[str, Any],
) -> StepResult:
    """CRM Action step — create/update/delete contacts and companies.

    Config:
        action: create_contact|update_contact|create_company|update_company|delete_contact|delete_company
        data: dict (entity fields)
        entity_id: str (for update/delete)
    """
    config = step.get("config", {})
    action = config.get("action", "")
    data = config.get("data", {})
    entity_id = config.get("entity_id")

    if not action:
        return StepResult(error="crm step requires action", abort=True)

    try:
        if action == "create_contact":
            from app.services.contact_service import create_contact
            result = await create_contact(db, tenant_id, data)
            return StepResult(output={"contact": result} if result else {})
        elif action == "update_contact":
            from app.services.contact_service import update_contact
            if not entity_id:
                return StepResult(error="update_contact requires entity_id", abort=True)
            result = await update_contact(db, tenant_id, uuid.UUID(entity_id), data)
            return StepResult(output={"contact": result} if result else {})
        elif action == "create_company":
            from app.services.contact_service import create_contact
            company_data = {**data, "type": "company"}
            result = await create_contact(db, tenant_id, company_data)
            return StepResult(output={"company": result} if result else {})
        elif action == "update_company":
            from app.services.contact_service import update_contact
            if not entity_id:
                return StepResult(error="update_company requires entity_id", abort=True)
            result = await update_contact(db, tenant_id, uuid.UUID(entity_id), data)
            return StepResult(output={"company": result} if result else {})
        else:
            return StepResult(error=f"unknown crm action: {action}", abort=True)
    except Exception as e:
        logger.warning("crm step failed: %s", e)
        return StepResult(error=f"crm action failed: {e}")


@register_step_type("event")
async def _handle_event(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    instance: WorkflowInstance,
    step: dict[str, Any],
) -> StepResult:
    """Event step — publishes an event to the event bus.

    Config:
        event_name: str
        payload: dict
    """
    config = step.get("config", {})
    event_name = config.get("event_name", "")
    payload = config.get("payload", {})

    if not event_name:
        return StepResult(error="event step requires event_name", abort=True)

    from app.core.event_bus import get_event_bus
    event_bus = get_event_bus()
    await event_bus.publish(event_name, {
        **payload,
        "tenant_id": str(tenant_id),
        "workflow_instance_id": str(instance.id),
    })
    return StepResult(output={"event_published": event_name})


@register_step_type("webhook")
async def _handle_webhook(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    instance: WorkflowInstance,
    step: dict[str, Any],
) -> StepResult:
    """Webhook step — sends an outgoing webhook.

    Config:
        url: str
        method: str (default POST)
        headers: dict
        body: dict
        secret: str (for HMAC signing)
    """
    config = step.get("config", {})
    url = config.get("url", "")
    method = config.get("method", "POST").upper()
    headers = config.get("headers", {})
    body = config.get("body", {})

    if not url:
        return StepResult(error="webhook step requires url", abort=True)

    if not _is_url_safe(url):
        return StepResult(error=f"URL blocked by SSRF protection: {url}", abort=True)

    import json

    import httpx

    try:
        async with httpx.AsyncClient(timeout=30, follow_redirects=False) as client:
            resp = await client.request(
                method,
                url,
                headers={"Content-Type": "application/json", **headers},
                content=json.dumps(body),
            )
        return StepResult(output={
            "status_code": resp.status_code,
            "response_body": resp.text[:5000],
        })
    except Exception as e:
        return StepResult(error=f"webhook failed: {e}", abort=True)


__all__ = [
    "StepResult",
    "register_step_type",
    "get_step_handler",
    "get_available_step_types",
]