from __future__ import annotations import uuid from datetime import datetime from enum import Enum as PyEnum from typing import TYPE_CHECKING from sqlalchemy import Boolean, CheckConstraint, DateTime, ForeignKey, Numeric, String, Text, UniqueConstraint, func from sqlalchemy.orm import Mapped, mapped_column, relationship, validates from backend.app.core.database import Base if TYPE_CHECKING: from backend.app.models.archive import PrintArchive from backend.app.models.print_queue import PrintQueueItem from backend.app.models.user import User class TransactionType(str, PyEnum): PRINT_CHARGE = "print_charge" DEPOSIT = "deposit" WITHDRAW = "withdraw" MANUAL_ADJUSTMENT = "manual_adjustment" VALID_TRANSACTION_TYPES = {item.value for item in TransactionType} def normalize_transaction_type(value: str | TransactionType) -> str: if isinstance(value, TransactionType): return value.value if value not in VALID_TRANSACTION_TYPES: raise ValueError(f"Invalid transaction type: {value}") return value class UserWallet(Base): """Per-user wallet balance. Balance updates are driven by wallet transactions. No currency column: an install has exactly one currency, held in the ``currency`` app setting, and nothing here converts between currencies. The column that used to sit on this table recorded whatever was configured when the row happened to be created, three of the four writers hardcoded "EUR" into it, and the Finance page rendered what it found -- so an install set to AUD reported euros (#3123). """ __tablename__ = "user_wallets" id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), unique=True, index=True) balance: Mapped[float] = mapped_column(Numeric(14, 2, asdecimal=False), default=0.0) updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now()) user: Mapped[User] = relationship() class CostCenter(Base): """Cost center for assigning print costs and budgets.""" __tablename__ = "cost_centers" id: Mapped[int] = mapped_column(primary_key=True) code: Mapped[str] = mapped_column(String(32), unique=True, index=True, default=lambda: uuid.uuid4().hex[:12]) name: Mapped[str] = mapped_column(String(150), index=True) is_active: Mapped[bool] = mapped_column(Boolean, default=True) is_private: Mapped[bool] = mapped_column(Boolean, default=False) owner_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True) total_budget: Mapped[float | None] = mapped_column(Numeric(14, 2, asdecimal=False), nullable=True) monthly_budget: Mapped[float | None] = mapped_column(Numeric(14, 2, asdecimal=False), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now()) owner: Mapped[User | None] = relationship() members: Mapped[list[CostCenterMember]] = relationship( "CostCenterMember", back_populates="cost_center", cascade="all, delete-orphan", lazy="selectin", ) class CostCenterMember(Base): """User-to-cost-center assignment with print permission.""" __tablename__ = "cost_center_members" __table_args__ = (UniqueConstraint("cost_center_id", "user_id", name="uq_cost_center_members_cc_user"),) id: Mapped[int] = mapped_column(primary_key=True) cost_center_id: Mapped[int] = mapped_column(ForeignKey("cost_centers.id", ondelete="CASCADE"), index=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True) can_print: Mapped[bool] = mapped_column(Boolean, default=True) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) cost_center: Mapped[CostCenter] = relationship("CostCenter", back_populates="members") user: Mapped[User] = relationship() class BudgetReservation(Base): """Persisted budget hold for accepted print work that has not been charged yet.""" __tablename__ = "budget_reservations" id: Mapped[int] = mapped_column(primary_key=True) cost_center_id: Mapped[int] = mapped_column(ForeignKey("cost_centers.id", ondelete="CASCADE"), index=True) amount: Mapped[float] = mapped_column(Numeric(14, 2, asdecimal=False)) status: Mapped[str] = mapped_column(String(20), default="active", index=True) source_type: Mapped[str] = mapped_column(String(50), index=True) source_id: Mapped[int | None] = mapped_column(index=True) print_archive_id: Mapped[int | None] = mapped_column( ForeignKey("print_archives.id", ondelete="SET NULL"), nullable=True, index=True ) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) released_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) cost_center: Mapped[CostCenter] = relationship() print_archive: Mapped[PrintArchive | None] = relationship() class WalletTransaction(Base): """Immutable wallet ledger entry.""" __tablename__ = "wallet_transactions" __table_args__ = ( CheckConstraint( "transaction_type IN ('print_charge', 'deposit', 'withdraw', 'manual_adjustment')", name="ck_wallet_transactions_transaction_type", ), ) id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True) cost_center_id: Mapped[int | None] = mapped_column( ForeignKey("cost_centers.id", ondelete="SET NULL"), nullable=True, index=True ) transaction_type: Mapped[str] = mapped_column(String(40), index=True) amount: Mapped[float] = mapped_column(Numeric(14, 2, asdecimal=False)) balance_after: Mapped[float | None] = mapped_column(Numeric(14, 2, asdecimal=False), nullable=True) description: Mapped[str | None] = mapped_column(Text, nullable=True) created_by_user_id: Mapped[int | None] = mapped_column( ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True ) print_run_id: Mapped[str | None] = mapped_column(String(100), nullable=True, index=True) print_archive_id: Mapped[int | None] = mapped_column( ForeignKey("print_archives.id", ondelete="SET NULL"), nullable=True, index=True ) print_queue_id: Mapped[int | None] = mapped_column( ForeignKey("print_queue.id", ondelete="SET NULL"), nullable=True, index=True ) # Voided ledger rows stay persisted as run-scoped idempotency tombstones. # They are excluded from balances and API listings, but their print_run_id # prevents a delayed duplicate completion callback from recreating a charge # that an administrator deliberately removed. is_voided: Mapped[bool] = mapped_column(Boolean, default=False, index=True) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), index=True) user: Mapped[User] = relationship(foreign_keys=[user_id]) cost_center: Mapped[CostCenter | None] = relationship() created_by: Mapped[User | None] = relationship(foreign_keys=[created_by_user_id]) print_archive: Mapped[PrintArchive | None] = relationship() print_queue: Mapped[PrintQueueItem | None] = relationship() @validates("transaction_type") def _validate_transaction_type(self, key: str, value: str | TransactionType) -> str: return normalize_transaction_type(value)