lnbits-exts/castle
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases 2 Packages Wiki Activity Actions
castle/views_api.py
padreug cb7e4ee555 Enhances super user experience with Castle wallet 
Simplifies the user experience for super users by automatically using the Castle wallet for transactions, removing the need to configure a separate user wallet.

This change streamlines the workflow for super users by:
- Automatically assigning the Castle wallet to super users
- Hiding the user wallet configuration options in the UI
- Reloading user wallet settings to reflect the Castle wallet
2025-10-22 15:11:00 +02:00

582 lines
18 KiB
Python
Raw Blame History

from http import HTTPStatus
from fastapi import APIRouter, Depends, HTTPException
from lnbits.core.models import User, WalletTypeInfo
from lnbits.decorators import (
check_super_user,
check_user_exists,
require_admin_key,
require_invoice_key,
)
from lnbits.utils.exchange_rates import allowed_currencies, fiat_amount_as_satoshis
from .crud import (
create_account,
create_journal_entry,
get_account,
get_account_balance,
get_account_by_name,
get_account_transactions,
get_all_accounts,
get_all_journal_entries,
get_journal_entries_by_user,
get_journal_entry,
get_or_create_user_account,
get_user_balance,
)
from .models import (
Account,
AccountType,
CastleSettings,
CreateAccount,
CreateEntryLine,
CreateJournalEntry,
ExpenseEntry,
JournalEntry,
ReceivableEntry,
RevenueEntry,
UserBalance,
UserWalletSettings,
)
from .services import get_settings, get_user_wallet, update_settings, update_user_wallet
castle_api_router = APIRouter()
# ===== HELPER FUNCTIONS =====
async def check_castle_wallet_configured() -> str:
"""Ensure castle wallet is configured, return wallet_id"""
settings = await get_settings("admin")
if not settings or not settings.castle_wallet_id:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail="Castle wallet not configured. Please contact the super user to configure the Castle wallet in settings.",
)
return settings.castle_wallet_id
async def check_user_wallet_configured(user_id: str) -> str:
"""Ensure user has configured their wallet, return wallet_id"""
from lnbits.settings import settings as lnbits_settings
# If user is super user, use the castle wallet
if user_id == lnbits_settings.super_user:
castle_settings = await get_settings("admin")
if castle_settings and castle_settings.castle_wallet_id:
return castle_settings.castle_wallet_id
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail="Castle wallet not configured. Please configure the Castle wallet in settings.",
)
# For regular users, check their personal wallet
user_wallet = await get_user_wallet(user_id)
if not user_wallet or not user_wallet.user_wallet_id:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail="You must configure your wallet in settings before using this feature.",
)
return user_wallet.user_wallet_id
# ===== UTILITY ENDPOINTS =====
@castle_api_router.get("/api/v1/currencies")
async def api_get_currencies() -> list[str]:
"""Get list of allowed currencies for fiat conversion"""
return allowed_currencies()
# ===== ACCOUNT ENDPOINTS =====
@castle_api_router.get("/api/v1/accounts")
async def api_get_accounts() -> list[Account]:
"""Get all accounts in the chart of accounts"""
return await get_all_accounts()
@castle_api_router.post("/api/v1/accounts", status_code=HTTPStatus.CREATED)
async def api_create_account(
data: CreateAccount,
wallet: WalletTypeInfo = Depends(require_admin_key),
) -> Account:
"""Create a new account (admin only)"""
return await create_account(data)
@castle_api_router.get("/api/v1/accounts/{account_id}")
async def api_get_account(account_id: str) -> Account:
"""Get a specific account"""
account = await get_account(account_id)
if not account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Account not found"
)
return account
@castle_api_router.get("/api/v1/accounts/{account_id}/balance")
async def api_get_account_balance(account_id: str) -> dict:
"""Get account balance"""
balance = await get_account_balance(account_id)
return {"account_id": account_id, "balance": balance}
@castle_api_router.get("/api/v1/accounts/{account_id}/transactions")
async def api_get_account_transactions(account_id: str, limit: int = 100) -> list[dict]:
"""Get all transactions for an account"""
transactions = await get_account_transactions(account_id, limit)
return [
{
"journal_entry": entry.dict(),
"entry_line": line.dict(),
}
for entry, line in transactions
]
# ===== JOURNAL ENTRY ENDPOINTS =====
@castle_api_router.get("/api/v1/entries")
async def api_get_journal_entries(limit: int = 100) -> list[JournalEntry]:
"""Get all journal entries"""
return await get_all_journal_entries(limit)
@castle_api_router.get("/api/v1/entries/user")
async def api_get_user_entries(
wallet: WalletTypeInfo = Depends(require_invoice_key),
limit: int = 100,
) -> list[JournalEntry]:
"""Get journal entries created by the current user"""
return await get_journal_entries_by_user(wallet.wallet.id, limit)
@castle_api_router.get("/api/v1/entries/{entry_id}")
async def api_get_journal_entry(entry_id: str) -> JournalEntry:
"""Get a specific journal entry"""
entry = await get_journal_entry(entry_id)
if not entry:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Journal entry not found"
)
return entry
@castle_api_router.post("/api/v1/entries", status_code=HTTPStatus.CREATED)
async def api_create_journal_entry(
data: CreateJournalEntry,
wallet: WalletTypeInfo = Depends(require_invoice_key),
) -> JournalEntry:
"""Create a new journal entry"""
try:
return await create_journal_entry(data, wallet.wallet.id)
except ValueError as e:
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e))
# ===== SIMPLIFIED ENTRY ENDPOINTS =====
@castle_api_router.post("/api/v1/entries/expense", status_code=HTTPStatus.CREATED)
async def api_create_expense_entry(
data: ExpenseEntry,
wallet: WalletTypeInfo = Depends(require_invoice_key),
) -> JournalEntry:
"""
Create an expense entry for a user.
If is_equity=True, records as equity contribution.
If is_equity=False, records as liability (castle owes user).
If currency is provided, amount is converted from fiat to satoshis.
"""
# Check that castle wallet is configured
await check_castle_wallet_configured()
# Check that user has configured their wallet
await check_user_wallet_configured(wallet.wallet.user)
# Handle currency conversion
amount_sats = int(data.amount)
metadata = {}
if data.currency:
# Validate currency
if data.currency.upper() not in allowed_currencies():
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail=f"Currency '{data.currency}' not allowed. Use one of: {', '.join(allowed_currencies())}",
)
# Convert fiat to satoshis
amount_sats = await fiat_amount_as_satoshis(data.amount, data.currency)
# Store currency metadata
metadata = {
"fiat_currency": data.currency.upper(),
"fiat_amount": round(data.amount, ndigits=3),
"fiat_rate": amount_sats / data.amount if data.amount > 0 else 0,
"btc_rate": (data.amount / amount_sats * 100_000_000) if amount_sats > 0 else 0,
}
# Get or create expense account
expense_account = await get_account_by_name(data.expense_account)
if not expense_account:
# Try to get it by ID
expense_account = await get_account(data.expense_account)
if not expense_account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND,
detail=f"Expense account '{data.expense_account}' not found",
)
# Get or create user-specific account
if data.is_equity:
# Equity contribution
user_account = await get_or_create_user_account(
data.user_wallet, AccountType.EQUITY, "Member Equity"
)
else:
# Liability (castle owes user)
user_account = await get_or_create_user_account(
data.user_wallet, AccountType.LIABILITY, "Accounts Payable"
)
# Create journal entry
# DR Expense, CR User Account (Liability or Equity)
description_suffix = f" ({metadata['fiat_amount']} {metadata['fiat_currency']})" if metadata else ""
entry_data = CreateJournalEntry(
description=data.description + description_suffix,
reference=data.reference,
lines=[
CreateEntryLine(
account_id=expense_account.id,
debit=amount_sats,
credit=0,
description=f"Expense paid by user {data.user_wallet[:8]}",
metadata=metadata,
),
CreateEntryLine(
account_id=user_account.id,
debit=0,
credit=amount_sats,
description=f"{'Equity contribution' if data.is_equity else 'Amount owed to user'}",
metadata=metadata,
),
],
)
return await create_journal_entry(entry_data, wallet.wallet.id)
@castle_api_router.post("/api/v1/entries/receivable", status_code=HTTPStatus.CREATED)
async def api_create_receivable_entry(
data: ReceivableEntry,
wallet: WalletTypeInfo = Depends(require_admin_key),
) -> JournalEntry:
"""
Create an accounts receivable entry (user owes castle).
Admin only to prevent abuse.
"""
# Get or create revenue account
revenue_account = await get_account_by_name(data.revenue_account)
if not revenue_account:
revenue_account = await get_account(data.revenue_account)
if not revenue_account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND,
detail=f"Revenue account '{data.revenue_account}' not found",
)
# Get or create user-specific receivable account
user_receivable = await get_or_create_user_account(
data.user_wallet, AccountType.ASSET, "Accounts Receivable"
)
# Create journal entry
# DR Accounts Receivable (User), CR Revenue
entry_data = CreateJournalEntry(
description=data.description,
reference=data.reference,
lines=[
CreateEntryLine(
account_id=user_receivable.id,
debit=data.amount,
credit=0,
description=f"Amount owed by user {data.user_wallet[:8]}",
),
CreateEntryLine(
account_id=revenue_account.id,
debit=0,
credit=data.amount,
description="Revenue earned",
),
],
)
return await create_journal_entry(entry_data, wallet.wallet.id)
@castle_api_router.post("/api/v1/entries/revenue", status_code=HTTPStatus.CREATED)
async def api_create_revenue_entry(
data: RevenueEntry,
wallet: WalletTypeInfo = Depends(require_admin_key),
) -> JournalEntry:
"""
Create a revenue entry (castle receives payment).
Admin only.
"""
# Get revenue account
revenue_account = await get_account_by_name(data.revenue_account)
if not revenue_account:
revenue_account = await get_account(data.revenue_account)
if not revenue_account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND,
detail=f"Revenue account '{data.revenue_account}' not found",
)
# Get payment method account
payment_account = await get_account_by_name(data.payment_method_account)
if not payment_account:
payment_account = await get_account(data.payment_method_account)
if not payment_account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND,
detail=f"Payment account '{data.payment_method_account}' not found",
)
# Create journal entry
# DR Cash/Lightning/Bank, CR Revenue
entry_data = CreateJournalEntry(
description=data.description,
reference=data.reference,
lines=[
CreateEntryLine(
account_id=payment_account.id,
debit=data.amount,
credit=0,
description="Payment received",
),
CreateEntryLine(
account_id=revenue_account.id,
debit=0,
credit=data.amount,
description="Revenue earned",
),
],
)
return await create_journal_entry(entry_data, wallet.wallet.id)
# ===== USER BALANCE ENDPOINTS =====
@castle_api_router.get("/api/v1/balance")
async def api_get_my_balance(
wallet: WalletTypeInfo = Depends(require_invoice_key),
) -> UserBalance:
"""Get current user's balance with the Castle"""
return await get_user_balance(wallet.wallet.id)
@castle_api_router.get("/api/v1/balance/{user_id}")
async def api_get_user_balance(user_id: str) -> UserBalance:
"""Get a specific user's balance with the Castle"""
return await get_user_balance(user_id)
# ===== PAYMENT ENDPOINTS =====
@castle_api_router.post("/api/v1/pay-balance")
async def api_pay_balance(
amount: int,
wallet: WalletTypeInfo = Depends(require_invoice_key),
) -> dict:
"""
Record a payment from user to castle (reduces what user owes or what castle owes user).
This should be called after an invoice is paid.
"""
wallet_id = wallet.wallet.id
# Get user's receivable account (what user owes)
user_receivable = await get_or_create_user_account(
wallet_id, AccountType.ASSET, "Accounts Receivable"
)
# Get lightning account
lightning_account = await get_account_by_name("Lightning Balance")
if not lightning_account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found"
)
# Create journal entry
# DR Lightning Balance, CR Accounts Receivable (User)
entry_data = CreateJournalEntry(
description=f"Payment received from user {wallet_id[:8]}",
lines=[
CreateEntryLine(
account_id=lightning_account.id,
debit=amount,
credit=0,
description="Lightning payment received",
),
CreateEntryLine(
account_id=user_receivable.id,
debit=0,
credit=amount,
description="Payment applied to balance",
),
],
)
entry = await create_journal_entry(entry_data, wallet_id)
# Get updated balance
balance = await get_user_balance(wallet_id)
return {
"journal_entry": entry.dict(),
"new_balance": balance.balance,
"message": "Payment recorded successfully",
}
@castle_api_router.post("/api/v1/pay-user")
async def api_pay_user(
user_id: str,
amount: int,
wallet: WalletTypeInfo = Depends(require_admin_key),
) -> dict:
"""
Record a payment from castle to user (reduces what castle owes user).
Admin only.
"""
# Get user's payable account (what castle owes)
user_payable = await get_or_create_user_account(
user_id, AccountType.LIABILITY, "Accounts Payable"
)
# Get lightning account
lightning_account = await get_account_by_name("Lightning Balance")
if not lightning_account:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found"
)
# Create journal entry
# DR Accounts Payable (User), CR Lightning Balance
entry_data = CreateJournalEntry(
description=f"Payment to user {user_id[:8]}",
lines=[
CreateEntryLine(
account_id=user_payable.id,
debit=amount,
credit=0,
description="Payment made to user",
),
CreateEntryLine(
account_id=lightning_account.id,
debit=0,
credit=amount,
description="Lightning payment sent",
),
],
)
entry = await create_journal_entry(entry_data, wallet.wallet.id)
# Get updated balance
balance = await get_user_balance(user_id)
return {
"journal_entry": entry.dict(),
"new_balance": balance.balance,
"message": "Payment recorded successfully",
}
# ===== SETTINGS ENDPOINTS =====
@castle_api_router.get("/api/v1/settings")
async def api_get_settings(
user: User = Depends(check_user_exists),
) -> CastleSettings:
"""Get Castle settings"""
user_id = "admin"
settings = await get_settings(user_id)
# Return empty settings if not configured (so UI can show setup screen)
if not settings:
return CastleSettings()
return settings
@castle_api_router.put("/api/v1/settings")
async def api_update_settings(
data: CastleSettings,
user: User = Depends(check_super_user),
) -> CastleSettings:
"""Update Castle settings (super user only)"""
if not data.castle_wallet_id:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail="Castle wallet ID is required",
)
user_id = "admin"
return await update_settings(user_id, data)
# ===== USER WALLET ENDPOINTS =====
@castle_api_router.get("/api/v1/user/wallet")
async def api_get_user_wallet(
user: User = Depends(check_user_exists),
) -> UserWalletSettings:
"""Get current user's wallet settings"""
from lnbits.settings import settings as lnbits_settings
# If user is super user, return the castle wallet
if user.id == lnbits_settings.super_user:
castle_settings = await get_settings("admin")
if castle_settings and castle_settings.castle_wallet_id:
return UserWalletSettings(user_wallet_id=castle_settings.castle_wallet_id)
return UserWalletSettings()
# For regular users, get their personal wallet
settings = await get_user_wallet(user.id)
# Return empty settings if not configured (so UI can show setup screen)
if not settings:
return UserWalletSettings()
return settings
@castle_api_router.put("/api/v1/user/wallet")
async def api_update_user_wallet(
data: UserWalletSettings,
user: User = Depends(check_user_exists),
) -> UserWalletSettings:
"""Update current user's wallet settings"""
from lnbits.settings import settings as lnbits_settings
# Super user cannot set their wallet separately - it's always the castle wallet
if user.id == lnbits_settings.super_user:
raise HTTPException(
status_code=HTTPStatus.FORBIDDEN,
detail="Super user wallet is automatically set to the Castle wallet. Update Castle settings instead.",
)
if not data.user_wallet_id:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail="User wallet ID is required",
)
return await update_user_wallet(user.id, data)
Reference in a new issue View git blame Copy permalink