Tokenised Subscriptions on Algorand

Introduction

In this blog post, we'll explore a novel way to implement periodic payments on Algorand.

The general idea is to write a smart contract that issues redeemable subscription tokens.

Each subscription is represented by a unique asset, and each unit of the asset is a token that can be redeemed for a single subscription payment:

These tokens can be freely traded, allowing token holders to transfer them to the application account at the designated time and receive a payment.

The contract will be written in Algorand Python.

State

Instead of storing subscription details in the contract's state, we store the information in asset parameters:

The metadata_hash is an arbitrary field of 32 bytes (256 bits), so we can fit the bytes of four UInt64 values in it:

metadata_hash = active_from.bytes + payment_amount.bytes + payment_frequency.bytes + arc4.UInt64(Global.round).bytes

The Global.round isn't really needed, but since we have 8 bytes left at the end, I've added it anyway.

For convenience and readability, we'll provide methods for parsing these values from a subscription token's metadata hash:

@arc4.abimethod
def active_from(self, subscription: Asset) -> UInt64:
    return op.extract_uint64(subscription.metadata_hash, 0)

@arc4.abimethod
def payment_amount(self, subscription: Asset) -> UInt64:
    return op.extract_uint64(subscription.metadata_hash, 8)

@arc4.abimethod
def payment_frequency(self, subscription: Asset) -> UInt64:
    return op.extract_uint64(subscription.metadata_hash, 16)

As a side note, it would be cool if Algorand Python supports the @property decorator in the future.

Payment Schedules

The total units of the subscription token represents the maximum number of payments that can be made.

If there are 100 units of the asset, the subscription has 100 pay cycles.

An individual should be able to claim a payment if they own a unit of the token, and transfer it to the contract during the pay cycle interval.

The current cycle number should be zero for all rounds prior to the subscription's active_from round.

For all subsequent rounds, the cycle number is calculated as the number of rounds that have elapsed since the active_from round, floor divided by the payment_frequency, plus 1:

@arc4.abimethod
def cycle_number(self, subscription: Asset, at_round: UInt64) -> UInt64:
    active_from = self.active_from(subscription)
    return UInt64(0) if at_round < active_from else (at_round - active_from) // self.payment_frequency(subscription) + 1

For example, if the subscription is active_from round 5, with a payment_frequency of 1 round:

If active_from is round 1,000, with a payment_frequency of 1,000 rounds:

Claiming Payments

After receiving a subscription token, the contract's balance of the token must exactly equal the current cycle number in order for a payment to be made.

This prevents the token owner from withdrawing funds outside of the current pay cycle interval.

The only way for someone to receive a payment is by transferring a token to the contract, so the contract's balance of the token has to increase each time.

It should be impossible for someone to make two claims in the same payment cycle, because on the subsequent application call the balance will no longer equal the payment cycle number.

In the contract:

@arc4.abimethod
def claim_payment(self, axfer: gtxn.AssetTransferTransaction) -> UInt64:
    subscription = axfer.xfer_asset
    assert (
        subscription.creator == Global.current_application_address
    ), "Asset must have been created by this application"
    assert axfer.asset_receiver == Global.current_application_address, "Asset receiver must be application account"
    assert axfer.asset_amount == 1, "Asset amount must be 1"

    cycle_number = self.cycle_number(subscription, Global.round)
    next_balance = subscription.balance(Global.current_application_address) + 1
    assert not next_balance > cycle_number, "Cannot claim payment for future cycle"
    if next_balance == cycle_number:
        return itxn.Payment(
            receiver=axfer.sender,
            amount=self.payment_amount(subscription),
            fee=0,
        ).submit().amount
    return UInt64(0)

We first check that the asset being transferred to the application account was created by the contract (meaning it's a valid subscription token), and that a single unit of the asset is being transferred.

The contract then calculates the current pay cycle number and the proposed next balance of the contract (current balance + the single unit being transferred).

If the proposed next balance will be > the current cycle number, an error is thrown (preventing the transfer).

This should prevent someone accidentally trying to deposit the token prior to the corresponding pay cycle, but obviously there's nothing stopping someone transferring it without calling this method.

In the latter scenario, the token owner would effectively forfeit the right to claim a payment for that cycle.

If the proposed next balance exactly equals the cycle number, then a payment is made.

Otherwise, if it's <= the current cycle number (meaning the owner failed to claim some previous payments on time), the contract accepts the additional token but doesn't transfer any funds to the claimant.

The Complete Contract Code

The code below is also available on GitHub.

# pyright: reportMissingModuleSource=false
from algopy import Account, ARC4Contract, Asset, Global, Txn, UInt64, arc4, gtxn, itxn, op


class TokenisedSubscriptions(ARC4Contract):
    """A contract for subscription payments using redeemable NFTs."""

    @arc4.abimethod
    def mint_tokens(
        self,
        initial_redeemer: Account,
        active_from: arc4.UInt64,
        payment_amount: arc4.UInt64,
        payment_frequency: arc4.UInt64,
        max_payments: arc4.UInt64,
    ) -> UInt64:
        """Creates

        Args:
            initial_redeemer (Account): _description_
            active_from (arc4.UInt64): _description_
            payment_amount (arc4.UInt64): _description_
            payment_frequency (arc4.UInt64): _description_
            max_payments (arc4.UInt64): _description_

        Returns:
            UInt64: _description_
        """
        assert Txn.sender == Global.creator_address, "Only the contract creator can call this method"
        return (
            itxn.AssetConfig(
                asset_name=Global.creator_address.bytes,
                unit_name=arc4.UInt64(Global.current_application_address.total_assets_created).bytes,
                total=max_payments.native,
                decimals=0,
                metadata_hash=active_from.bytes
                + payment_amount.bytes
                + payment_frequency.bytes
                + arc4.UInt64(Global.round).bytes,
                manager=Global.current_application_address,
                reserve=initial_redeemer,
                freeze=Global.current_application_address,
                clawback=Global.current_application_address,
                fee=0,
            )
            .submit()
            .created_asset.id
        )

    @arc4.abimethod
    def withdraw_tokens(self, subscription: Asset) -> None:
        """Transfers the subscription tokens to the initial redeemer.

        Args:
            subscription (Asset): The subscription token.
        """
        assert subscription.reserve == Txn.sender
        itxn.AssetTransfer(
            xfer_asset=subscription, asset_receiver=Txn.sender, asset_amount=subscription.total, fee=0
        ).submit()

    @arc4.abimethod
    def active_from(self, subscription: Asset) -> UInt64:
        """Parses the 'active from' round number from the subscription token.

        Args:
            subscription (Asset): The subscription token.

        Returns:
            UInt64: The 'active from' round number.
        """
        return op.extract_uint64(subscription.metadata_hash, 0)

    @arc4.abimethod
    def payment_amount(self, subscription: Asset) -> UInt64:
        """Parses the payment amount from the subscription token.

        Args:
            subscription (Asset): The subscription token.

        Returns:
            UInt64: The payment amount in MicroAlgos.
        """
        return op.extract_uint64(subscription.metadata_hash, 8)

    @arc4.abimethod
    def payment_frequency(self, subscription: Asset) -> UInt64:
        """Parses the payment frequency from the subscription token.

        Args:
            subscription (Asset): The subscription token.

        Returns:
            UInt64: The payment frequency (number of rounds).
        """
        return op.extract_uint64(subscription.metadata_hash, 16)

    @arc4.abimethod
    def cycle_number(self, subscription: Asset, at_round: UInt64) -> UInt64:
        """Calculates the payment cycle number 'as at' a specific round.

        Args:
            subscription (Asset): The subscription token.
            at_round (UInt64): The round to calculate the payment cycle 'as at'.

        Returns:
            UInt64: The payment cycle number.
        """
        active_from = self.active_from(subscription)
        return (
            UInt64(0)
            if at_round < active_from
            else (at_round - active_from) // self.payment_frequency(subscription) + 1
        )

    @arc4.abimethod
    def claim_payment(self, axfer: gtxn.AssetTransferTransaction) -> UInt64:
        """Makes a payment to the subscription token owner, if eligible.

        Args:
            axfer (gtxn.AssetTransferTransaction): The transaction transferring a unit of the subscription token to the contract account.

        Returns:
            UInt64: The amount of MicroAlgos paid.
        """
        subscription = axfer.xfer_asset
        assert (
            subscription.creator == Global.current_application_address
        ), "Asset must have been created by this application"
        assert axfer.asset_receiver == Global.current_application_address, "Asset receiver must be application account"
        assert axfer.asset_amount == 1, "Asset amount must be 1"

        cycle_number = self.cycle_number(subscription, Global.round)
        next_balance = subscription.balance(Global.current_application_address) + 1
        assert not next_balance > cycle_number, "Cannot claim payment for future cycle"
        if next_balance == cycle_number:
            return (
                itxn.Payment(
                    receiver=axfer.sender,
                    amount=self.payment_amount(subscription),
                    fee=0,
                )
                .submit()
                .amount
            )
        return UInt64(0)

Writing Tests

The code below is also available on GitHub.

from base64 import b64decode

import algokit_utils
import pytest
from algokit_utils import TransactionParameters, TransferParameters, get_localnet_default_account, opt_in, transfer
from algokit_utils.config import config
from algosdk.atomic_transaction_composer import (
    AccountTransactionSigner,
    TransactionWithSigner,
)
from algosdk.transaction import AssetTransferTxn
from algosdk.v2client.algod import AlgodClient
from algosdk.v2client.indexer import IndexerClient
from pydantic import BaseModel
from sympy import Piecewise, symbols

from smart_contracts.artifacts.tokenised_subscriptions.client import TokenisedSubscriptionsClient


@pytest.fixture(scope="session")
def app_client(algod_client: AlgodClient, indexer_client: IndexerClient) -> TokenisedSubscriptionsClient:
    account = get_localnet_default_account(algod_client)

    config.configure(
        debug=True,
        # trace_all=True,
    )

    client = TokenisedSubscriptionsClient(
        algod_client,
        creator=account,
        indexer_client=indexer_client,
    )

    client.deploy(
        on_schema_break=algokit_utils.OnSchemaBreak.AppendApp,
        on_update=algokit_utils.OnUpdate.AppendApp,
    )

    transfer(
        algod_client,
        TransferParameters(
            from_account=get_localnet_default_account(algod_client),
            to_address=client.app_address,
            micro_algos=1_000_000_000,
        ),
    )

    return client


class Subscription(BaseModel):
    asset_id: int
    initial_redeemer: str
    active_from: int
    payment_amount: int
    payment_frequency: int
    max_payments: int


@pytest.fixture
def subscription(algod_client: AlgodClient, app_client: TokenisedSubscriptionsClient) -> Subscription:
    last_round = algod_client.status()["last-round"]
    account = get_localnet_default_account(algod_client)

    sp = app_client.algod_client.suggested_params()
    sp.fee = 2_000
    sp.flat_fee = True

    initial_redeemer = account.address
    active_from = last_round
    payment_amount = 1_000_000
    payment_frequency = 5
    max_payments = 10

    asset_id = app_client.mint_tokens(
        initial_redeemer=initial_redeemer,
        active_from=active_from,
        payment_amount=payment_amount,
        payment_frequency=payment_frequency,
        max_payments=max_payments,
        transaction_parameters=TransactionParameters(suggested_params=sp),
    ).return_value

    return Subscription(
        asset_id=asset_id,
        initial_redeemer=initial_redeemer,
        active_from=active_from,
        payment_amount=payment_amount,
        payment_frequency=payment_frequency,
        max_payments=max_payments,
    )


def test_mint_tokens(
    algod_client: AlgodClient, app_client: TokenisedSubscriptionsClient, subscription: Subscription
) -> None:
    """Tests the mint_tokens() and withdraw_tokens() methods."""
    assert isinstance(subscription.asset_id, int) and subscription.asset_id > 0, "Asset minting failed"

    app_balance = algod_client.account_asset_info(address=app_client.app_address, asset_id=subscription.asset_id)[
        "asset-holding"
    ]["amount"]
    assert app_balance == subscription.max_payments, "Asset not owned by application account"


def test_asset_params(
    algod_client: AlgodClient, app_client: TokenisedSubscriptionsClient, subscription: Subscription
) -> None:
    """Tests the subscription token's asset params."""
    asset_params = algod_client.asset_info(subscription.asset_id)["params"]

    assert asset_params["total"] == subscription.max_payments, "Asset total units is incorrect"
    assert (
        asset_params["creator"]
        == asset_params["manager"]
        == asset_params["freeze"]
        == asset_params["clawback"]
        == app_client.app_address
    ), "Incorrect asset params"
    assert asset_params["reserve"] == subscription.initial_redeemer, "Incorrect initial redeemer"

    # Check token state in metadata hash
    datum = b64decode(asset_params["metadata-hash"])
    btoi = lambda b: int.from_bytes(b, "big")
    active_from = btoi(datum[0:8])
    payment_amount = btoi(datum[8:16])
    payment_frequency = btoi(datum[16:24])

    assert active_from == subscription.active_from, "Incorrect active from round"
    assert payment_amount == subscription.payment_amount, "Incorrect payment amount"
    assert payment_frequency == subscription.payment_frequency, "Incorrect payment_frequency"


def test_withdraw_tokens(
    algod_client: AlgodClient, app_client: TokenisedSubscriptionsClient, subscription: Subscription
) -> None:
    """Tests the withdraw_tokens() method."""
    account = get_localnet_default_account(algod_client)
    opt_in(algod_client, account, asset_ids=[subscription.asset_id])

    sp = app_client.algod_client.suggested_params()
    sp.fee = 2_000
    sp.flat_fee = True

    balance_before = algod_client.account_asset_info(account.address, subscription.asset_id)["asset-holding"]["amount"]
    assert balance_before == 0, "Starting balance of token should be zero"
    app_client.withdraw_tokens(
        subscription=subscription.asset_id, transaction_parameters=TransactionParameters(suggested_params=sp)
    )
    balance_after = algod_client.account_asset_info(account.address, subscription.asset_id)["asset-holding"]["amount"]
    assert balance_after == subscription.max_payments, "Balance of token after withdrawing is incorrect"


def test_cycle_number(app_client: TokenisedSubscriptionsClient, subscription: Subscription) -> None:
    """Tests the cycle_number() method against a reference implementation in Sympy."""
    active_from, payment_frequency, at_round = symbols(
        r"{active\_from} {payment\_frequency} {at\_round}", integer=True, nonnegative=True
    )
    # Sympy reference implementation
    expr = Piecewise(
        (0, at_round < active_from), ((at_round - active_from) // payment_frequency + 1, True), evaluate=False
    )

    for r in range(subscription.active_from - 2, subscription.active_from + subscription.max_payments + 10):
        reference_value = int(
            expr.subs(
                [
                    (active_from, subscription.active_from),
                    (at_round, r),
                    (payment_frequency, subscription.payment_frequency),
                ]
            )
        )
        assert (
            app_client.cycle_number(subscription=subscription.asset_id, at_round=r).return_value == reference_value
        ), "Cycle number does not match Sympy reference implementation"


def test_claim_payment(
    algod_client: AlgodClient, app_client: TokenisedSubscriptionsClient, subscription: Subscription
) -> None:
    """Tests the claim_payment() method."""

    account = get_localnet_default_account(algod_client)
    opt_in(algod_client, account, asset_ids=[subscription.asset_id])

    sp = app_client.algod_client.suggested_params()
    sp.fee = 2_000
    sp.flat_fee = True

    # Withdraw subscription tokens first
    app_client.withdraw_tokens(
        subscription=subscription.asset_id, transaction_parameters=TransactionParameters(suggested_params=sp)
    )

    account_balance_start = algod_client.account_info(account.address)["amount"]

    fees = 0
    received = 0

    while True:
        token_balance = algod_client.account_asset_info(account.address, subscription.asset_id)["asset-holding"][
            "amount"
        ]

        if not token_balance:
            break
        try:
            axfer_txn = AssetTransferTxn(
                sp=sp,
                sender=account.address,
                receiver=app_client.app_address,
                amt=1,
                index=subscription.asset_id,
            )
            received += app_client.claim_payment(
                axfer=TransactionWithSigner(txn=axfer_txn, signer=AccountTransactionSigner(account.private_key))
            ).return_value

            fees += 3_000
        except algokit_utils.logic_error.LogicError:
            # Trigger another round on test net
            app_client.cycle_number(subscription=subscription.asset_id, at_round=0)
            fees += 1_000

    account_balance_end = algod_client.account_info(account.address)["amount"]
    token_balance_end = algod_client.account_asset_info(account.address, subscription.asset_id)["asset-holding"][
        "amount"
    ]
    assert token_balance_end == 0, "Tokens not transferred"
    assert (
        account_balance_end == account_balance_start + (subscription.max_payments * subscription.payment_amount) - fees
    ), "Incorrect balances after withdrawing"
    assert received == subscription.max_payments * subscription.payment_amount, "Incorrect amount received"