> ## Documentation Index
> Fetch the complete documentation index at: https://docs.goldsky.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Stablecoin compliance & AML monitoring

> 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](/turbo-pipelines/introduction) screens every stablecoin transfer against a **live watchlist** you can update in seconds, and [Compose](/compose/introduction) turns a hit into an attested, auditable decision.

## Why real-time and provable

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](/turbo-pipelines/transforms/dynamic-tables) lets you push a new address to a *running* pipeline and start flagging within seconds.
* **Regulators want the "why," not just the "what."** [Compose](/compose/introduction) 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.

## How it works

```mermaid theme={null}
flowchart LR
    A[TIP-20 transfers on Tempo] --> B[Turbo pipeline]
    W[(Live watchlist<br/>dynamic table)] --> B
    B -->|no hit| C[(Compliance ledger)]
    B -->|hit| D[Compose screening task]
    D -->|verify| E[Sanctions / KYC provider]
    D -->|attest + notify| F[Case management + on-chain attestation]
    OFAC[OFAC / internal update] -.->|INSERT, no redeploy| W
```

## Prerequisites

* The [Turbo CLI extension](/turbo-pipelines/cli#installation) installed and logged in.
* A Postgres database and a [Goldsky secret](/turbo-pipelines/pipeline-config#secrets). 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](https://app.goldsky.com/data-sources).

## Step 1: Screen transfers against a live watchlist

The watchlist lives in a dynamic table. The pipeline checks both counterparties of every transfer against it with `dynamic_table_check()`.

```yaml aml-screening.yaml expandable theme={null}
name: stablecoin-aml-screening
resource_size: s

sources:
  tempo_transfers:
    type: dataset
    dataset_name: tempo.tip20_transfers
    version: 1.0.0
    start_at: latest

transforms:
  # 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
```

```bash theme={null}
goldsky turbo apply aml-screening.yaml
```

## Step 2: React to a sanctions update in seconds

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:

```sql theme={null}
-- New OFAC SDN address - live within seconds
INSERT 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.

<Tip>
  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>

## Step 3: Monitor issuer-level enforcement

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.

```yaml policy-events.yaml expandable theme={null}
name: tip20-policy-events
resource_size: s

sources:
  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.

## Step 4: Attest every decision with Compose

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.

```typescript src/tasks/screen-transfer.ts theme={null}
import type { TaskContext } from "compose";

export async function main(
  context: TaskContext,
  payload: { transaction_hash: string; sender: string; recipient: string; amount: string; watchlist_hit: string },
) {
  const { fetch, evm } = context;

  // 1. Verify against your sanctions / KYC provider (retried durably)
  const screen = await fetch<{ decision: "clear" | "block"; caseId: string }>(
    "https://screening.example.com/v1/screen",
    {
      method: "POST",
      body: JSON.stringify(payload),
      max_attempts: 3,
      initial_interval_ms: 1000,
      backoff_factor: 2,
    },
  );

  // 2. Write a tamper-evident attestation on-chain
  const wallet = await evm.wallet({ name: "compliance-attestor", sponsorGas: true });
  await wallet.writeContract(
    evm.chains.tempo,
    "0xYOUR_ATTESTATION_REGISTRY",
    "recordDecision(bytes32,string,string)",
    [payload.transaction_hash, screen.decision, screen.caseId],
  );

  // 3. Return the decision - the full trace (inputs, outputs, attestation) is saved
  return { decision: screen.decision, caseId: screen.caseId };
}
```

Trigger this task from the `case_management` webhook in Step 1, or directly from an [onchain event trigger](/compose/task-triggers). Every run is [traceable](/compose/debugging) step by step in the CLI and UI.

## Business outcomes

* **Minutes, not days, to enforce a sanctions change** - the watchlist updates a running pipeline in seconds.
* **Defensible decisions** - TEE attestations and full execution traces answer "how did you decide?" with a record, not a reconstruction.
* **Data stays in your boundary** - the watchlist and flagged-transfer tables live in your database, easing residency and retention obligations.
* **One surveillance layer across chains and tokens** - the same pipeline pattern covers TIP-20 on Tempo and standard stablecoins on any [supported chain](/chains/supported-networks).

## Resources

* [Dynamic tables](/turbo-pipelines/transforms/dynamic-tables)
* [Compose task authoring](/compose/tasks) and [triggers](/compose/task-triggers)
* [Real-time reconciliation](/solutions/real-time-reconciliation)

Can't find what you're looking for? Reach out to us at [support@goldsky.com](mailto:support@goldsky.com) for help.