Screen stablecoin flows against a live watchlist with Turbo, and attest every compliance decision with Compose.
Onchain settlement is often described as a compliance problem. In practice it is a compliance advantage - every transfer is visible, in real time, on a public ledger. The challenge is operational: you need to watch every flow as it happens, react to sanctions changes in minutes, and prove to a regulator exactly what you did and why.This guide builds that surveillance layer. A Turbo pipeline screens every stablecoin transfer against a live watchlist you can update in seconds, and Compose turns a hit into an attested, auditable decision.
Two properties make onchain monitoring different from a nightly batch screen:
Sanctions lists change without warning. When OFAC adds an address, you cannot wait for tomorrow’s redeploy. A dynamic table lets you push a new address to a running pipeline and start flagging within seconds.
Regulators want the “why,” not just the “what.”Compose traces every external call with inputs and outputs and can run in a Trusted Execution Environment that attests the exact code executed. Each freeze, alert, or SAR trigger comes with a verifiable record.
A Postgres database and a Goldsky secret. This database also backs the dynamic-table watchlist, so you can keep it inside your own compliance boundary.
The stablecoin/TIP-20 contract addresses you settle in. Confirm the Tempo dataset name in the Datasource explorer.
The watchlist lives in a dynamic table. The pipeline checks both counterparties of every transfer against it with dynamic_table_check().
aml-screening.yaml
name: stablecoin-aml-screeningresource_size: ssources: tempo_transfers: type: dataset dataset_name: tempo.tip20_transfers version: 1.0.0 start_at: latesttransforms: # The watchlist - updatable at runtime with no redeploy watchlist: type: dynamic_table backend_type: Postgres backend_entity_name: sanctioned_addresses secret_name: MY_POSTGRES # Flag any transfer touching a watchlisted address flagged_transfers: type: sql primary_key: id sql: | SELECT id, address AS token, sender, recipient, amount, block_timestamp, transaction_hash, CASE WHEN dynamic_table_check('watchlist', sender) THEN 'sender' WHEN dynamic_table_check('watchlist', recipient) THEN 'recipient' END AS watchlist_hit, _gs_op FROM tempo_transfers WHERE dynamic_table_check('watchlist', sender) OR dynamic_table_check('watchlist', recipient)sinks: # Persist every hit for case management and audit flagged_sink: type: postgres from: flagged_transfers schema: compliance table: flagged_transfers secret_name: MY_POSTGRES primary_key: id # Push each hit to your case-management system immediately case_management: type: webhook from: flagged_transfers url: https://compliance.example.com/alerts/aml-hit one_row_per_request: true secret_name: COMPLIANCE_WEBHOOK_AUTH
This is the capability that batch systems cannot match. When a new address must be watched, insert it into the dynamic table from any Postgres client - no redeploy, no re-sync:
-- New OFAC SDN address - live within secondsINSERT INTO streamling.sanctioned_addresses (value)VALUES (lower('0xNEWLY_SANCTIONED_ADDRESS'));
The running pipeline picks it up on the next batch and begins flagging matching transfers immediately. You can automate this by having your sanctions-list ingestion job write directly to the table.
Because the watchlist is a table in your database, you can populate it from OFAC’s published SDN list, a vendor feed (Chainalysis, TRM, Elliptic), and your own internal blocklist at once - and you can prune it by updated_at to satisfy data-retention rules.
TIP-20’s compliance controls emit their own events. Decode them from raw_logs to keep a real-time view of every policy change and blocked transfer - the issuer-side counterpart to your own screening.Decoding happens in two steps - one transform defines decoded, the next filters on it - because Flink SQL can’t reference a column alias in the WHERE of the same query.
policy-events.yaml
name: tip20-policy-eventsresource_size: ssources: tempo_tip20_logs: type: dataset dataset_name: tempo.raw_logs version: 1.0.0 start_at: latest filter: address = lower('0xYOUR_TIP20_TOKEN')transforms: # 1. Decode every log with the TIP-20 ABI decoded_logs: type: sql primary_key: id sql: | SELECT id, address AS token, _gs_log_decode( _gs_fetch_abi('https://raw.githubusercontent.com/your-org/abis/main/tip20.json', 'raw'), topics, data ) AS decoded, block_timestamp, transaction_hash, _gs_op FROM tempo_tip20_logs # 2. Keep only compliance/policy events (this transform can filter on `decoded`) policy_events: type: sql primary_key: id sql: | SELECT id, token, decoded.event_signature AS event, to_timestamp(block_timestamp) AS at, transaction_hash, _gs_op FROM decoded_logs WHERE decoded IS NOT NULL AND decoded.event_signature IN ( 'TransferPolicyUpdate', 'TransferBlocked', 'BurnBlocked', 'PauseStateUpdate' )sinks: policy_ledger: type: postgres from: policy_events schema: compliance table: policy_events secret_name: MY_POSTGRES primary_key: id
The four events this keeps are the issuer-side enforcement signals:
TransferPolicyUpdate - the token’s whitelist/blacklist policy (TIP-403) changed.
TransferBlocked - a transfer was redirected by receive-policy enforcement.
BurnBlocked - tokens were removed from a policy-restricted address.
PauseStateUpdate - the token was paused or unpaused.
A flag is not a decision. Route each hit to a Compose task that verifies against your KYC/sanctions provider and records the outcome. Running the task in a TEE produces an attestation that the exact screening logic ran, which is what makes the decision defensible.
Trigger this task from the case_management webhook in Step 1, or directly from an onchain event trigger. Every run is traceable step by step in the CLI and UI.
