app/models/stripe_event.py

"""Author RAG — Stripe Event Log Model.

Stores every received Stripe webhook event for idempotency and debugging.
The Stripe event ID is used as the primary key — if Stripe retries the
same event (which it will on any non-2xx response), the INSERT fails with
a unique constraint violation and we safely skip re-processing.

Also used for: manual replay, auditing, debugging failed webhooks.
"""

from datetime import datetime

from sqlalchemy import DateTime, String, Text
from sqlalchemy.orm import Mapped, mapped_column

from app.models.base import Base


class StripeEvent(Base):
    """Idempotent log of all received Stripe webhook events."""

    __tablename__ = "stripe_events"

    # Stripe event ID is the PK — guarantees exactly-once processing.
    # e.g. "evt_1NxAbCDEfGhIjKlMnOpQrStU"
    id: Mapped[str] = mapped_column(String(255), primary_key=True)

    event_type: Mapped[str] = mapped_column(
        String(100), nullable=False, index=True
    )  # e.g. "customer.subscription.created"

    # Stripe object ID from the event (customer, subscription, invoice, etc.)
    related_object_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
    related_object_type: Mapped[str | None] = mapped_column(String(100), nullable=True)

    # Processing metadata
    received_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    processing_error: Mapped[str | None] = mapped_column(String(1000), nullable=True)
    status: Mapped[str] = mapped_column(
        String(20), nullable=False, default="pending"
    )  # "pending" | "processed" | "failed" | "skipped"

    # Full JSON payload stored for replay / debugging
    raw_payload: Mapped[str] = mapped_column(Text, nullable=False)

    def __repr__(self) -> str:
        return f"<StripeEvent id={self.id} type={self.event_type} status={self.status}>"