tgit/app/db/models.py

"""Data model (ARCHITECTURE §2.10). One schema for PostgreSQL (prod) and SQLite (tests).

Enums are stored as VARCHAR + CHECK rather than native PG enums: portable and painless to migrate.
"""

from datetime import UTC, datetime
from enum import StrEnum
from typing import Any, ClassVar

from sqlalchemy import (
    JSON,
    BigInteger,
    Boolean,
    DateTime,
    Enum,
    Float,
    ForeignKey,
    Index,
    Integer,
    LargeBinary,
    MetaData,
    String,
    Text,
    UniqueConstraint,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

# Stable constraint names so Alembic migrations stay deterministic across databases.
NAMING = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

# Surrogate keys: BIGINT on Postgres, INTEGER on SQLite (only INTEGER PRIMARY KEY autoincrements there).
Id = BigInteger().with_variant(Integer(), "sqlite")
Json = JSON().with_variant(JSONB(), "postgresql")
TgId = BigInteger


def utcnow() -> datetime:
    return datetime.now(UTC)


def enum(cls: type[StrEnum], name: str) -> Enum:
    return Enum(
        cls,
        name=name,
        native_enum=False,
        create_constraint=True,  # SQLAlchemy 2 skips the CHECK for non-native enums unless asked
        length=20,
        values_callable=lambda e: [m.value for m in e],
    )


class Role(StrEnum):
    MEMBER = "member"
    ADMIN = "admin"


class ChangeOp(StrEnum):
    ADD = "add"
    MODIFY = "modify"
    DELETE = "delete"


class Stage(StrEnum):
    CI = "ci"
    CD = "cd"


class PipelineStatus(StrEnum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    TIMEOUT = "timeout"
    SKIPPED = "skipped"


class AlertKind(StrEnum):
    CRASH = "crash"
    OOM = "oom"
    RESTART_LOOP = "restart_loop"
    UNHEALTHY = "unhealthy"
    OVERLOAD = "overload"
    DEPLOY_REFUSED = "deploy_refused"
    CD_FAILED = "cd_failed"
    DISK_LOW = "disk_low"
    DEPENDENCY_DOWN = "dependency_down"
    AGENT_ATTACK = "agent_attack"


class LogMode(StrEnum):
    TAIL = "tail"
    GREP = "grep"
    AGENT = "agent"


class Verdict(StrEnum):
    SAFE = "safe"
    ATTACK = "attack"
    ERROR = "error"


class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING)
    type_annotation_map: ClassVar[dict[Any, Any]] = {
        datetime: DateTime(timezone=True),
        dict[str, Any]: Json,
        list[str]: Json,
    }


class User(Base):
    __tablename__ = "users"

    tg_id: Mapped[int] = mapped_column(TgId, primary_key=True, autoincrement=False)
    username: Mapped[str | None] = mapped_column(String(64))
    full_name: Mapped[str] = mapped_column(String(256))
    role: Mapped[Role] = mapped_column(enum(Role, "role"), default=Role.MEMBER)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    invited_by: Mapped[int | None] = mapped_column(TgId, ForeignKey("users.tg_id"))
    joined_at: Mapped[datetime] = mapped_column(default=utcnow)
    last_seen_at: Mapped[datetime | None]


class Invite(Base):
    __tablename__ = "invites"

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    # sha256 hex of the token; the token itself is never stored (SPEC A-3)
    token_hash: Mapped[str] = mapped_column(String(64), unique=True)
    created_by: Mapped[int] = mapped_column(TgId, ForeignKey("users.tg_id"))
    created_at: Mapped[datetime] = mapped_column(default=utcnow)
    expires_at: Mapped[datetime]
    used_by: Mapped[int | None] = mapped_column(TgId, ForeignKey("users.tg_id"))
    used_at: Mapped[datetime | None]


class AccessAttempt(Base):
    __tablename__ = "access_attempts"
    __table_args__ = (Index("ix_access_attempts_tg_id_at", "tg_id", "at"),)

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    tg_id: Mapped[int] = mapped_column(TgId)
    at: Mapped[datetime] = mapped_column(default=utcnow)
    ok: Mapped[bool] = mapped_column(Boolean)


class FsmState(Base):
    __tablename__ = "fsm_state"

    key: Mapped[str] = mapped_column(String(255), primary_key=True)
    state: Mapped[str | None] = mapped_column(String(255))
    data: Mapped[dict[str, Any]] = mapped_column(default=dict)


class Draft(Base):
    __tablename__ = "drafts"

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    user_id: Mapped[int] = mapped_column(TgId, ForeignKey("users.tg_id", ondelete="CASCADE"), unique=True)
    base_sha: Mapped[str] = mapped_column(String(40))
    message: Mapped[str | None] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(default=utcnow)
    expires_at: Mapped[datetime]

    changes: Mapped[list["DraftChange"]] = relationship(
        back_populates="draft", cascade="all, delete-orphan", order_by="DraftChange.id"
    )


class DraftChange(Base):
    __tablename__ = "draft_changes"
    __table_args__ = (UniqueConstraint("draft_id", "path"),)

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    draft_id: Mapped[int] = mapped_column(Id, ForeignKey("drafts.id", ondelete="CASCADE"))
    path: Mapped[str] = mapped_column(String(500))
    op: Mapped[ChangeOp] = mapped_column(enum(ChangeOp, "change_op"))
    content: Mapped[bytes | None] = mapped_column(LargeBinary)

    draft: Mapped[Draft] = relationship(back_populates="changes")


class Push(Base):
    __tablename__ = "pushes"

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    user_id: Mapped[int] = mapped_column(TgId, ForeignKey("users.tg_id"))
    sha: Mapped[str] = mapped_column(String(40), unique=True)
    message: Mapped[str] = mapped_column(Text)
    paths: Mapped[list[str]] = mapped_column(default=list)
    chat_id: Mapped[int] = mapped_column(TgId)
    message_id: Mapped[int | None] = mapped_column(BigInteger)
    ci_status: Mapped[PipelineStatus] = mapped_column(enum(PipelineStatus, "ci_status"), default=PipelineStatus.PENDING)
    cd_status: Mapped[PipelineStatus] = mapped_column(enum(PipelineStatus, "cd_status"), default=PipelineStatus.PENDING)
    created_at: Mapped[datetime] = mapped_column(default=utcnow)


class PipelineRun(Base):
    __tablename__ = "pipeline_runs"
    __table_args__ = (UniqueConstraint("push_id", "stage"),)

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    push_id: Mapped[int] = mapped_column(Id, ForeignKey("pushes.id", ondelete="CASCADE"))
    # CI and CD are tasks of one SourceCraft workflow (check-and-deploy), hence "stage", not "workflow"
    stage: Mapped[Stage] = mapped_column(enum(Stage, "stage"))
    run_slug: Mapped[str] = mapped_column(String(50))
    status: Mapped[PipelineStatus] = mapped_column(enum(PipelineStatus, "run_status"))
    failed_cube: Mapped[str | None] = mapped_column(String(100))
    url: Mapped[str | None] = mapped_column(String(500))
    updated_at: Mapped[datetime] = mapped_column(default=utcnow, onupdate=utcnow)


class ContainerSnapshot(Base):
    __tablename__ = "container_snapshots"
    __table_args__ = (Index("ix_container_snapshots_container_at", "container", "at"),)

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    container: Mapped[str] = mapped_column(String(100))
    at: Mapped[datetime] = mapped_column(default=utcnow)
    restart_count: Mapped[int] = mapped_column(Integer)
    cpu_pct: Mapped[float] = mapped_column(Float)
    mem_pct: Mapped[float] = mapped_column(Float)
    health: Mapped[str | None] = mapped_column(String(20))


class Alert(Base):
    __tablename__ = "alerts"
    __table_args__ = (Index("ix_alerts_kind_service_resolved_at", "kind", "service", "resolved_at"),)

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    kind: Mapped[AlertKind] = mapped_column(enum(AlertKind, "alert_kind"))
    service: Mapped[str | None] = mapped_column(String(100))
    payload: Mapped[dict[str, Any]] = mapped_column(default=dict)
    count: Mapped[int] = mapped_column(Integer, default=1)
    opened_at: Mapped[datetime] = mapped_column(default=utcnow)
    last_at: Mapped[datetime] = mapped_column(default=utcnow)
    resolved_at: Mapped[datetime | None]

    deliveries: Mapped[list["AlertDelivery"]] = relationship(back_populates="alert", cascade="all, delete-orphan")


class AlertDelivery(Base):
    __tablename__ = "alert_deliveries"
    __table_args__ = (UniqueConstraint("alert_id", "chat_id"),)

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    alert_id: Mapped[int] = mapped_column(Id, ForeignKey("alerts.id", ondelete="CASCADE"))
    chat_id: Mapped[int] = mapped_column(TgId)
    message_id: Mapped[int] = mapped_column(BigInteger)

    alert: Mapped[Alert] = relationship(back_populates="deliveries")


class LogQuery(Base):
    __tablename__ = "log_queries"

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    user_id: Mapped[int] = mapped_column(TgId, ForeignKey("users.tg_id"))
    container: Mapped[str] = mapped_column(String(100))
    mode: Mapped[LogMode] = mapped_column(enum(LogMode, "log_mode"))
    query: Mapped[dict[str, Any]] = mapped_column(default=dict)
    lines: Mapped[int] = mapped_column(Integer, default=0)
    at: Mapped[datetime] = mapped_column(default=utcnow)


class AgentRequest(Base):
    __tablename__ = "agent_requests"

    id: Mapped[int] = mapped_column(Id, primary_key=True)
    user_id: Mapped[int] = mapped_column(TgId, ForeignKey("users.tg_id"))
    prompt: Mapped[str] = mapped_column(Text)
    verdict: Mapped[Verdict] = mapped_column(enum(Verdict, "verdict"))
    reason: Mapped[str | None] = mapped_column(Text)
    log_query: Mapped[dict[str, Any] | None]
    tokens: Mapped[int] = mapped_column(Integer, default=0)
    at: Mapped[datetime] = mapped_column(default=utcnow)