path: root/webmentions_ssg/models.py

# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me>
#
# 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"<User {self.username}>"

    @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"<ReceivedWebmention {self.source!r} -> {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"<Source {self.url!r}>"

    @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"<SentWebmention {self.target!r}>"

    @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)