client.py

"""
Calanque Python Client

Supports both gRPC and HTTP (CloudEvents batch) protocols.
"""

import asyncio
import json
from dataclasses import dataclass
from datetime import datetime
from typing import AsyncIterator, Optional
from contextlib import asynccontextmanager

import httpx


@dataclass
class CloudEvent:
    """CloudEvents 1.0 representation."""
    id: str
    source: str
    type: str
    specversion: str = "1.0"
    time: Optional[datetime] = None
    datacontenttype: Optional[str] = None
    subject: Optional[str] = None
    data: Optional[dict] = None
    extensions: Optional[dict] = None
    
    # Calanque metadata
    message_id: str = ""
    partition: int = 0
    offset: int = 0


class CalanqueHTTPClient:
    """
    HTTP client for Calanque Pull API.

    Uses CloudEvents batch format (application/cloudevents-batch+json).
    """
    
    def __init__(
        self,
        base_url: str,
        subscription: str,
        timeout: float = 30.0,
    ):
        self.base_url = base_url.rstrip("/")
        self.subscription = subscription
        self.timeout = timeout
        self._client: Optional[httpx.AsyncClient] = None
        self._pending_ids: list[str] = []
    
    async def __aenter__(self):
        self._client = httpx.AsyncClient(timeout=self.timeout)
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._client:
            await self._client.aclose()
    
    async def pull(
        self,
        max_events: int = 10,
        timeout: str = "5s",
    ) -> tuple[list[CloudEvent], str]:
        """
        Pull a batch of events.
        
        Returns:
            Tuple of (events, batch_id)
        """
        response = await self._client.get(
            f"{self.base_url}/events",
            params={
                "subscription": self.subscription,
                "maxEvents": max_events,
                "timeout": timeout,
            },
        )
        response.raise_for_status()
        
        batch_id = response.headers.get("X-Batch-Id", "")
        events_data = response.json()
        
        events = []
        self._pending_ids = []
        
        for e in events_data:
            event = CloudEvent(
                id=e["id"],
                source=e["source"],
                type=e["type"],
                specversion=e.get("specversion", "1.0"),
                time=datetime.fromisoformat(e["time"]) if e.get("time") else None,
                datacontenttype=e.get("datacontenttype"),
                subject=e.get("subject"),
                data=e.get("data"),
                extensions=e.get("extensions"),
                message_id=e.get("messageid", ""),
                partition=e.get("partition", 0),
                offset=e.get("offset", 0),
            )
            events.append(event)
            self._pending_ids.append(event.message_id)
        
        return events, batch_id
    
    async def ack(
        self,
        message_ids: Optional[list[str]] = None,
        batch_id: Optional[str] = None,
    ) -> bool:
        """Acknowledge processed messages."""
        ids = message_ids or self._pending_ids
        
        response = await self._client.post(
            f"{self.base_url}/ack",
            params={"subscription": self.subscription},
            json={"messageIds": ids, "batchId": batch_id},
        )
        response.raise_for_status()
        
        self._pending_ids = [
            mid for mid in self._pending_ids if mid not in ids
        ]
        return True
    
    async def nack(
        self,
        message_ids: Optional[list[str]] = None,
        requeue: bool = True,
    ) -> bool:
        """
        Negative acknowledge messages.
        
        Args:
            message_ids: IDs to nack (default: all pending)
            requeue: If True, message will be retried. If False, sent to DLQ.
        """
        ids = message_ids or self._pending_ids
        
        response = await self._client.post(
            f"{self.base_url}/nack",
            params={"subscription": self.subscription},
            json={"messageIds": ids, "requeue": requeue},
        )
        response.raise_for_status()
        
        self._pending_ids = [
            mid for mid in self._pending_ids if mid not in ids
        ]
        return True
    
    async def ack_one(self, message_id: str) -> bool:
        """Acknowledge a single message."""
        return await self.ack(message_ids=[message_id])
    
    async def nack_one(self, message_id: str, requeue: bool = True) -> bool:
        """Negative acknowledge a single message."""
        return await self.nack(message_ids=[message_id], requeue=requeue)


class ProcessingError(Exception):
    """Exception for message processing errors."""
    
    def __init__(self, message: str, retryable: bool = True):
        super().__init__(message)
        self.retryable = retryable


@asynccontextmanager
async def calanque_client(
    base_url: str,
    subscription: str,
    timeout: float = 30.0,
) -> AsyncIterator[CalanqueHTTPClient]:
    """Context manager for Calanque HTTP client."""
    async with CalanqueHTTPClient(base_url, subscription, timeout) as client:
        yield client


# Example usage
async def main():
    """Example: Order processing with retry handling."""

    async with calanque_client(
        "http://order-processor-consumer:8080",
        "order-processor",
    ) as client:
        
        while True:
            # Pull batch of events
            events, batch_id = await client.pull(max_events=10)
            
            if not events:
                await asyncio.sleep(1)
                continue
            
            print(f"Received {len(events)} events (batch: {batch_id})")
            
            # Process each event individually
            for event in events:
                try:
                    await process_order(event)
                    # Ack successful processing
                    await client.ack_one(event.message_id)
                    print(f"✓ Processed: {event.type} - {event.id}")
                    
                except ProcessingError as e:
                    if e.retryable:
                        # Requeue for retry
                        await client.nack_one(event.message_id, requeue=True)
                        print(f"↻ Retrying: {event.type} - {event.id}")
                    else:
                        # Send to DLQ
                        await client.nack_one(event.message_id, requeue=False)
                        print(f"✗ DLQ: {event.type} - {event.id}")
                        
                except Exception as e:
                    # Unknown error - retry
                    await client.nack_one(event.message_id, requeue=True)
                    print(f"↻ Error, retrying: {event.type} - {e}")


async def process_order(event: CloudEvent):
    """Example order processor."""
    
    if event.type == "order.created":
        order = event.data
        
        # Validate order
        if not order.get("items"):
            raise ProcessingError("Order has no items", retryable=False)
        
        # Check inventory (might fail temporarily)
        if order.get("_test_fail"):
            raise ProcessingError("Inventory check failed", retryable=True)
        
        print(f"  Processing order: {order.get('id')}")
        
    elif event.type == "order.updated":
        print(f"  Order updated: {event.data.get('id')}")
        
    elif event.type == "order.cancelled":
        print(f"  Order cancelled: {event.data.get('id')}")


if __name__ == "__main__":
    asyncio.run(main())