ThinkStorm/thinkstorm/api/signal_integration.py

"""
Signal Gateway Integration API Router
Receives, authenticates, and durably processes inbound webhooks and events from Signal Gateway.
"""

import os
import json
import secrets
import base64
from pathlib import Path
from typing import Optional, Dict, Any, List, Tuple
from fastapi import APIRouter, Request, HTTPException, status, Header
from pydantic import BaseModel, Field

from ..config import config
from ..database import get_db, next_sequence, get_utc_now
from ..queue.worker import job_queue
from ..services.image_handler import (
    validate_and_sanitize_image,
    save_image_artifact,
    ImageValidationError
)

router = APIRouter(prefix="/api/integrations/signal", tags=["signal_integration"])

def extract_signal_image_candidate(attachment: Any) -> Optional[Tuple[bytes, str]]:
    """
    Extracts raw bytes and filename from a Signal attachment item.
    Returns (raw_bytes, filename) or None.
    """
    if isinstance(attachment, dict):
        # 1. Base64 data payload
        b64_data = attachment.get("data") or attachment.get("base64") or attachment.get("bytes")
        filename = attachment.get("filename") or attachment.get("name") or "signal_attachment"
        if b64_data and isinstance(b64_data, str):
            try:
                if "," in b64_data:
                    b64_data = b64_data.split(",", 1)[1]
                return base64.b64decode(b64_data), filename
            except Exception:
                pass

        # 2. Filesystem path
        file_path_str = attachment.get("path") or attachment.get("file") or attachment.get("stored_filename")
        if file_path_str and isinstance(file_path_str, str):
            p = Path(file_path_str)
            if p.exists() and p.is_file():
                try:
                    return p.read_bytes(), p.name
                except Exception:
                    pass

    elif isinstance(attachment, str):
        # Could be path or base64
        p = Path(attachment)
        if p.exists() and p.is_file():
            try:
                return p.read_bytes(), p.name
            except Exception:
                pass
        else:
            try:
                raw_str = attachment
                if "," in raw_str:
                    raw_str = raw_str.split(",", 1)[1]
                return base64.b64decode(raw_str), "signal_attachment"
            except Exception:
                pass

    elif isinstance(attachment, (bytes, bytearray)):
        return bytes(attachment), "signal_attachment"

    return None

def get_configured_callback_secret() -> str:
    """Retrieves configured callback bearer secret dynamically."""
    secret = config.services.signal_gateway_callback_secret or os.getenv("SIGNAL_GATEWAY_CALLBACK_SECRET", "")
    if secret:
        return secret
    # Try reloading .env in case it was updated after server startup
    from ..config import _load_env_file, BASE_DIR
    from pathlib import Path
    _load_env_file(Path("/root/.env"))
    _load_env_file(BASE_DIR / ".env")
    _load_env_file(BASE_DIR / "thinkstorm" / ".env")
    secret = os.getenv("SIGNAL_GATEWAY_CALLBACK_SECRET", "")
    if secret:
        config.services.signal_gateway_callback_secret = secret
        return secret
    try:
        with get_db() as conn:
            row = conn.execute(
                "SELECT api_key_raw, config_json FROM service_configurations WHERE id = 'signal_gateway'"
            ).fetchone()
            if row:
                conf = json.loads(row["config_json"] or "{}")
                if "callback_secret" in conf and conf["callback_secret"]:
                    return conf["callback_secret"]
    except Exception:
        pass
    return ""

def verify_callback_auth(request: Request):
    """
    Enforces constant-time authentication comparison on the Bearer credential.
    Returns 401 on missing or invalid authentication without leaking details.
    """
    auth_header = request.headers.get("Authorization", "")
    if not auth_header or not auth_header.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Unauthorized"
        )

    provided_token = auth_header[7:].strip()
    expected_secret = get_configured_callback_secret()

    if not expected_secret or not provided_token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Unauthorized"
        )

    # Constant-time comparison to prevent timing attacks
    if not secrets.compare_digest(provided_token, expected_secret):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Unauthorized"
        )

@router.post("/events", status_code=status.HTTP_200_OK)
async def handle_signal_events(
    request: Request,
    idempotency_key_header: Optional[str] = Header(None, alias="Idempotency-Key")
):
    """
    Inbound Signal Gateway Webhook Callback.
    Handles both test probes and actual normalized Signal messages.
    """
    # 1. Authenticate Request
    verify_callback_auth(request)

    # 2. Parse Raw JSON Body
    try:
        body = await request.json()
    except Exception:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Malformed JSON body."
        )

    if not isinstance(body, dict):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Invalid JSON structure."
        )

    # Check schema_version
    schema_version = body.get("schema_version")
    if schema_version != 1:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Unsupported schema_version: {schema_version}. Expected 1."
        )

    # 3. Handle Callback Test Request
    event_type = body.get("event_type")
    if event_type == "signal_gateway.callback_test":
        # Quick 2xx response without triggering message processing or database side effects
        return {
            "accepted": True,
            "event_type": "signal_gateway.callback_test"
        }

    # 4. Handle Actual Signal Event
    event_id = body.get("event_id")
    if not event_id or not isinstance(event_id, str):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Missing or invalid event_id."
        )

    # Validate Idempotency-Key header against JSON event_id
    if not idempotency_key_header or idempotency_key_header.strip() != event_id.strip():
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Missing or mismatched Idempotency-Key header."
        )

    channel = body.get("channel", "signal")
    if channel != "signal":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Unsupported channel: {channel}."
        )

    sender_uuid = body.get("sender_uuid")
    if not sender_uuid or not isinstance(sender_uuid, str):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Missing or invalid sender_uuid."
        )

    text = body.get("text")
    if text is None or not isinstance(text, str):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Missing or invalid text field."
        )

    raw_text = text.strip()
    if not raw_text:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Message text cannot be empty."
        )

    sender_number = body.get("sender_number")
    sender_name = body.get("sender_name")
    group_id = body.get("group_id")
    message_id = body.get("message_id") or str(int(request.state.time() if hasattr(request.state, "time") else 0))
    reply_to = body.get("reply_to")
    received_at = body.get("received_at") or get_utc_now()
    now_iso = get_utc_now()

    # 5. Durable Idempotency & Persistence
    is_duplicate = False
    idea_id = None

    with get_db() as conn:
        # Check if event already exists
        existing = conn.execute(
            "SELECT event_id, processing_status, idea_id FROM signal_inbound_events WHERE event_id = ?",
            (event_id,)
        ).fetchone()

        if existing:
            is_duplicate = True
            idea_id = existing["idea_id"]
        else:
            # Generate new Idea ID
            idea_id = next_sequence("idea")

            # Process optional Signal attachments (One-Image Rule)
            accepted_image_meta = None
            attachments = body.get("attachments") or body.get("attachment") or []
            if not isinstance(attachments, list):
                attachments = [attachments]

            for att in attachments:
                if not att:
                    continue
                candidate = extract_signal_image_candidate(att)
                if not candidate:
                    # Non-image or unreadable attachment -> ignore
                    continue
                raw_img_bytes, filename = candidate
                if not raw_img_bytes:
                    continue

                try:
                    sanitized_bytes, img_meta = validate_and_sanitize_image(
                        raw_bytes=raw_img_bytes,
                        original_filename=filename,
                        source="signal",
                        idea_id=idea_id
                    )
                    save_image_artifact(idea_id, sanitized_bytes, img_meta)
                    accepted_image_meta = img_meta
                    # Once first valid image is accepted, ignore all subsequent attachments!
                    break
                except ImageValidationError:
                    # Non-image or corrupted candidate -> ignore and check next attachment
                    continue
                except Exception:
                    continue

            img_json = json.dumps(accepted_image_meta) if accepted_image_meta else None

            # Insert idea into ThinkStorm ideas table first (so foreign key in signal_inbound_events is satisfied)
            conn.execute(
                """
                INSERT INTO ideas (
                    id, original_text, submitted_at, title, summary,
                    lifecycle_state, processing_state, enrichment_level,
                    source_channel, source_sender_uuid, source_group_id, source_event_id,
                    submission_image, created_at, updated_at
                ) VALUES (?, ?, ?, ?, ?, 'SUBMITTED', 'QUEUED', 0, 'signal', ?, ?, ?, ?, ?, ?)
                """,
                (
                    idea_id,
                    raw_text,
                    received_at,
                    "Processing New Idea (Signal)...",
                    "Analyzing inbound Signal submission text...",
                    sender_uuid,
                    group_id,
                    event_id,
                    img_json,
                    now_iso,
                    now_iso
                )
            )

            # Persist inbound event record referencing the new idea
            conn.execute(
                """
                INSERT INTO signal_inbound_events (
                    event_id, channel, sender_uuid, sender_number, sender_name,
                    group_id, message_id, reply_to, text, received_at,
                    processing_status, idea_id, created_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'QUEUED', ?, ?)
                """,
                (
                    event_id, channel, sender_uuid, sender_number, sender_name,
                    group_id, str(message_id), reply_to, raw_text, received_at,
                    idea_id, now_iso
                )
            )

    if is_duplicate:
        return {
            "accepted": True,
            "event_id": event_id,
            "idea_id": idea_id,
            "duplicate": True,
            "status": "already_processed"
        }

    # 6. Enqueue Background Intake Pipeline
    await job_queue.enqueue_foreground("intake", idea_id)

    return {
        "accepted": True,
        "event_id": event_id,
        "idea_id": idea_id,
        "status": "queued"
    }