from __future__ import annotations import uuid from datetime import datetime, timezone from enum import StrEnum from flask_login import UserMixin from sqlalchemy import ( Boolean, DateTime, Enum, ForeignKey, Integer, String, Text, UniqueConstraint, Uuid, ) from sqlalchemy.orm import Mapped, mapped_column, relationship from werkzeug.security import check_password_hash, generate_password_hash from . import Base class SentWebmentionStatus(StrEnum): SENT = "sent" UNSUPPORTED = "unsupported" FAILED = "failed" class User(UserMixin, Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) username: Mapped[str] = mapped_column(Text(), index=True, unique=True) password_hash: Mapped[str] = mapped_column(Text(), nullable=False) def __repr__(self) -> str: return f"" def set_password(self, password: str) -> None: self.password_hash = generate_password_hash(password) def check_password(self, password: str) -> bool: return check_password_hash(self.password_hash, password) class ReceivedWebmention(Base): __tablename__ = "received_webmentions" uuid: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True) source: Mapped[str] = mapped_column(Text(), nullable=False) target: Mapped[str] = mapped_column(Text(), nullable=False) status: Mapped[str] = mapped_column(Text(), nullable=False, default="received") failure_reason: Mapped[str | None] = mapped_column( Text(), nullable=True, default=None ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc), nullable=False, ) __table_args__ = ( UniqueConstraint("source", "target", name="uq_webmention_source_target"), ) @property def verified(self) -> bool: return self.status == "verified" @property def created_at(self) -> datetime: return uuid7_to_datetime(self.uuid) class Source(Base): __tablename__ = "sources" uuid: Mapped[uuid.UUID] = mapped_column( Uuid(as_uuid=True), primary_key=True, default=uuid.uuid7 ) path: Mapped[str] = mapped_column(Text(), nullable=False, unique=True) url: Mapped[str] = mapped_column(Text(), nullable=False, unique=True) content_hash: Mapped[str] = mapped_column(String(32), nullable=False) revision: Mapped[int] = mapped_column(Integer(), nullable=False, default=1) last_seen_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False ) revised_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False ) deleted_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, default=None ) sent_webmentions: Mapped[list[SentWebmention]] = relationship( back_populates="source", cascade="all, delete-orphan", lazy="selectin" ) @property def created_at(self) -> datetime: return uuid7_to_datetime(self.uuid) class SentWebmention(Base): __tablename__ = "sent_webmentions" uuid: Mapped[uuid.UUID] = mapped_column( Uuid(as_uuid=True), primary_key=True, default=uuid.uuid7 ) source_id: Mapped[uuid.UUID] = mapped_column( Uuid(as_uuid=True), ForeignKey("sources.uuid", ondelete="CASCADE"), nullable=False, index=True, ) target: Mapped[str] = mapped_column(Text(), nullable=False) active: Mapped[bool] = mapped_column(Boolean(), nullable=False, default=True) desired_revision: Mapped[int] = mapped_column(Integer(), nullable=False) processed_revision: Mapped[int | None] = mapped_column( Integer(), nullable=True, default=None ) sent_revision: Mapped[int | None] = mapped_column( Integer(), nullable=True, default=None ) status: Mapped[SentWebmentionStatus | None] = mapped_column( Enum( SentWebmentionStatus, name="sent_webmention_status", values_callable=lambda enum_type: [member.value for member in enum_type], native_enum=False, ), nullable=True, default=None, ) failure_reason: Mapped[str | None] = mapped_column( Text(), nullable=True, default=None ) endpoint: Mapped[str | None] = mapped_column(Text(), nullable=True, default=None) response_status: Mapped[int | None] = mapped_column( Integer(), nullable=True, default=None ) status_url: Mapped[str | None] = mapped_column(Text(), nullable=True, default=None) last_attempted_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, default=None ) last_sent_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, default=None ) source: Mapped[Source] = relationship(back_populates="sent_webmentions") __table_args__ = ( UniqueConstraint( "source_id", "target", name="uq_sent_webmention_source_target" ), ) @property def pending(self) -> bool: return ( self.processed_revision is None or self.processed_revision < self.desired_revision ) @property def created_at(self) -> datetime: return uuid7_to_datetime(self.uuid) def uuid7_to_datetime(identifier: uuid.UUID) -> datetime: return datetime.fromtimestamp(identifier.time / 1000, tz=timezone.utc)