# services/user-service/app/main.py
from contextlib import asynccontextmanager
from datetime import datetime
from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, Response, status
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
from pydantic import BaseModel, EmailStr, Field
from pydantic_settings import BaseSettings

from app.observability.tracing import init_tracing
from app.observability.metrics import metrics_middleware
from app.observability.logger import logger


class Settings(BaseSettings):
    mongodb_uri: str
    port: int = 3000
    otel_exporter_otlp_endpoint: Optional[str] = None


settings = Settings()

# Initialize tracing before other imports
init_tracing("user-service")


class Address(BaseModel):
    street: Optional[str] = None
    city: Optional[str] = None
    country: Optional[str] = None
    postal_code: Optional[str] = None
    is_default: bool = False


class Profile(BaseModel):
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    phone: Optional[str] = None


class CreateUserRequest(BaseModel):
    email: EmailStr
    password_hash: str
    profile: Optional[Profile] = None
    addresses: list[Address] = Field(default_factory=list)


class UserResponse(BaseModel):
    id: str
    email: EmailStr


@asynccontextmanager
async def lifespan(app: FastAPI):
    client = AsyncIOMotorClient(settings.mongodb_uri)
    app.state.db = client.get_default_database()
    await app.state.db.users.create_index("email", unique=True)
    logger.info("User service starting", extra={"port": settings.port})
    yield
    client.close()


app = FastAPI(lifespan=lifespan)
app.middleware("http")(metrics_middleware(service_name="user-service"))


def get_db() -> AsyncIOMotorDatabase:
    return app.state.db


@app.post("/users", status_code=status.HTTP_201_CREATED, response_model=UserResponse)
async def create_user(payload: CreateUserRequest, db: AsyncIOMotorDatabase = Depends(get_db)):
    try:
        doc = payload.model_dump()
        doc["created_at"] = datetime.utcnow()
        result = await db.users.insert_one(doc)
        logger.info("User created", extra={"user_id": str(result.inserted_id)})
        return UserResponse(id=str(result.inserted_id), email=payload.email)
    except Exception as exc:
        logger.error("Failed to create user", extra={"error": str(exc)})
        raise HTTPException(status_code=400, detail=str(exc))


@app.get("/users/{user_id}")
async def get_user(user_id: str, db: AsyncIOMotorDatabase = Depends(get_db)):
    from bson import ObjectId

    user = await db.users.find_one(
        {"_id": ObjectId(user_id)}, {"password_hash": 0}
    )
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    user["id"] = str(user.pop("_id"))
    return user


# Health checks
@app.get("/health/live")
async def liveness():
    return {"status": "UP"}


@app.get("/health/ready")
async def readiness(db: AsyncIOMotorDatabase = Depends(get_db)):
    try:
        await db.command("ping")
        return {"status": "READY", "database": True}
    except Exception:
        return Response(
            content='{"status":"NOT_READY","database":false}',
            status_code=503,
            media_type="application/json",
        )


@app.get("/metrics")
async def get_metrics():
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)

# services/order-service/app/sagas/order_saga.py
import json
from typing import Any

from aiokafka import AIOKafkaConsumer, AIOKafkaProducer

from app.repositories.order_repository import OrderRepository


class OrderSaga:
    def __init__(
        self,
        order_repository: OrderRepository,
        producer: AIOKafkaProducer,
        consumer: AIOKafkaConsumer,
    ) -> None:
        self.order_repository = order_repository
        self.producer = producer
        self.consumer = consumer

    async def start(self) -> None:
        await self.producer.start()
        await self.consumer.start()
        # Consumer is pre-configured to subscribe to:
        # payment.completed, payment.failed, inventory.reserved, inventory.failed
        try:
            async for message in self.consumer:
                event = json.loads(message.value.decode("utf-8"))
                await self.handle_event(message.topic, event)
        finally:
            await self.consumer.stop()
            await self.producer.stop()

    async def create_order(self, order_data: dict[str, Any]) -> dict[str, Any]:
        # Step 1: Create order in PENDING state
        order = await self.order_repository.create(
            {**order_data, "status": "PENDING", "saga_state": "STARTED"}
        )

        # Step 2: Request inventory reservation
        await self.producer.send_and_wait(
            "inventory.reserve",
            key=order["id"].encode("utf-8"),
            value=json.dumps(
                {"order_id": order["id"], "items": order["items"]}
            ).encode("utf-8"),
        )
        return order

    async def handle_event(self, topic: str, event: dict[str, Any]) -> None:
        order = await self.order_repository.find_by_id(event["order_id"])
        if not order:
            # Order may have been deleted or this is a stale event -- safe to ignore
            return

        if topic == "inventory.reserved":
            # Step 3: Request payment
            await self.order_repository.update(
                order["id"], {"saga_state": "INVENTORY_RESERVED"}
            )
            await self.producer.send_and_wait(
                "payment.process",
                key=order["id"].encode("utf-8"),
                value=json.dumps(
                    {
                        "order_id": order["id"],
                        "customer_id": order["customer_id"],
                        "amount": order["total"],
                    }
                ).encode("utf-8"),
            )

        elif topic == "payment.completed":
            # Step 4: Complete order
            await self.order_repository.update(
                order["id"],
                {
                    "status": "CONFIRMED",
                    "saga_state": "COMPLETED",
                    "payment_id": event["payment_id"],
                },
            )
            await self.producer.send_and_wait(
                "order.confirmed",
                key=order["id"].encode("utf-8"),
                value=json.dumps({"order_id": order["id"]}).encode("utf-8"),
            )

        elif topic == "payment.failed":
            # Compensate: release inventory -- this is the saga's compensation logic.
            # If payment fails, we must undo the inventory reservation or those items
            # remain locked forever (a common production bug).
            await self.order_repository.update(
                order["id"],
                {"status": "FAILED", "saga_state": "COMPENSATING"},
            )
            await self.producer.send_and_wait(
                "inventory.release",
                key=order["id"].encode("utf-8"),
                value=json.dumps(
                    {"order_id": order["id"], "items": order["items"]}
                ).encode("utf-8"),
            )

        elif topic == "inventory.failed":
            # No compensation needed, just fail the order
            await self.order_repository.update(
                order["id"],
                {
                    "status": "FAILED",
                    "saga_state": "FAILED",
                    "failure_reason": "Insufficient inventory",
                },
            )

# services/payment-service/app/main.py
import asyncio
import json
from contextlib import asynccontextmanager
from typing import Any, Optional

import stripe
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from fastapi import FastAPI
from pydantic import BaseModel
from pydantic_settings import BaseSettings
from redis.asyncio import Redis

from app.observability.tracing import init_tracing
from app.observability.logger import logger


class Settings(BaseSettings):
    stripe_secret_key: str
    kafka_brokers: str
    redis_url: str = "redis://redis:6379"


settings = Settings()
init_tracing("payment-service")
stripe.api_key = settings.stripe_secret_key


class PaymentEvent(BaseModel):
    order_id: str
    customer_id: str
    amount: float


class PaymentResult(BaseModel):
    payment_id: str
    order_id: str
    status: str


async def get_customer_payment_method(customer_id: str) -> dict[str, Any]:
    # Fetch from User Service or local cache; stubbed for brevity
    ...


async def process_payment(
    event: dict[str, Any],
    producer: AIOKafkaProducer,
    idempotency_store: Redis,
) -> Optional[PaymentResult]:
    payload = PaymentEvent(**event)

    # Check idempotency -- Redis-backed, so restarts do not lose state.
    # An in-memory dict would let duplicate charges slip through on deploy.
    cached = await idempotency_store.get(f"payment:{payload.order_id}")
    if cached:
        logger.info("Payment already processed", extra={"order_id": payload.order_id})
        return PaymentResult(**json.loads(cached))

    try:
        customer = await get_customer_payment_method(payload.customer_id)

        # Create payment intent (sync call to Stripe; unavoidable)
        payment_intent = stripe.PaymentIntent.create(
            amount=round(payload.amount * 100),
            currency="usd",
            customer=customer["stripe_customer_id"],
            payment_method=customer["default_payment_method"],
            confirm=True,
            metadata={"order_id": payload.order_id},
            idempotency_key=f"order-{payload.order_id}",
        )

        result = PaymentResult(
            payment_id=payment_intent.id,
            order_id=payload.order_id,
            status="COMPLETED",
        )

        # Persist for idempotency (24h TTL)
        await idempotency_store.setex(
            f"payment:{payload.order_id}",
            86400,
            result.model_dump_json(),
        )

        await producer.send_and_wait(
            "payment.completed",
            key=payload.order_id.encode("utf-8"),
            value=result.model_dump_json().encode("utf-8"),
        )
        return result

    except Exception as exc:
        await producer.send_and_wait(
            "payment.failed",
            key=payload.order_id.encode("utf-8"),
            value=json.dumps(
                {"order_id": payload.order_id, "error": str(exc)}
            ).encode("utf-8"),
        )
        raise


@asynccontextmanager
async def lifespan(app: FastAPI):
    producer = AIOKafkaProducer(bootstrap_servers=settings.kafka_brokers)
    consumer = AIOKafkaConsumer(
        "payment.process",
        bootstrap_servers=settings.kafka_brokers,
        group_id="payment-service",
    )
    idempotency_store = Redis.from_url(settings.redis_url)

    await producer.start()
    await consumer.start()
    app.state.producer = producer
    app.state.idempotency_store = idempotency_store

    async def consume_loop():
        async for message in consumer:
            event = json.loads(message.value.decode("utf-8"))
            await process_payment(event, producer, idempotency_store)

    task = asyncio.create_task(consume_loop())
    try:
        yield
    finally:
        task.cancel()
        await consumer.stop()
        await producer.stop()
        await idempotency_store.aclose()


app = FastAPI(lifespan=lifespan)


@app.get("/health/live")
async def liveness():
    return {"status": "UP"}

# tests/contract/test_order_payment_consumer.py
from pathlib import Path

import pytest
from pact import Consumer, Like, Provider

from app.clients.payment_client import PaymentClient


PACT_DIR = Path.cwd() / "pacts"

pact = Consumer("OrderService").has_pact_with(
    Provider("PaymentService"),
    host_name="localhost",
    port=8081,
    pact_dir=str(PACT_DIR),
)


@pytest.fixture(scope="module", autouse=True)
def pact_lifecycle():
    pact.start_service()
    yield
    pact.stop_service()


@pytest.mark.asyncio
async def test_process_payment_successfully():
    (
        pact
        .given("customer has valid payment method")
        .upon_receiving("payment request")
        .with_request(
            method="POST",
            path="/payments",
            body={
                "order_id": "12345",
                "amount": 99.99,
                "customer_id": "cust-123",
            },
        )
        .will_respond_with(
            status=200,
            body={
                "payment_id": Like("pay_abc123"),
                "status": "COMPLETED",
            },
        )
    )

    with pact:
        client = PaymentClient(base_url="http://localhost:8081")
        result = await client.process(
            order_id="12345",
            amount=99.99,
            customer_id="cust-123",
        )

    assert result["status"] == "COMPLETED"