tracer-cloud--opensre/tools/slack_send_message_tool/delivery.py

"""Credential resolution and transport dispatch for Slack messages."""

from __future__ import annotations

import os

from integrations.slack.delivery import send_slack_webhook_message
from tools.slack_send_message_tool.models import SlackDeliveryTarget


def resolve_webhook_url(webhook_url: str = "") -> tuple[SlackDeliveryTarget | None, str]:
    """Resolve a Slack incoming webhook URL from the caller or integration store."""
    explicit = str(webhook_url or "").strip()
    if explicit:
        return SlackDeliveryTarget(webhook_url=explicit), ""

    env_url = os.getenv("SLACK_WEBHOOK_URL", "").strip()
    if env_url:
        return SlackDeliveryTarget(webhook_url=env_url), ""

    try:
        from integrations.catalog import resolve_effective_integrations

        slack_integration = resolve_effective_integrations().get("slack") or {}
        config = slack_integration.get("config") if isinstance(slack_integration, dict) else {}
        stored_url = str(config.get("webhook_url", "") if isinstance(config, dict) else "").strip()
    except Exception as exc:
        return None, str(exc)

    if not stored_url:
        return None, (
            "Slack is not configured. Set SLACK_WEBHOOK_URL or configure the Slack integration."
        )
    return SlackDeliveryTarget(webhook_url=stored_url), ""


def dispatch_message(message: str, target: SlackDeliveryTarget) -> tuple[bool, str]:
    """Post a message via the resolved Slack incoming webhook."""
    ok, error = send_slack_webhook_message(message, webhook_url=target.webhook_url)
    if ok:
        return True, ""
    if error == "no_webhook":
        return False, "Slack webhook URL is not configured."
    return False, "Slack webhook delivery failed."