NEAR_ft_contract_py

ft_contract.py

from near_sdk_py import Contract, view, call, init, AccountId
from near_sdk_py import Promise
from near_sdk_py.contract import StorageError, ContractPanic, InvalidInput, ContractStorage
from near_sdk_py.collections import LookupMap, UnorderedMap

class FungibleToken(Contract):

    # def safe_get(self, key):
    #     if not key in self.storage:
    #         raise StorageError
    #     return self.storage[key]

    # @property
    # def owner(self) -> str:
    #     return self.safe_get('owner')

    # @property
    # def total_supply(self) -> int:
    #     return self.safe_get('total_supply')

    # @property
    # def metadata(self) -> dict:
    #     return self.safe_get('metadata')

    # @property
    # def balances(self) -> dict[str, int]:
    #     return self.safe_get('balances')

    # @property
    # def storage_deposits(self) -> dict[str, int]:
    #     return self.safe_get('storage_deposits')

    # Mint new tokens
    def ft_mint_event(
        self,
        owner_id: AccountId,
        amount: int,
        memo: str | None
    ):
        """
        receiver_id: AccountId,
        amount: Integer
        """
        amount = int(amount)
        event = "ft_mint"
        if not memo:
            mint_log = {
                "owner_id": owner_id,
                "amount": amount,
            }
        else:
            mint_log = {
                "owner_id": owner_id,
                "amount": amount,
                "memo": memo
            }
        self.log_event(
            event, mint_log
        )

    # Burn tokens (extension)
    def ft_burn_event(
        self,
        owner_id: AccountId,
        amount: int,
        memo: str | None
    ):
        """
        amount: Integer
        """
        amount = int(amount)
        event = "ft_burn"
        burn_log = {
            "owner_id": owner_id,
            "amount": amount
        } | {"memo": memo} if memo else {}
        self.log_event(
            event, burn_log
        )

    def ft_transfer_event(
        self,
        old_owner_id: AccountId,
        new_owner_id: AccountId,
        amount: int,
        memo: str | None = None
    ):
        """
        amount: Integer
        """
        amount = int(amount)
        event = "ft_transfer"
        transfer_log = {
            "old_owner_id": old_owner_id,
            "new_owner_id": new_owner_id,
            "amount": amount
        } | {"memo": memo} if memo else {}
        self.log_event(
            event, transfer_log
        )

    @call
    def ft_transfer(
        self,
        receiver_id: AccountId,
        amount: int,
        memo: str | None = None
    ) -> None:
        """
        Transfer tokens to a receiver

        Pseudocode:
        1. Verify the caller is registered (has storage paid)
        2. Assert amount > 0
        3. Assert the caller has sufficient balance
        4. Deduct amount from caller's balance
        5. If receiver is not registered, panic
        6. Add amount to receiver's balance
        7. If memo is provided, log it with the transfer
        8. Log the transfer event
        """
        amount = int(amount)
        sender_id = self.predecessor_account_id
        # 1. Verify the caller is registered
        if not self.storage['storage_deposits'].get(sender_id):
            raise ContractPanic(f"Sender {sender_id} has not registred.")
        # 5. If receiver is not registered, panic
        if not self.storage['storage_deposits'].get(receiver_id):
            raise ContractPanic(f"Receiver {receiver_id} has not registred.")
        # 2
        if amount <= 0:
            raise ContractPanic(f"Token amount should be positive. amount: {amount}")
        # 3
        sender_balance = self.storage['balances'].get(sender_id, 0)
        if sender_balance < amount:
            raise ContractPanic(f"Not enough token balance, current balance: {sender_balance}")
        # 4 Deduct amount from caller's balance
        self.storage['balances'][sender_id] -= amount
        # 6. Add amount to receiver's balance
        self.storage['balances'][receiver_id] += amount

        # // Log the transfer with optional memo
        self.ft_transfer_event(sender_id, receiver_id, amount, memo)


    @view
    def ft_total_supply(self) -> int:
        """
        Return the total token supply

        Pseudocode:
        1. Return the total_supply state variable
        """
        total_supply = self.storage.get('total_supply', 0)
        return total_supply

    @view
    def ft_balance_of(
        self,
        account_id: AccountId
    ) -> int:
        """
        Return the token balance of the account_id

        Pseudocode:
        1. If account_id is not registered, return 0
        2. Return the balance for account_id from balances mapping
        """
        balance_of = self.storage['balances'].get(account_id, 0)
        return balance_of

    @call
    def storage_deposit(
        self,
        account_id: AccountId | None = None,
        registration_only: bool | None = None
    ):
        """
            Pay for account storage

            Pseudocode:
            1. Set target account to account_id or default to caller
            2. Calculate storage already paid and required minimum
            3. If account already registered and registration_only=True:
            - Refund attached deposit
            - Return current balance
            4. If attached deposit + existing storage payment >= minimum:
            - Register the account
            - If attached deposit > minimum and registration_only=True:
                - Refund excess deposit
            5. Return StorageBalance object with total/available amounts
        """
        # If an account was specified, use that. Otherwise, use the predecessor account.
        account_id = account_id or self.predecessor_account_id
        sender_id = self.predecessor_account_id
        registration_only = registration_only or False

        # Get the amount of $NEAR to deposit
        amount = self.attached_deposit

        # Get storage balance bounds
        bounds = self.storage_balance_bounds()
        min_balance_bound = int(bounds.get('min'))
        max_balance_bound = int(bounds.get('max')) if bounds.get('max') else None

        current_balance = self.storage_deposits.get(account_id)
        is_registered = current_balance is not None

        new_balance = 0
        refund = 0
        # If the account is already registered, refund the deposit.
        if is_registered and registration_only:
            self.log_info("The account is already registered, refunding the deposit")
            refund = amount

        # Register the account and refund any excess $NEAR
        elif not is_registered:
            if amount < min_balance_bound:
                ContractPanic("The attached deposit is less than the minimum storage balance")
            if registration_only:
                # Only keep the minimum required
                new_balance = min_balance_bound
                refund = amount - min_balance_bound
            else:
                # Accept the entire deposit up to max if specified
                if max_balance_bound is not None and amount > max_balance_bound:
                    new_balance = max_balance_bound
                    refund = amount - max_balance_bound
                else:
                    new_balance = amount

        elif is_registered and not registration_only:
            if max_balance_bound is not None:
                # Check if adding amount would exceed max
                if current_balance + amount > max_balance_bound:
                    refund = current_balance + amount - max_balance_bound
                    new_balance = max_balance_bound
                else:
                    new_balance = current_balance + amount
            else:
                # No max, add the entire amount
                new_balance = current_balance + amount

        # Register the account
        if new_balance and new_balance != current_balance:
            self.storage_deposits[account_id] = new_balance
            self.log_info(f"User {account_id} was registred with balance {new_balance}")
        # Refound excsesive deposit
        if refund > 0:
            Promise.create_batch(sender_id).transfer(refund)
            self.log_info(f"Will be refund to {sender_id} amount: {refund}")
        result = {
            "total": new_balance,
            "available": max(min_balance_bound - new_balance, 0)
        }
        return result

    @view
    def list_registred_accounts(self):
        return [self.storage_deposits.keys()]

    @view
    def storage_balance_bounds(self) -> dict:
        """
            Return the min/max storage balance requirements

            Pseudocode:
            1. Calculate minimum balance needed for account registration
            2. Return StorageBalanceBounds object with:
            - min: minimum required balance
            - max: None (unlimited) or maximum allowed deposit
        """
        result = {
            "min": self.storage.get('min_storage_balance'),
            "max": self.storage.get('min_storage_balance')
        }
        return result

    @view
    def storage_balance_of(
        self,
        account_id: AccountId
    ) -> dict | None:
        """
            Return storage balance info for account

            Pseudocode:
            1. If account is not registered, return None
            2. Calculate total storage balance
            3. Calculate available balance (total - minimum required)
            4. Return StorageBalance object with total/available amounts
        """
        storage_balance = self.storage.get('storage_deposits', {}).get(account_id)
        if not storage_balance:
            return None
        available_balance = storage_balance - self.storage.get('min_storage_balance')
        result =  {
            "total": storage_balance,
            "available": available_balance
        }
        return result

    @view
    def ft_metadata(self):
        """
        Return token metadata

        Pseudocode:
        1. Return the metadata object with:
           - spec: "ft-1.0.0"
           - name: Token name
           - symbol: Token symbol
           - decimals: Number of decimal places
           - icon: Optional icon (as data URL)
           - reference: Optional link to additional info
           - reference_hash: Optional hash of reference content
        """
        metadata = self.storage.get('metadata')
        return metadata

    # def __init__(self):
    #     self._storage = ContractStorage()
    #     # self.balances = UnorderedMap('balances')
    #     self.balances: UnorderedMap[str, int] = UnorderedMap("balances")
    #     # self.storage_deposits = UnorderedMap('storage_deposits')
    #     self.storage_deposits: UnorderedMap[str, int] = UnorderedMap("storage_deposits")
    def __init__(self):
        self._storage = LookupMap("storage")
        self.balances: LookupMap[str, int] = LookupMap("balances")
        self.storage_deposits: LookupMap[str, int] = LookupMap("storage_deposits")


    @init
    def new(
        self,
        owner_id: AccountId,
        total_supply: int,
        metadata: dict
    ):
        total_supply = int(total_supply)
        self.log_info(f"Total supply: {total_supply}")
        owner_id = str(owner_id)
        self.log_info(f"Total supply: {owner_id}")
        metadata = dict(metadata)
        self.log_info(f"Metadata: {metadata}")

        MIN_STORAGE_BALANCE = 2_350_000_000_000_000_000_000
        # Validate parameters
        if total_supply <= 0:
            raise InvalidInput("Initial supply must be positive")

        if not(0 < metadata['decimals'] <= 24):
            raise InvalidInput("Decimals must be between 0 and 24")

        # Set metadata
        self.storage['owner'] = owner_id
        self.storage['total_supply'] = total_supply
        self.storage['metadata'] = metadata
        self.log_info("Meta data was set")

        # Assign initial supply to owner
        # self.balances[owner_id] = total_supply
        self.balances[owner_id] = total_supply
        self.log_info("Balances storage was set")
        # stores the token balances for each account

        # Register owner for storage
        self.storage['min_storage_balance'] = MIN_STORAGE_BALANCE
        # self.storage_deposits[owner_id] = MIN_STORAGE_BALANCE
        self.storage_deposits[owner_id] = total_supply
        self.log_info("Storage deposite storage was set")
        # tracks the storage deposits made by each account

        # Emit event
        event = "ft_mint"
        memo = f"Initial supply mint for {owner_id}"
        mint_log = {
            "owner_id": owner_id,
            "amount": total_supply,
            "memo": memo
        }
        self.log_event(
            event, mint_log
        )