382 lines
12 KiB
Python
382 lines
12 KiB
Python
from datetime import UTC, date, datetime, time
|
|
from enum import Enum as PyEnum
|
|
from typing import TypedDict
|
|
|
|
from sqlalchemy import (
|
|
Column,
|
|
Date,
|
|
DateTime,
|
|
Enum,
|
|
Float,
|
|
ForeignKey,
|
|
Integer,
|
|
String,
|
|
Table,
|
|
Time,
|
|
UniqueConstraint,
|
|
select,
|
|
)
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from database import Base
|
|
|
|
|
|
class RoleConfig(TypedDict):
|
|
description: str
|
|
permissions: list["Permission"]
|
|
|
|
|
|
class Permission(str, PyEnum):
|
|
"""All available permissions in the system."""
|
|
|
|
# Audit permissions
|
|
VIEW_AUDIT = "view_audit"
|
|
FETCH_PRICE = "fetch_price"
|
|
|
|
# Profile permissions
|
|
MANAGE_OWN_PROFILE = "manage_own_profile"
|
|
|
|
# Invite permissions
|
|
MANAGE_INVITES = "manage_invites"
|
|
VIEW_OWN_INVITES = "view_own_invites"
|
|
|
|
# Exchange permissions (regular users)
|
|
CREATE_EXCHANGE = "create_exchange"
|
|
VIEW_OWN_EXCHANGES = "view_own_exchanges"
|
|
CANCEL_OWN_EXCHANGE = "cancel_own_exchange"
|
|
|
|
# Availability/Exchange permissions (admin)
|
|
MANAGE_AVAILABILITY = "manage_availability"
|
|
VIEW_ALL_EXCHANGES = "view_all_exchanges"
|
|
CANCEL_ANY_EXCHANGE = "cancel_any_exchange"
|
|
COMPLETE_EXCHANGE = "complete_exchange"
|
|
|
|
|
|
class InviteStatus(str, PyEnum):
|
|
"""Status of an invite."""
|
|
|
|
READY = "ready"
|
|
SPENT = "spent"
|
|
REVOKED = "revoked"
|
|
|
|
|
|
class ExchangeStatus(str, PyEnum):
|
|
"""Status of an exchange trade."""
|
|
|
|
BOOKED = "booked"
|
|
COMPLETED = "completed"
|
|
CANCELLED_BY_USER = "cancelled_by_user"
|
|
CANCELLED_BY_ADMIN = "cancelled_by_admin"
|
|
NO_SHOW = "no_show"
|
|
|
|
|
|
class TradeDirection(str, PyEnum):
|
|
"""Direction of a trade from the user's perspective."""
|
|
|
|
BUY = "buy" # User buys BTC, gives EUR
|
|
SELL = "sell" # User sells BTC, gets EUR
|
|
|
|
|
|
class BitcoinTransferMethod(str, PyEnum):
|
|
"""Bitcoin transfer method for exchange trades."""
|
|
|
|
ONCHAIN = "onchain"
|
|
LIGHTNING = "lightning"
|
|
|
|
|
|
# Role name constants
|
|
ROLE_ADMIN = "admin"
|
|
ROLE_REGULAR = "regular"
|
|
|
|
# Role definitions with their permissions
|
|
ROLE_DEFINITIONS: dict[str, RoleConfig] = {
|
|
ROLE_ADMIN: {
|
|
"description": "Administrator with audit/invite/exchange access",
|
|
"permissions": [
|
|
Permission.VIEW_AUDIT,
|
|
Permission.FETCH_PRICE,
|
|
Permission.MANAGE_INVITES,
|
|
Permission.MANAGE_AVAILABILITY,
|
|
Permission.VIEW_ALL_EXCHANGES,
|
|
Permission.CANCEL_ANY_EXCHANGE,
|
|
Permission.COMPLETE_EXCHANGE,
|
|
],
|
|
},
|
|
ROLE_REGULAR: {
|
|
"description": "Regular user with profile, invite, and exchange access",
|
|
"permissions": [
|
|
Permission.MANAGE_OWN_PROFILE,
|
|
Permission.VIEW_OWN_INVITES,
|
|
Permission.CREATE_EXCHANGE,
|
|
Permission.VIEW_OWN_EXCHANGES,
|
|
Permission.CANCEL_OWN_EXCHANGE,
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
# Association table: Role <-> Permission (many-to-many)
|
|
role_permissions = Table(
|
|
"role_permissions",
|
|
Base.metadata,
|
|
Column(
|
|
"role_id",
|
|
Integer,
|
|
ForeignKey("roles.id", ondelete="CASCADE"),
|
|
primary_key=True,
|
|
),
|
|
Column("permission", Enum(Permission), primary_key=True),
|
|
)
|
|
|
|
|
|
# Association table: User <-> Role (many-to-many)
|
|
user_roles = Table(
|
|
"user_roles",
|
|
Base.metadata,
|
|
Column(
|
|
"user_id",
|
|
Integer,
|
|
ForeignKey("users.id", ondelete="CASCADE"),
|
|
primary_key=True,
|
|
),
|
|
Column(
|
|
"role_id",
|
|
Integer,
|
|
ForeignKey("roles.id", ondelete="CASCADE"),
|
|
primary_key=True,
|
|
),
|
|
)
|
|
|
|
|
|
class Role(Base):
|
|
__tablename__ = "roles"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
|
|
description: Mapped[str] = mapped_column(String(255), nullable=True)
|
|
|
|
# Relationship to users
|
|
users: Mapped[list["User"]] = relationship(
|
|
"User",
|
|
secondary=user_roles,
|
|
back_populates="roles",
|
|
)
|
|
|
|
async def get_permissions(self, db: AsyncSession) -> set[Permission]:
|
|
"""Get all permissions for this role."""
|
|
query = select(role_permissions.c.permission).where(
|
|
role_permissions.c.role_id == self.id
|
|
)
|
|
result = await db.execute(query)
|
|
return {row[0] for row in result.fetchall()}
|
|
|
|
async def set_permissions(
|
|
self, db: AsyncSession, permissions: list[Permission]
|
|
) -> None:
|
|
"""Set all permissions for this role (replaces existing)."""
|
|
delete_query = role_permissions.delete().where(
|
|
role_permissions.c.role_id == self.id
|
|
)
|
|
await db.execute(delete_query)
|
|
for perm in permissions:
|
|
insert_query = role_permissions.insert().values(
|
|
role_id=self.id, permission=perm
|
|
)
|
|
await db.execute(insert_query)
|
|
|
|
|
|
class User(Base):
|
|
__tablename__ = "users"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
email: Mapped[str] = mapped_column(
|
|
String(255), unique=True, nullable=False, index=True
|
|
)
|
|
hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
|
|
|
|
# Contact details (all optional)
|
|
contact_email: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
|
telegram: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
|
signal: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
|
nostr_npub: Mapped[str | None] = mapped_column(String(63), nullable=True)
|
|
|
|
# Godfather (who invited this user) - null for seeded/admin users
|
|
godfather_id: Mapped[int | None] = mapped_column(
|
|
Integer, ForeignKey("users.id"), nullable=True
|
|
)
|
|
godfather: Mapped["User | None"] = relationship(
|
|
"User",
|
|
remote_side="User.id",
|
|
foreign_keys=[godfather_id],
|
|
)
|
|
|
|
# Relationship to roles
|
|
roles: Mapped[list[Role]] = relationship(
|
|
"Role",
|
|
secondary=user_roles,
|
|
back_populates="users",
|
|
lazy="selectin",
|
|
)
|
|
|
|
async def get_permissions(self, db: AsyncSession) -> set[Permission]:
|
|
"""Get all permissions from all roles in a single query."""
|
|
result = await db.execute(
|
|
select(role_permissions.c.permission)
|
|
.join(user_roles, role_permissions.c.role_id == user_roles.c.role_id)
|
|
.where(user_roles.c.user_id == self.id)
|
|
)
|
|
return {row[0] for row in result.fetchall()}
|
|
|
|
async def has_permission(self, db: AsyncSession, permission: Permission) -> bool:
|
|
"""Check if user has a specific permission through any of their roles."""
|
|
permissions = await self.get_permissions(db)
|
|
return permission in permissions
|
|
|
|
@property
|
|
def role_names(self) -> list[str]:
|
|
"""Get list of role names for API responses."""
|
|
return [role.name for role in self.roles]
|
|
|
|
|
|
class Invite(Base):
|
|
__tablename__ = "invites"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
identifier: Mapped[str] = mapped_column(
|
|
String(64), unique=True, nullable=False, index=True
|
|
)
|
|
status: Mapped[InviteStatus] = mapped_column(
|
|
Enum(InviteStatus), nullable=False, default=InviteStatus.READY
|
|
)
|
|
|
|
# Godfather - the user who owns this invite
|
|
godfather_id: Mapped[int] = mapped_column(
|
|
Integer, ForeignKey("users.id"), nullable=False, index=True
|
|
)
|
|
godfather: Mapped[User] = relationship(
|
|
"User",
|
|
foreign_keys=[godfather_id],
|
|
lazy="joined",
|
|
)
|
|
|
|
# User who used this invite (null until spent)
|
|
used_by_id: Mapped[int | None] = mapped_column(
|
|
Integer, ForeignKey("users.id"), nullable=True
|
|
)
|
|
used_by: Mapped[User | None] = relationship(
|
|
"User",
|
|
foreign_keys=[used_by_id],
|
|
lazy="joined",
|
|
)
|
|
|
|
# Timestamps
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), default=lambda: datetime.now(UTC)
|
|
)
|
|
spent_at: Mapped[datetime | None] = mapped_column(
|
|
DateTime(timezone=True), nullable=True
|
|
)
|
|
revoked_at: Mapped[datetime | None] = mapped_column(
|
|
DateTime(timezone=True), nullable=True
|
|
)
|
|
|
|
|
|
class Availability(Base):
|
|
"""Admin availability slots for booking."""
|
|
|
|
__tablename__ = "availability"
|
|
__table_args__ = (
|
|
UniqueConstraint("date", "start_time", name="uq_availability_date_start"),
|
|
)
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
date: Mapped[date] = mapped_column(Date, nullable=False, index=True)
|
|
start_time: Mapped[time] = mapped_column(Time, nullable=False)
|
|
end_time: Mapped[time] = mapped_column(Time, nullable=False)
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), default=lambda: datetime.now(UTC)
|
|
)
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
default=lambda: datetime.now(UTC),
|
|
onupdate=lambda: datetime.now(UTC),
|
|
)
|
|
|
|
|
|
class PriceHistory(Base):
|
|
"""Price history records from external exchanges."""
|
|
|
|
__tablename__ = "price_history"
|
|
__table_args__ = (
|
|
UniqueConstraint("source", "pair", "timestamp", name="uq_price_source_pair_ts"),
|
|
)
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
source: Mapped[str] = mapped_column(String(50), nullable=False, index=True)
|
|
pair: Mapped[str] = mapped_column(String(20), nullable=False)
|
|
price: Mapped[float] = mapped_column(Float, nullable=False)
|
|
timestamp: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), nullable=False, index=True
|
|
)
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), default=lambda: datetime.now(UTC)
|
|
)
|
|
|
|
|
|
class Exchange(Base):
|
|
"""Bitcoin exchange trades booked by users."""
|
|
|
|
__tablename__ = "exchanges"
|
|
__table_args__ = (UniqueConstraint("slot_start", name="uq_exchange_slot_start"),)
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
user_id: Mapped[int] = mapped_column(
|
|
Integer, ForeignKey("users.id"), nullable=False, index=True
|
|
)
|
|
user: Mapped[User] = relationship("User", foreign_keys=[user_id], lazy="joined")
|
|
|
|
# Slot timing
|
|
slot_start: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), nullable=False, index=True
|
|
)
|
|
slot_end: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
|
|
|
|
# Trade details
|
|
direction: Mapped[TradeDirection] = mapped_column(
|
|
Enum(TradeDirection), nullable=False
|
|
)
|
|
bitcoin_transfer_method: Mapped[BitcoinTransferMethod] = mapped_column(
|
|
Enum(BitcoinTransferMethod),
|
|
nullable=False,
|
|
default=BitcoinTransferMethod.ONCHAIN,
|
|
)
|
|
eur_amount: Mapped[int] = mapped_column(Integer, nullable=False) # EUR cents
|
|
sats_amount: Mapped[int] = mapped_column(Integer, nullable=False) # Satoshis
|
|
|
|
# Price information (snapshot at booking time)
|
|
market_price_eur: Mapped[float] = mapped_column(
|
|
Float, nullable=False
|
|
) # EUR per BTC
|
|
agreed_price_eur: Mapped[float] = mapped_column(
|
|
Float, nullable=False
|
|
) # EUR per BTC with premium
|
|
premium_percentage: Mapped[int] = mapped_column(
|
|
Integer, nullable=False
|
|
) # e.g. 5 for 5%
|
|
|
|
# Status
|
|
status: Mapped[ExchangeStatus] = mapped_column(
|
|
Enum(ExchangeStatus), nullable=False, default=ExchangeStatus.BOOKED
|
|
)
|
|
|
|
# Timestamps
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), default=lambda: datetime.now(UTC)
|
|
)
|
|
cancelled_at: Mapped[datetime | None] = mapped_column(
|
|
DateTime(timezone=True), nullable=True
|
|
)
|
|
completed_at: Mapped[datetime | None] = mapped_column(
|
|
DateTime(timezone=True), nullable=True
|
|
)
|