mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-08-11 11:47:50 +00:00
162 lines
4.5 KiB
Python
162 lines
4.5 KiB
Python
import os
|
|
from typing import Literal
|
|
|
|
from cashu.core.base import Token
|
|
from cashu.wallet.helpers import deserialize_token_from_string, send
|
|
from cashu.wallet.wallet import Wallet
|
|
|
|
from .core import db, get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
CurrencyUnit = Literal["sat", "msat"]
|
|
|
|
CASHU_MINTS = os.environ.get("CASHU_MINTS", "https://mint.minibits.cash/Bitcoin")
|
|
TRUSTED_MINTS = CASHU_MINTS.split(",")
|
|
PRIMARY_MINT_URL = TRUSTED_MINTS[0]
|
|
|
|
|
|
async def get_balance(unit: CurrencyUnit) -> int:
|
|
wallet = await Wallet.with_db(
|
|
PRIMARY_MINT_URL,
|
|
db=".wallet",
|
|
load_all_keysets=True,
|
|
unit=unit,
|
|
)
|
|
await wallet.load_proofs()
|
|
return wallet.available_balance.amount
|
|
|
|
|
|
async def recieve_token(
|
|
token: str,
|
|
) -> tuple[int, CurrencyUnit, str]: # amount, unit, mint_url
|
|
token_obj = deserialize_token_from_string(token)
|
|
if len(token_obj.keysets) > 1:
|
|
raise ValueError("Multiple keysets per token currently not supported")
|
|
|
|
wallet = await Wallet.with_db(
|
|
token_obj.mint,
|
|
db=".wallet",
|
|
load_all_keysets=True,
|
|
unit=token_obj.unit,
|
|
)
|
|
await wallet.load_mint(token_obj.keysets[0])
|
|
|
|
if token_obj.mint not in TRUSTED_MINTS:
|
|
return await swap_to_primary_mint(token_obj, wallet)
|
|
|
|
await wallet.redeem(token_obj.proofs)
|
|
return token_obj.amount, token_obj.unit, token_obj.mint
|
|
|
|
|
|
async def send_token(
|
|
amount: int, unit: CurrencyUnit, mint_url: str | None = None
|
|
) -> str:
|
|
wallet = await Wallet.with_db(
|
|
mint_url or PRIMARY_MINT_URL,
|
|
db=".wallet",
|
|
load_all_keysets=True,
|
|
unit=unit,
|
|
)
|
|
balance, token = await send(wallet, amount=amount, lock="", legacy=False)
|
|
return token
|
|
|
|
|
|
async def swap_to_primary_mint(
|
|
token_obj: Token, token_wallet: Wallet
|
|
) -> tuple[int, CurrencyUnit, str]:
|
|
logger.info(
|
|
"swap_to_primary_mint",
|
|
extra={
|
|
"mint": token_obj.mint,
|
|
"amount": token_obj.amount,
|
|
"unit": token_obj.unit,
|
|
},
|
|
)
|
|
if token_obj.unit == "sat":
|
|
amount_msat = token_obj.amount * 1000
|
|
elif token_obj.unit == "msat":
|
|
amount_msat = token_obj.amount
|
|
else:
|
|
raise ValueError("Invalid unit")
|
|
estimated_fee_sat = max(amount_msat // 1000 * 0.01, 2)
|
|
amount_msat_after_fee = amount_msat - estimated_fee_sat * 1000
|
|
primary_wallet = await Wallet.with_db(
|
|
PRIMARY_MINT_URL, db=".wallet", load_all_keysets=True, unit="sat"
|
|
)
|
|
await primary_wallet.load_mint()
|
|
|
|
minted_amount = amount_msat_after_fee // 1000
|
|
mint_quote = await primary_wallet.request_mint(minted_amount)
|
|
|
|
melt_quote = await token_wallet.melt_quote(mint_quote.request)
|
|
_ = await token_wallet.melt(
|
|
proofs=token_obj.proofs,
|
|
invoice=mint_quote.request,
|
|
fee_reserve_sat=melt_quote.fee_reserve,
|
|
quote_id=melt_quote.quote,
|
|
)
|
|
_ = await primary_wallet.mint(minted_amount, quote_id=mint_quote.quote)
|
|
|
|
return minted_amount, "sat", PRIMARY_MINT_URL
|
|
|
|
|
|
async def credit_balance(
|
|
cashu_token: str, key: db.ApiKey, session: db.AsyncSession
|
|
) -> int:
|
|
amount, unit, mint_url = await recieve_token(cashu_token)
|
|
if unit == "sat":
|
|
amount = amount * 1000
|
|
if mint_url != PRIMARY_MINT_URL:
|
|
raise ValueError("Mint URL is not supported by this proxy")
|
|
key.balance += amount
|
|
session.add(key)
|
|
await session.commit()
|
|
logger.info(
|
|
"Cashu token successfully redeemed and stored",
|
|
extra={"amount": amount, "unit": unit, "mint_url": mint_url},
|
|
)
|
|
return amount
|
|
|
|
|
|
async def send_to_lnurl(amount: int, unit: CurrencyUnit, lnurl: str) -> dict[str, int]:
|
|
raise NotImplementedError
|
|
|
|
|
|
async def periodic_payout() -> None:
|
|
logger.warning("periodic_payout, temporary not implemented")
|
|
|
|
|
|
# class Proof:
|
|
# """
|
|
# Represents an ecash bill
|
|
# """
|
|
|
|
|
|
# def redeem_to_proofs(self, token: str) -> list[Proof]:
|
|
# raise NotImplementedError
|
|
|
|
|
|
# class Payment:
|
|
# """
|
|
# Stores all cashu payment related data
|
|
# """
|
|
|
|
# def __init__(self, token: str) -> None:
|
|
# self.initial_token = token
|
|
# amount, unit, mint_url = self.parse_token(token)
|
|
# self.amount = amount
|
|
# self.unit = unit
|
|
# self.mint_url = mint_url
|
|
|
|
# self.claimed_proofs = redeem_to_proofs(token)
|
|
|
|
# def parse_token(self, token: str) -> tuple[int, CurrencyUnit, str]:
|
|
# raise NotImplementedError
|
|
|
|
# def refund_full(self) -> None:
|
|
# raise NotImplementedError
|
|
|
|
# def refund_partial(self, amount: int) -> None:
|
|
# raise NotImplementedError
|