"""SQLAlchemy ORM models and engine helpers for PostgreSQL persistence."""
from __future__ import annotations
import uuid
from datetime import datetime
from uuid import UUID
from sqlalchemy import ARRAY, BigInteger, Boolean, DateTime, Float, ForeignKey, String, Text, UniqueConstraint
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from corvix.env import get_env_value
[docs]
class Base(DeclarativeBase):
"""Shared declarative base for all ORM models."""
[docs]
class User(Base):
"""Registered user with an encrypted GitHub token."""
[docs]
__tablename__ = "users"
[docs]
id: Mapped[UUID] = mapped_column(postgresql.UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
[docs]
github_login: Mapped[str] = mapped_column(Text, unique=True, nullable=False)
[docs]
github_token: Mapped[str] = mapped_column(Text, nullable=False) # Fernet-encrypted
[docs]
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
[docs]
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
[docs]
preferences: Mapped[UserPreferences | None] = relationship("UserPreferences", back_populates="user", uselist=False)
[docs]
push_subscriptions: Mapped[list[PushSubscription]] = relationship("PushSubscription", back_populates="user")
[docs]
notification_records: Mapped[list[NotificationRecordRow]] = relationship(
"NotificationRecordRow", back_populates="user"
)
[docs]
class NotificationRecordRow(Base):
"""Persisted notification record scoped to a user."""
[docs]
__tablename__ = "notification_records"
[docs]
__table_args__ = (UniqueConstraint("user_id", "account_id", "thread_id"),)
[docs]
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
[docs]
user_id: Mapped[UUID] = mapped_column(postgresql.UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
[docs]
account_id: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
account_label: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
thread_id: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
repository: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
reason: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
subject_title: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
subject_type: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
unread: Mapped[bool] = mapped_column(Boolean, nullable=False)
[docs]
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
[docs]
thread_url: Mapped[str | None] = mapped_column(Text, nullable=True)
[docs]
web_url: Mapped[str | None] = mapped_column(Text, nullable=True)
[docs]
score: Mapped[float] = mapped_column(Float, nullable=False)
[docs]
excluded: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
[docs]
matched_rules: Mapped[list[str]] = mapped_column(ARRAY(String), default=list)
[docs]
actions_taken: Mapped[list[str]] = mapped_column(ARRAY(String), default=list)
[docs]
context: Mapped[dict[str, object]] = mapped_column(postgresql.JSONB, nullable=False, default=dict)
[docs]
dismissed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
[docs]
snapshot_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
[docs]
user: Mapped[User] = relationship("User", back_populates="notification_records")
[docs]
class UserPreferences(Base):
"""Per-user preferences (theme, browser notifications)."""
[docs]
__tablename__ = "user_preferences"
[docs]
user_id: Mapped[UUID] = mapped_column(postgresql.UUID(as_uuid=True), ForeignKey("users.id"), primary_key=True)
[docs]
theme: Mapped[str] = mapped_column(Text, nullable=False, default="default")
[docs]
browser_notify: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
[docs]
user: Mapped[User] = relationship("User", back_populates="preferences")
[docs]
class PushSubscription(Base):
"""Browser push subscription for a user."""
[docs]
__tablename__ = "push_subscriptions"
[docs]
__table_args__ = (UniqueConstraint("user_id", "endpoint"),)
[docs]
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
[docs]
user_id: Mapped[UUID] = mapped_column(postgresql.UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
[docs]
endpoint: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
p256dh_key: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
auth_key: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
[docs]
user: Mapped[User] = relationship("User", back_populates="push_subscriptions")
[docs]
def get_database_url(url_env: str = "DATABASE_URL") -> str | None:
"""Return DB URL from env, supporting `${URL_ENV}_FILE` Docker secret files."""
return get_env_value(url_env)