from datetime import datetime from decimal import Decimal from enum import Enum from typing import Optional from pydantic import BaseModel, Field class AccountType(str, Enum): ASSET = "asset" LIABILITY = "liability" EQUITY = "equity" REVENUE = "revenue" EXPENSE = "expense" class JournalEntryFlag(str, Enum): """Transaction status flags (Beancount-compatible) Beancount only supports two user-facing flags: - * (CLEARED): Completed transactions - ! (PENDING): Transactions needing attention For voided/flagged transactions, use tags instead: - Voided: Use "!" flag + #voided tag - Flagged: Use "!" flag + #review tag """ CLEARED = "*" # Fully reconciled/confirmed PENDING = "!" # Not yet confirmed/awaiting approval class Account(BaseModel): id: str name: str account_type: AccountType description: Optional[str] = None user_id: Optional[str] = None # For user-specific accounts created_at: datetime is_active: bool = True # Soft delete flag is_virtual: bool = False # Virtual parent account (metadata-only, not in Beancount) class CreateAccount(BaseModel): name: str account_type: AccountType description: Optional[str] = None user_id: Optional[str] = None is_virtual: bool = False # Set to True to create virtual parent account class EntryLine(BaseModel): id: str journal_entry_id: str account_id: str amount: int # in satoshis; positive = debit, negative = credit description: Optional[str] = None metadata: dict = {} # Stores currency info: fiat_currency, fiat_amount, fiat_rate, etc. class CreateEntryLine(BaseModel): account_id: str amount: int # in satoshis; positive = debit, negative = credit description: Optional[str] = None metadata: dict = {} # Stores currency info class JournalEntry(BaseModel): id: str description: str entry_date: datetime created_by: str # wallet ID of user who created it created_at: datetime reference: Optional[str] = None # Invoice ID or reference number lines: list[EntryLine] = [] flag: JournalEntryFlag = JournalEntryFlag.CLEARED # Transaction status meta: dict = {} # Metadata: source, tags, links, notes, etc. class CreateJournalEntry(BaseModel): description: str entry_date: Optional[datetime] = None reference: Optional[str] = None lines: list[CreateEntryLine] flag: JournalEntryFlag = JournalEntryFlag.CLEARED meta: dict = {} class UserBalance(BaseModel): user_id: str balance: int # positive = castle owes user, negative = user owes castle accounts: list[Account] = [] fiat_balances: dict[str, Decimal] = {} # e.g. {"EUR": Decimal("250.0"), "USD": Decimal("100.0")} class ExpenseEntry(BaseModel): """Helper model for creating expense entries""" description: str amount: Decimal # Amount in the specified currency (or satoshis if currency is None) expense_account: str # account name or ID is_equity: bool = False # True = equity contribution, False = liability (castle owes user) user_wallet: str reference: Optional[str] = None currency: Optional[str] = None # If None, amount is in satoshis. Otherwise, fiat currency code (EUR, USD, etc.) entry_date: Optional[datetime] = None # Date of the expense transaction class ReceivableEntry(BaseModel): """Helper model for creating accounts receivable entries""" description: str amount: Decimal # Amount in the specified currency (or satoshis if currency is None) revenue_account: str # account name or ID user_id: str # The user_id (not wallet_id) of the user who owes the castle reference: Optional[str] = None currency: Optional[str] = None # If None, amount is in satoshis. Otherwise, fiat currency code class RevenueEntry(BaseModel): """Helper model for creating revenue entries""" description: str amount: Decimal # Amount in the specified currency (or satoshis if currency is None) revenue_account: str payment_method_account: str # e.g., "Cash", "Bank", "Lightning" reference: Optional[str] = None currency: Optional[str] = None # If None, amount is in satoshis. Otherwise, fiat currency code class CastleSettings(BaseModel): """Settings for the Castle extension""" castle_wallet_id: Optional[str] = None # The wallet ID that represents the Castle # Fava/Beancount integration - ALL accounting is done via Fava fava_url: str = "http://localhost:3333" # Base URL of Fava server fava_ledger_slug: str = "castle-ledger" # Ledger identifier in Fava URL fava_timeout: float = 10.0 # Request timeout in seconds updated_at: datetime = Field(default_factory=lambda: datetime.now()) @classmethod def is_admin_only(cls) -> bool: return True class UserCastleSettings(CastleSettings): """User-specific settings (stored with user_id)""" id: str class UserWalletSettings(BaseModel): """Per-user wallet settings""" user_wallet_id: Optional[str] = None # The wallet ID for this specific user updated_at: datetime = Field(default_factory=lambda: datetime.now()) class StoredUserWalletSettings(UserWalletSettings): """Stored user wallet settings with user ID""" id: str # user_id class ManualPaymentRequest(BaseModel): """Manual payment request from user to castle""" id: str user_id: str amount: int # in satoshis description: str status: str = "pending" # pending, approved, rejected created_at: datetime reviewed_at: Optional[datetime] = None reviewed_by: Optional[str] = None # user_id of castle admin who reviewed journal_entry_id: Optional[str] = None # set when approved class CreateManualPaymentRequest(BaseModel): """Create a manual payment request""" amount: int description: str class GeneratePaymentInvoice(BaseModel): """Generate payment invoice request""" amount: int user_id: Optional[str] = None # For admin-generated settlement invoices class RecordPayment(BaseModel): """Record a payment""" payment_hash: str class SettleReceivable(BaseModel): """Manually settle a receivable (user pays castle in person)""" user_id: str amount: Decimal # Amount in the specified currency (or satoshis if currency is None) payment_method: str # "cash", "bank_transfer", "check", "lightning", "btc_onchain", "other" description: str # Description of the payment reference: Optional[str] = None # Optional reference (receipt number, transaction ID, etc.) currency: Optional[str] = None # If None, amount is in satoshis. Otherwise, fiat currency code (EUR, USD, etc.) amount_sats: Optional[int] = None # Equivalent amount in sats (for reference/conversion tracking) payment_hash: Optional[str] = None # For lightning payments txid: Optional[str] = None # For on-chain Bitcoin transactions class PayUser(BaseModel): """Pay a user (castle pays user for expense/liability)""" user_id: str amount: Decimal # Amount in the specified currency (or satoshis if currency is None) payment_method: str # "cash", "bank_transfer", "check", "lightning", "btc_onchain", "other" description: Optional[str] = None # Description of the payment reference: Optional[str] = None # Optional reference (receipt number, transaction ID, etc.) currency: Optional[str] = None # If None, amount is in satoshis. Otherwise, fiat currency code (EUR, USD, etc.) amount_sats: Optional[int] = None # Equivalent amount in sats (for reference/conversion tracking) payment_hash: Optional[str] = None # For lightning payments txid: Optional[str] = None # For on-chain Bitcoin transactions class AssertionStatus(str, Enum): """Status of a balance assertion""" PENDING = "pending" # Not yet checked PASSED = "passed" # Assertion passed (balance matches) FAILED = "failed" # Assertion failed (balance mismatch) class BalanceAssertion(BaseModel): """Assert expected balance at a specific date for reconciliation""" id: str date: datetime account_id: str expected_balance_sats: int # Expected balance in satoshis expected_balance_fiat: Optional[Decimal] = None # Optional fiat balance fiat_currency: Optional[str] = None # Currency for fiat balance (EUR, USD, etc.) tolerance_sats: int = 0 # Allow +/- this much difference in sats tolerance_fiat: Decimal = Decimal("0") # Allow +/- this much difference in fiat checked_balance_sats: Optional[int] = None # Actual balance found checked_balance_fiat: Optional[Decimal] = None # Actual fiat balance found difference_sats: Optional[int] = None # Difference in sats difference_fiat: Optional[Decimal] = None # Difference in fiat status: AssertionStatus = AssertionStatus.PENDING created_by: str created_at: datetime checked_at: Optional[datetime] = None class CreateBalanceAssertion(BaseModel): """Create a balance assertion""" account_id: str date: Optional[datetime] = None # If None, use current time expected_balance_sats: int expected_balance_fiat: Optional[Decimal] = None fiat_currency: Optional[str] = None tolerance_sats: int = 0 tolerance_fiat: Decimal = Decimal("0") class UserEquityStatus(BaseModel): """Tracks user's equity eligibility and status""" user_id: str # User's wallet ID is_equity_eligible: bool # Can user convert expenses to equity? equity_account_name: Optional[str] = None # e.g., "Equity:Alice" notes: Optional[str] = None # Admin notes granted_by: str # Admin who granted eligibility granted_at: datetime revoked_at: Optional[datetime] = None # If eligibility was revoked class CreateUserEquityStatus(BaseModel): """Create or update user equity eligibility""" user_id: str is_equity_eligible: bool equity_account_name: Optional[str] = None # Auto-generated as "Equity:User-{user_id}" if not provided notes: Optional[str] = None class UserInfo(BaseModel): """User information including equity eligibility""" user_id: str is_equity_eligible: bool equity_account_name: Optional[str] = None class PermissionType(str, Enum): """Types of permissions for account access""" READ = "read" # Can view account and its balance SUBMIT_EXPENSE = "submit_expense" # Can submit expenses to this account MANAGE = "manage" # Can modify account (admin level) class AccountPermission(BaseModel): """Defines which accounts a user can access""" id: str # Unique permission ID user_id: str # User's wallet ID (from invoice key) account_id: str # Account ID from accounts table permission_type: PermissionType granted_by: str # Admin user ID who granted permission granted_at: datetime expires_at: Optional[datetime] = None # Optional expiration notes: Optional[str] = None # Admin notes about this permission class CreateAccountPermission(BaseModel): """Create account permission""" user_id: str account_id: str permission_type: PermissionType expires_at: Optional[datetime] = None notes: Optional[str] = None class BulkGrantPermission(BaseModel): """Bulk grant same permission to multiple users""" user_ids: list[str] # List of user IDs to grant permission to account_id: str # Account to grant permission on permission_type: PermissionType # Type of permission to grant expires_at: Optional[datetime] = None # Optional expiration notes: Optional[str] = None # Notes for all permissions class BulkGrantResult(BaseModel): """Result of bulk grant operation""" granted: list[AccountPermission] # Successfully granted permissions failed: list[dict] # Failed grants with errors total: int # Total attempted success_count: int # Number of successful grants failure_count: int # Number of failed grants class AccountWithPermissions(BaseModel): """Account with user-specific permission metadata""" id: str name: str account_type: AccountType description: Optional[str] = None user_id: Optional[str] = None created_at: datetime is_active: bool = True # Soft delete flag is_virtual: bool = False # Virtual parent account (metadata-only) # Only included when filter_by_user=true user_permissions: Optional[list[PermissionType]] = None inherited_from: Optional[str] = None # Parent account ID if inherited # Hierarchical structure parent_account: Optional[str] = None # Parent account name level: Optional[int] = None # Depth in hierarchy (0 = top level) has_children: Optional[bool] = None # Whether this account has sub-accounts # ===== ROLE-BASED ACCESS CONTROL (RBAC) MODELS ===== class Role(BaseModel): """Role definition for RBAC system""" id: str name: str # Display name (e.g., "Employee", "Contractor") description: Optional[str] = None is_default: bool = False # Auto-assign this role to new users created_by: str # User ID who created the role created_at: datetime class CreateRole(BaseModel): """Create a new role""" name: str description: Optional[str] = None is_default: bool = False class UpdateRole(BaseModel): """Update an existing role""" name: Optional[str] = None description: Optional[str] = None is_default: Optional[bool] = None class RolePermission(BaseModel): """Permission granted to a role for a specific account""" id: str role_id: str account_id: str permission_type: PermissionType notes: Optional[str] = None created_at: datetime class CreateRolePermission(BaseModel): """Create a permission for a role""" role_id: str account_id: str permission_type: PermissionType notes: Optional[str] = None class UserRole(BaseModel): """Assignment of a user to a role""" id: str user_id: str # User's wallet ID role_id: str granted_by: str # Admin who assigned the role granted_at: datetime expires_at: Optional[datetime] = None notes: Optional[str] = None class AssignUserRole(BaseModel): """Assign a user to a role""" user_id: str role_id: str expires_at: Optional[datetime] = None notes: Optional[str] = None class RoleWithPermissions(BaseModel): """Role with its associated permissions and user count""" role: Role permissions: list[RolePermission] user_count: int # Number of users assigned to this role class UserWithRoles(BaseModel): """User information with their assigned roles""" user_id: str roles: list[Role] direct_permissions: list[AccountPermission] # Individual permissions not from roles