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:
| Datum | Description | Asset Parameter |
initial_redeemer | The account that can initially withdraw the subscription tokens. | reserve |
active_from | The round that the subscription is active from. | metadata_hash |
payment_amount | The amount of MicroAlgos that can be withdrawn each payment cycle. | metadata_hash |
payment_frequency | The number of rounds per payment cycle. | metadata_hash |
max_payments | The maximum number of payments that can be claimed for a subscription. | total |
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"