# SPDX-FileCopyrightText: 2026 Dennis Fink # # SPDX-License-Identifier: BSD-3-Clause from __future__ import annotations import uuid from datetime import UTC, datetime from enum import StrEnum from typing import Never 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): """ Represent the outcome of a Webmention send attempt. """ SENT = "sent" UNSUPPORTED = "unsupported" FAILED = "failed" class User(UserMixin, Base): """ Represent an authenticated application user. """ __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 a developer-readable representation of the user. :return: Representation containing the username. """ return f"" @property def password(self) -> Never: """ Prevent access to the user's plain-text password. :raises AttributeError: Always, because the password is write-only. """ raise AttributeError("Password is write-only") @password.setter def password(self, password: str) -> None: """ Hash and store a new password. :param password: Plain-text password to hash. """ self.password_hash = generate_password_hash(password) def check_password(self, password: str) -> bool: """ Check a plain-text password against the stored password hash. :param password: Plain-text password to check. :return: Whether the password matches the stored hash. """ return check_password_hash(self.password_hash, password) class ReceivedWebmention(Base): """ Represent a Webmention received for verification. """ __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(UTC), onupdate=lambda: datetime.now(UTC), nullable=False, ) __table_args__ = ( UniqueConstraint("source", "target", name="uq_webmention_source_target"), ) def __repr__(self) -> str: """ Return a developer-readable representation of the received Webmention. :return: Representation containing the identifier, source, and target. """ return f" {self.target!r}>" @property def verified(self) -> bool: """ Return whether the Webmention has been successfully verified. :return: Whether the Webmention status is ``verified``. """ return self.status == "verified" @property def created_at(self) -> datetime: """ Return the creation time encoded in the Webmention UUID. :return: Creation time as a timezone-aware datetime. """ return uuid7_to_datetime(self.uuid) class Source(Base): """ Represent a generated source document containing outgoing Webmentions. """ __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" ) def __repr__(self) -> str: """ Return a developer-readable representation of the source. :return: Representation containing the identifier and public URL. """ return f"" @property def created_at(self) -> datetime: """ Return the creation time encoded in the source UUID. :return: Creation time as a timezone-aware datetime. """ return uuid7_to_datetime(self.uuid) class SentWebmention(Base): """ Represent an outgoing Webmention associated with a source. """ __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" ), ) def __repr__(self) -> str: """ Return a developer-readable representation of the sent Webmention. :return: Representation containing the identifier and target URL. """ return f"" @property def pending(self) -> bool: """ Return whether the Webmention has an unprocessed source revision. :return: Whether the desired revision still requires processing. """ return ( self.processed_revision is None or self.processed_revision < self.desired_revision ) @property def created_at(self) -> datetime: """ Return the creation time encoded in the Webmention UUID. :return: Creation time as a timezone-aware datetime. """ return uuid7_to_datetime(self.uuid) def uuid7_to_datetime(identifier: uuid.UUID) -> datetime: """ Convert a UUID version 7 timestamp to a datetime. :param identifier: UUID whose embedded timestamp should be converted. :return: Timestamp as a timezone-aware UTC datetime. """ return datetime.fromtimestamp(identifier.time / 1000, tz=UTC)