Merge pull request #60 from skalenetwork/update-credit-distributor

Update credit distributor - multiple source chains and custom value

Author: dmytrotkk

credit-distributor/README.md

@@ -1,6 +1,8 @@
 # credit-distributor
 
-`credit-distributor` is a service that listens for `PaymentReceived` events on the mainnet `CreditStation` contract and fulfills the corresponding payments on the schain `Ledger` contract.
+`credit-distributor` watches `PaymentReceived` events on one or more `CreditStation` deployments (the **sources**) and fulfills each payment on a single **destination** schain `Ledger`.
+
+Each source is an independent `CreditStation` deployment on its own chain, initialized with a unique on-chain `sourceId` (set via `setPaymentIdOffset`) so payment IDs never collide. The distributor reads from every source sequentially each cycle and forwards `event.value` (wei) as the native amount to the purchaser on the destination schain.
 
 ### Usage
 
@@ -10,17 +12,21 @@ Configuration is loaded from a local TOML file.
 
 1. Create your config:
 
-```bash
-cp config.toml.example config.toml
-```
+   ```bash
+   cp config.toml.example config.toml
+   ```
 
-2. Edit `config.toml` for your environment.
+2. Edit `config.toml`:
 
-See `config.toml.example` for the full list of required/optional settings and their meaning.
+   * `[general]` — destination `schain_name` and the distributor's `eth_private_key` (used to call `Ledger.fulfill` on the destination schain).
+   * `[destination]` — destination schain `endpoint` and `Ledger` `contract` address.
+   * `[[sources]]` — one entry per `CreditStation` deployment. Repeat the block for each source chain. Each needs a unique `name` (used as a state-file key), the source-chain `endpoint`, the `CreditStation` `contract` address, and an initial `from_block` to scan from.
 
-#### Running the service
+See `config.toml.example` for a complete annotated sample with two sources.
+
+The distributor maintains per-source progress in `state.json` (path configurable via `[general].state_file`). Adding a new source later just appends a `[[sources]]` block and starts at its configured `from_block`; existing sources keep their stored progress.
 
-Run the service:
+#### Running the service
 
 ```bash
 docker compose up -d --build

credit-distributor/config.toml.example

@@ -1,20 +1,28 @@
 [general]
 schain_name = 'test' # required
 eth_private_key = '0x0' # required
-from_block = 1234 # required
+state_file = 'state.json' # optional
 
-[contracts]
-mainnet = '0x0' # required
-schain = '0x0' # required
+[destination]
+endpoint = 'https://example.com' # required - destination schain RPC
+contract = '0x0' # required - Ledger contract address on destination schain
 
-[endpoints]
-mainnet = 'https://example.com' # required
-schain = 'https://example.com' # required
+# One [[sources]] entry per CreditStation deployment to monitor.
+# Each deployment must have been initialized with a unique on-chain `sourceId`
+# (see CreditStation.setPaymentIdOffset). The `name` is an off-chain label
+# used as a key in the state file and in logs.
+[[sources]]
+name = 'ethereum' # required - unique label
+endpoint = 'https://example.com' # required - source chain RPC
+contract = '0x0' # required - CreditStation contract address
+from_block = 1234 # required - initial block to start scanning from
+
+[[sources]]
+name = 'base'
+endpoint = 'https://example.com'
+contract = '0x0'
+from_block = 5678
 
 [agent] # optional
 loop_sleep = 120 # optional
-exception_sleep = 10 # optional
-
-[payment] # optional
-value_eth = 1 # optional
-value_wei = 10000 # optional
+exception_sleep = 10 # optional

credit-distributor/pyproject.toml

@@ -2,7 +2,7 @@
 name = "credit-distributor"
 version = "0.0.1"
 requires-python = ">=3.13"
-dependencies = ["skale.py==7.9dev0", "pydantic-settings==2.12.0"]
+dependencies = ["skale.py==7.14dev0", "pydantic-settings==2.12.0"]
 
 [project.optional-dependencies]
 dev = ["ruff==0.14.5", "mypy==1.7.1"]

credit-distributor/src/configs.py

@@ -29,33 +29,31 @@
     SettingsConfigDict,
     TomlConfigSettingsSource,
 )
-from skale.types.schain import SchainName
+from skale_core.types import SchainName
 
 CONFIG_FILEPATH = os.path.join(os.path.dirname(__file__), os.pardir, 'config.toml')
 
 
-class Endpoints(BaseModel):
-    mainnet: str
-    schain: str
+class Source(BaseModel):
+    name: str
+    endpoint: str
+    contract: str
+    from_block: int
+    source_id: int | None = None
 
 
-class Contracts(BaseModel):
-    mainnet: str
-    schain: str
+class Destination(BaseModel):
+    endpoint: str
+    contract: str
 
 
 class Agent(BaseModel):
     loop_sleep: int = 120
     exception_sleep: int = 10
 
 
-class Payment(BaseModel):
-    value_eth: int = 1
-
-
 class General(BaseModel):
     schain_name: SchainName
-    from_block: int
     eth_private_key: HexStr
     state_file: str = 'state.json'
 
@@ -64,10 +62,9 @@ class Config(BaseSettings):
     model_config = SettingsConfigDict(toml_file=CONFIG_FILEPATH)
 
     general: General
-    endpoints: Endpoints
-    contracts: Contracts
+    destination: Destination
+    sources: list[Source]
     agent: Agent = Agent()
-    payment: Payment = Payment()
 
     @classmethod
     def settings_customise_sources(

credit-distributor/src/main.py

@@ -26,73 +26,94 @@
 from skale.utils.web3_utils import init_web3
 from skale.wallets import Web3Wallet
 
-from src.configs import Config, get_config
+from src.configs import Config, Source, get_config
 from src.state import State, StateManager
 
 logger = logging.getLogger(__name__)
 
+SOURCE_ID_SHIFT = 248
+
+
+def extract_source_id(payment_id: int) -> int:
+    return payment_id >> SOURCE_ID_SHIFT
+
+
+def resolve_source_id(mainnet_cs: MainnetCreditStation) -> int:
+    last_payment_id = mainnet_cs.credit_station.contract.functions.getLastPaymentId().call()
+    return extract_source_id(last_payment_id)
+
 
 def run_distributor() -> None:
     config = get_config()
     state_manager = StateManager(state_file=config.general.state_file)
-    state = state_manager.load(config.general.from_block)
+    state = state_manager.load({s.name: s.from_block for s in config.sources})
 
-    schain_web3 = init_web3(config.endpoints.schain)
+    schain_web3 = init_web3(config.destination.endpoint)
     schain_wallet = Web3Wallet(config.general.eth_private_key, schain_web3)
-    mainnet_cs = MainnetCreditStation(config.endpoints.mainnet, config.contracts.mainnet)
-    schain_cs = SchainCreditStation(config.endpoints.schain, config.contracts.schain, schain_wallet)
+    schain_cs = SchainCreditStation(
+        config.destination.endpoint, config.destination.contract, schain_wallet
+    )
+    source_clients = {
+        source.name: MainnetCreditStation(source.endpoint, source.contract)
+        for source in config.sources
+    }
+
+    for source in config.sources:
+        source.source_id = resolve_source_id(source_clients[source.name])
+        logger.info(f'[{source.name}] Resolved on-chain source_id={source.source_id}')
+
     while True:
-        try:
-            logging.info('Starting credit distribution cycle')
-            state = distribute_credits(mainnet_cs, schain_cs, config, state)
-            state_manager.save(state)
-            logger.info(f'Sleeping for {config.agent.loop_sleep} seconds before next cycle')
-            sleep(config.agent.loop_sleep)
-        except Exception as e:
-            logging.exception(f'Error during credit distribution cycle: {e}')
-            logger.info(f'Sleeping for {config.agent.exception_sleep} seconds before retrying')
-            sleep(config.agent.exception_sleep)
-
-
-def distribute_credits(
+        logger.info('Starting credit distribution cycle')
+        for source in config.sources:
+            try:
+                state = distribute_credits_for_source(
+                    source, source_clients[source.name], schain_cs, config, state
+                )
+                state_manager.save(state)
+            except Exception as e:
+                logger.exception(
+                    f'Error processing source {source.name!r}: {e}; continuing with next'
+                )
+        logger.info(f'Sleeping for {config.agent.loop_sleep} seconds before next cycle')
+        sleep(config.agent.loop_sleep)
+
+
+def distribute_credits_for_source(
+    source: Source,
     mainnet_cs: MainnetCreditStation,
     schain_cs: SchainCreditStation,
     config: Config,
     state: State,
 ) -> State:
+    from_block = state.from_blocks[source.name]
+    logger.info(f'[{source.name}] Fetching events from block {from_block}')
     all_events = mainnet_cs.credit_station.get_payment_received_events(
-        from_block=state.from_block, schain_name=config.general.schain_name
+        from_block=from_block, schain_name=config.general.schain_name
     )
-    last_block = state.from_block
     for event in all_events:
-        fulfill_payment(event, schain_cs, config)
+        fulfill_payment(source, event, schain_cs)
 
-    last_block_with_event = 0
-    if len(all_events) != 0:
-        last_block_with_event = all_events[-1]['block_number']
+    if all_events:
+        state.from_blocks[source.name] = all_events[-1]['block_number'] + 1
     else:
-        logger.info('No new PaymentReceived events found.')
-    state.from_block = max(last_block, last_block_with_event + 1)
+        logger.info(f'[{source.name}] No new PaymentReceived events found.')
     return state
 
 
-def get_payment_wei_value(config: Config) -> int:
-    return config.payment.value_eth * 10**18
-
-
 def fulfill_payment(
-    event: PaymentReceivedEvent, schain_cs: SchainCreditStation, config: Config
+    source: Source,
+    event: PaymentReceivedEvent,
+    schain_cs: SchainCreditStation,
 ) -> None:
-    logger.info(f'Checking payment: {event["payment_id"]}')
-    is_fulfilled = schain_cs.ledger.is_fulfilled(event['payment_id'])
+    payment_id = event['payment_id']
+    logger.info(f'[{source.name}] Checking payment: {payment_id}')
+    is_fulfilled = schain_cs.ledger.is_fulfilled(payment_id)
     if not is_fulfilled:
-        logger.info(f'Fulfilling payment: {event["payment_id"]}')
-        schain_cs.ledger.fulfill(
-            event['payment_id'], event['to_address'], value=get_payment_wei_value(config)
-        )
-        logger.info(f'Payment {event["payment_id"]} fulfilled successfully.')
+        logger.info(f'[{source.name}] Fulfilling payment: {payment_id}')
+        schain_cs.ledger.fulfill(payment_id, event['to_address'], value=event['value'])
+        logger.info(f'[{source.name}] Payment {payment_id} fulfilled successfully.')
     else:
-        logger.debug(f'Payment {event["payment_id"]} is already fulfilled.')
+        logger.debug(f'[{source.name}] Payment {payment_id} is already fulfilled.')
 
 
 if __name__ == '__main__':

credit-distributor/src/state.py

@@ -17,7 +17,6 @@
 #   You should have received a copy of the GNU Affero General Public License
 #   along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
-import json
 import logging
 from pathlib import Path
 
@@ -27,30 +26,24 @@
 
 
 class State(BaseModel):
-    from_block: int
+    from_blocks: dict[str, int]
 
 
 class StateManager:
     def __init__(self, state_file: str | Path):
-        self.state_file = Path(state_file) if isinstance(state_file, str) else state_file
+        self.state_file = Path(state_file)
 
-    def load(self, initial_from_block: int) -> State:
+    def load(self, initial_from_blocks: dict[str, int]) -> State:
+        from_blocks = dict(initial_from_blocks)
         if self.state_file.exists():
-            try:
-                data = json.loads(self.state_file.read_text())
-                logger.info(f'Loaded state from {self.state_file}')
-                return State(**data)
-            except Exception as e:
-                logger.warning(
-                    f'Failed to load state from {self.state_file}: {e}. Using initial value.'
-                )
-        logger.info(f'Creating new state with from_block={initial_from_block}')
-        return State(from_block=initial_from_block)
+            stored = State.model_validate_json(self.state_file.read_text()).from_blocks
+            logger.info(f'Loaded state from {self.state_file}')
+            for name in from_blocks:
+                if name in stored:
+                    from_blocks[name] = stored[name]
+        else:
+            logger.info(f'Creating new state with from_blocks={from_blocks}')
+        return State(from_blocks=from_blocks)
 
     def save(self, state: State) -> None:
-        try:
-            self.state_file.write_text(state.model_dump_json(indent=2))
-            logger.debug(f'Saved state to {self.state_file}')
-        except Exception as e:
-            logger.error(f'Failed to save state to {self.state_file}: {e}')
-            raise
+        self.state_file.write_text(state.model_dump_json(indent=2))