feat: Implement comprehensive credit-based cost calculation and lifecycle management (#836)

* feature: support aleph credits

* feat: parse new credit airdrop messages and refactor credit_balances table and primary key

* fix: change credit amount type from Decimal to int

* fix: lint errors

* feat: implement credit cost calculation and garbage collector

* point to tmp branch for aleph-message package

* fix: added missing migration for error code

* fix: balance Decimal type in http response

* fix: fix migrations order

* feat: filter accepted credit messages with channel + address + post types

* feat: remove /api/v0/addresses/{address}/credit_balance and integrate it in /api/v0/addresses/{address}/balance

* feat: support credit distribution, transfer, expense

* fix: added more tests

* fix: credit_history table and fifo calculation taking into account expiration and expenses

* feat: credit history endpoint + lint fixes

* fix: validate_balance_for_payment to take into account all resources + new one for the next 24 hours

* fix: validate_balance_for_payment to take into account all resources + new one for the next 24 hours

* Solve format issue on balance pre-check (#855)

* Fix: Solve issue on format getting IPFS file sizes before downloading the entire content.

* Fix: Added test cae for string dag_node format

* Fix RabbitMQ connection timeouts during long operations. (#856)

* Fix IPFS conf: use the recommended method to update IPFS configuration (#846)

* fix: lint

* fix: pending_message queue getting stuck due to bad error handling

* Fix: Added a default internal code for errors.

* fix: bugfix caching expired credit balances

* fix: lint fix

* fix: CI fixes due to hatch problematic dep

* feat: new index in credit_history.expiration_date column

* fix: added cost_credit in cost http response

* fix: change pricing aggregate owner for testing

* feat: added charged_address in /price and /price/estimate response

* Fix: Upgrade `aleph_message` version to `1.0.5`

* Fix: Pin pydantic version to the previous one from latest.

* Fix: Solved failing test and put legacy and current PAYG message tests.

* fix: fix account addresses for credit and settings

* fix: new credit channel

---------

Co-authored-by: nesitor <[email protected]>
Co-authored-by: Alie.E <[email protected]>

Author: 3 people

.github/workflows/code-quality.yml

@@ -18,7 +18,7 @@ jobs:
       - name: Install pip and hatch
         run: |
           sudo apt-get install -y python3-pip
-          pip3 install hatch hatch-vcs
+          pip3 install "click<8.2" hatch hatch-vcs
 
       - name: Cache dependencies
         uses: actions/cache@v4

deployment/docker-build/dev/Dockerfile

@@ -54,7 +54,7 @@ COPY .git ./.git
 COPY src ./src
 
 RUN pip install -e .
-RUN pip install hatch
+RUN pip install "click<8.2" hatch
 
 FROM base
 

deployment/docker-build/test/Dockerfile

@@ -53,7 +53,7 @@ COPY src ./src
 
 # Install project deps and test deps
 RUN pip install -e .[testing,docs]
-RUN pip install hatch
+RUN pip install "click<8.2" hatch
 
 # Install project test deps
 RUN apt-get update && apt-get install -y \

deployment/migrations/versions/0036_a1b2c3d4e5f6_add_credit_balances_table.py

@@ -0,0 +1,70 @@
+"""add credit_history and credit_balances tables
+
+Revision ID: a1b2c3d4e5f6
+Revises: 35a67ccc4451
+Create Date: 2025-07-14 00:00:00.000000
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.sql import func
+
+
+# revision identifiers, used by Alembic.
+revision = 'a1b2c3d4e5f6'
+down_revision = '35a67ccc4451'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Create credit_history table (detailed audit trail)
+    op.create_table(
+        'credit_history',
+        sa.Column('id', sa.BigInteger(), autoincrement=True),
+        sa.Column('address', sa.String(), nullable=False),
+        sa.Column('amount', sa.BigInteger(), nullable=False),
+        sa.Column('ratio', sa.DECIMAL(), nullable=True),
+        sa.Column('tx_hash', sa.String(), nullable=True),
+        sa.Column('token', sa.String(), nullable=True),
+        sa.Column('chain', sa.String(), nullable=True),
+        sa.Column('provider', sa.String(), nullable=True),
+        sa.Column('origin', sa.String(), nullable=True),
+        sa.Column('origin_ref', sa.String(), nullable=True),
+        sa.Column('payment_method', sa.String(), nullable=True),
+        sa.Column('credit_ref', sa.String(), nullable=False),
+        sa.Column('credit_index', sa.Integer(), nullable=False),
+        sa.Column('expiration_date', sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column('message_timestamp', sa.TIMESTAMP(timezone=True), nullable=False),
+        sa.Column('last_update', sa.TIMESTAMP(timezone=True), nullable=False, 
+                  server_default=func.now(), onupdate=func.now()),
+        sa.PrimaryKeyConstraint('credit_ref', 'credit_index', name='credit_history_pkey'),
+    )
+    
+    # Create indexes on credit_history for efficient lookups
+    op.create_index('ix_credit_history_address', 'credit_history', ['address'], unique=False)
+    op.create_index('ix_credit_history_message_timestamp', 'credit_history', ['message_timestamp'], unique=False)
+    
+    # Create credit_balances table (cached balance summary)
+    op.create_table(
+        'credit_balances',
+        sa.Column('address', sa.String(), nullable=False),
+        sa.Column('balance', sa.BigInteger(), nullable=False, default=0),
+        sa.Column('last_update', sa.TIMESTAMP(timezone=True), nullable=False, 
+                  server_default=func.now(), onupdate=func.now()),
+        sa.PrimaryKeyConstraint('address', name='credit_balances_pkey'),
+    )
+    
+    # Create index on address for the cached balances table
+    op.create_index('ix_credit_balances_address', 'credit_balances', ['address'], unique=False)
+
+
+def downgrade() -> None:
+    # Drop the credit_balances table
+    op.drop_index('ix_credit_balances_address', 'credit_balances')
+    op.drop_table('credit_balances')
+    
+    # Drop the credit_history table
+    op.drop_index('ix_credit_history_address', 'credit_history')
+    op.drop_index('ix_credit_history_message_timestamp', 'credit_history')
+    op.drop_table('credit_history')

deployment/migrations/versions/0037_b7c8d9e0f1a2_add_cost_credit_column.py

@@ -0,0 +1,38 @@
+"""add cost_credit column to account_costs table
+
+Revision ID: b7c8d9e0f1a2
+Revises: a1b2c3d4e5f6
+Create Date: 2025-01-11 00:00:00.000000
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'b7c8d9e0f1a2'
+down_revision = 'a1b2c3d4e5f6'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Add cost_credit column to account_costs table
+    op.add_column('account_costs', sa.Column('cost_credit', sa.DECIMAL(), nullable=False, server_default='0'))
+    
+    # Add missing CREDIT_INSUFFICIENT error code
+    op.execute(
+        """
+        INSERT INTO error_codes(code, description) VALUES 
+            (6, 'Insufficient credit')
+        ON CONFLICT (code) DO NOTHING
+        """
+    )
+
+
+def downgrade() -> None:
+    # Remove CREDIT_INSUFFICIENT error code
+    op.execute("DELETE FROM error_codes WHERE code = 6")
+    
+    # Remove cost_credit column from account_costs table
+    op.drop_column('account_costs', 'cost_credit')

deployment/migrations/versions/0038_c8d9e0f1a2b3_add_credit_balance_cron_job.py

@@ -0,0 +1,34 @@
+"""add credit_balance cron job
+
+Revision ID: c8d9e0f1a2b3
+Revises: b7c8d9e0f1a2
+Create Date: 2025-01-11 00:00:00.000000
+
+"""
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = 'c8d9e0f1a2b3'
+down_revision = 'b7c8d9e0f1a2'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Add credit_balance cron job to run every hour (3600 seconds)
+    op.execute(
+        """
+        INSERT INTO cron_jobs(id, interval, last_run) 
+        VALUES ('credit_balance', 3600, '2025-01-01 00:00:00')
+        """
+    )
+
+
+def downgrade() -> None:
+    # Remove credit_balance cron job
+    op.execute(
+        """
+        DELETE FROM cron_jobs WHERE id = 'credit_balance'
+        """
+    )

deployment/migrations/versions/0039_d0e1f2a3b4c5_add_credit_history_expiration_date_index.py

@@ -0,0 +1,32 @@
+"""add credit_history expiration_date index for cache invalidation performance
+
+Revision ID: d0e1f2a3b4c5
+Revises: c8d9e0f1a2b3
+Create Date: 2025-01-11 00:00:00.000000
+
+"""
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = 'd0e1f2a3b4c5'
+down_revision = 'c8d9e0f1a2b3'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Add index on expiration_date for efficient cache invalidation queries
+    # This index optimizes the query that checks if any credits expired
+    # after the cached balance was last updated
+    op.create_index(
+        'ix_credit_history_expiration_date',
+        'credit_history',
+        ['expiration_date'],
+        unique=False
+    )
+
+
+def downgrade() -> None:
+    # Drop the expiration_date index
+    op.drop_index('ix_credit_history_expiration_date', 'credit_history')

pyproject.toml

@@ -28,7 +28,7 @@ dependencies = [
   "aiohttp-jinja2==1.6",
   "aioipfs~=0.7.1",
   "alembic==1.15.1",
-  "aleph-message~=1.0.2",
+  "aleph-message~=1.0.5",
   "aleph-nuls2==0.1",
   "aleph-p2p-client @ git+https://github.com/aleph-im/p2p-service-client-python@cbfebb871db94b2ca580e66104a67cd730c5020c",
   "asyncpg==0.30",
@@ -47,7 +47,7 @@ dependencies = [
   "orjson>=3.7.7",                                                                                                         # Minimum version for Python 3.11
   "psycopg2-binary==2.9.10",                                                                                               # Note: psycopg3 is not yet supported by SQLAlchemy
   "pycryptodome==3.22.0",
-  "pydantic>=2.0.0,<3.0.0",
+  "pydantic==2.11.10",
   "pymultihash==0.8.2",                                                                                                    # for libp2p-stubs
   "pynacl==1.5",
   "pytezos-crypto==3.13.4.1",

src/aleph/commands.py

@@ -30,6 +30,7 @@
 from aleph.exceptions import InvalidConfigException, KeyNotFoundException
 from aleph.jobs import start_jobs
 from aleph.jobs.cron.balance_job import BalanceCronJob
+from aleph.jobs.cron.credit_balance_job import CreditBalanceCronJob
 from aleph.jobs.cron.cron_job import CronJob, cron_job_task
 from aleph.network import listener_tasks
 from aleph.services import p2p
@@ -151,7 +152,10 @@ async def main(args: List[str]) -> None:
         )
         cron_job = CronJob(
             session_factory=session_factory,
-            jobs={"balance": BalanceCronJob(session_factory=session_factory)},
+            jobs={
+                "balance": BalanceCronJob(session_factory=session_factory),
+                "credit_balance": CreditBalanceCronJob(session_factory=session_factory),
+            },
         )
         chain_data_service = ChainDataService(
             session_factory=session_factory,

src/aleph/config.py

@@ -33,6 +33,20 @@ def get_defaults():
                 # POST message type for balance updates.
                 "post_type": "balances-update",
             },
+            "credit_balances": {
+                # Addresses allowed to publish credit balance updates.
+                "addresses": [
+                    "0x2E4454fAD1906c0Ce6e45cBFA05cE898Ac3AC1dC",
+                ],
+                # POST message types for credit balance updates.
+                "post_types": [
+                    "aleph_credit_distribution",
+                    "aleph_credit_transfer",
+                    "aleph_credit_expense",
+                ],
+                # Allowed channels for credit balance messages.
+                "channels": ["ALEPH_CREDIT"],
+            },
             "jobs": {
                 "pending_messages": {
                     # Maximum number of retries for a message.