> ## 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. # Detect incoming assets to a set of addresses > Maintain a live set of addresses and know the moment assets land in any of them — with a queryable index, running balances, and webhook alerts. ## Overview Many products need to answer one deceptively simple question: **"Did assets just land in one of our addresses?"** Exchanges, neobanks, on-ramps, and payment apps all maintain a large, changing set of receiving addresses and need to react the instant funds arrive. Doing this with raw RPC polling is inefficient (you poll thousands of addresses on every block), and standing up custom indexing infrastructure is overkill. Turbo sits in between: you get a **fast, queryable index** that you point at a set of addresses you maintain, without running any indexing infrastructure yourself. This guide shows how to build that with a Turbo pipeline: Store the addresses you care about in a [dynamic table](/turbo-pipelines/transforms/dynamic-tables) and add/remove them at any time — no redeploy. Filter the transfer stream down to your addresses and land the matches in Postgres you can query directly. Fire a [webhook](/turbo-pipelines/sinks/webhook) the moment a watched address receives funds. ## How it works ```mermaid actions={false} theme={null} flowchart LR A[Transfers dataset
e.g. base.erc20_transfers] -->|token transfers| C{SQL filter
recipient ∈ watched set} B[(Dynamic table
your addresses)] --> C C -->|matches| D[(Postgres sink
queryable index)] C -->|matches| E[Webhook sink
real-time alert] C -->|matches| F[(Postgres aggregate
running balances)] ``` * The **source** is a curated transfer dataset (for example `base.erc20_transfers`) — a normalized, cross-chain stream of token transfers. * The **dynamic table** holds the addresses you're watching. It's backed by Postgres, so you can `INSERT`/`DELETE` addresses live and the pipeline picks up changes within a second or two — no redeploy. * A **SQL transform** keeps only transfers whose `recipient` is in your watched set, using [`dynamic_table_check()`](/turbo-pipelines/reference/sql-functions#dynamic-table-check). * **Sinks** deliver the result: a Postgres table you can query, a webhook for real-time alerts, or a running-balance table (all three can run in the same pipeline). ## Prerequisites * The [Turbo CLI extension](/turbo-pipelines/cli#installation) installed, and a Goldsky account logged in to your project. * A Postgres database for the dynamic table and sinks. You have two options: * **Goldsky-hosted Postgres** (recommended) — Goldsky provisions and manages it for you. Available on **Scale** plans and above. * **Bring your own Postgres** — Neon, Supabase, RDS, Cloud SQL, or self-hosted. ## Step 1: Provision Postgres The dynamic table and the sinks all connect to Postgres through a Goldsky [secret](/turbo-pipelines/pipeline-config#secrets). You can use one secret for everything in the pipeline. Provision a managed database and register its secret in one step: ```bash theme={null} goldsky hosted-sink create --type postgres --name MY_POSTGRES ``` The command prints the secret's **name**, **ID**, and **type**. Look up the raw connection string in the web app under **Sinks** when you need to connect a SQL client. You can also provision from the [web app](https://app.goldsky.com) under **Sinks → New sink → Hosted Postgres**. Hosted Postgres is a **Scale** plan feature (and above). Adding a credit card to your account upgrades you to Scale. See [pricing](/pricing/summary#hosted-databases). Create a writer role Goldsky can use: ```sql theme={null} CREATE ROLE goldsky_writer WITH LOGIN PASSWORD 'your_secure_password'; GRANT CREATE ON DATABASE your_database TO goldsky_writer; GRANT USAGE, CREATE ON SCHEMA public TO goldsky_writer; ``` Then store the connection string as a secret: ```bash theme={null} goldsky secret create MY_POSTGRES ``` When prompted, paste your connection string: ``` postgres://goldsky_writer:your_secure_password@db.example.com:5432/your_database?sslmode=require ``` If your database only accepts connections from allowlisted IPs, see [static IPs and `use_dedicated_ip`](/turbo-pipelines/transforms/dynamic-tables#postgresql-setup). Whichever path you choose, the rest of this guide refers to the secret as `MY_POSTGRES`. ## Step 2: Create the pipeline Create a file named `address-watcher.yaml`. This pipeline maintains a `watched_addresses` table, filters the Base ERC-20 transfer stream down to transfers **received** by one of those addresses, and writes the matches to a Postgres table you can query. ```yaml address-watcher.yaml theme={null} name: address-watcher resource_size: s sources: base_transfers: type: dataset dataset_name: base.erc20_transfers version: 1.2.0 start_at: latest transforms: # The set of addresses you're watching. Backed by Postgres so you can # add/remove addresses live without redeploying the pipeline. watched_addresses: type: dynamic_table backend_type: Postgres backend_entity_name: watched_addresses schema: public secret_name: MY_POSTGRES # Keep only transfers received by a watched address. incoming_transfers: type: sql primary_key: id sql: | SELECT id, address AS token, -- ERC-20 contract address sender, recipient, amount, block_number, to_timestamp(block_timestamp) AS block_time, transaction_hash, _gs_op FROM base_transfers WHERE dynamic_table_check('watched_addresses', recipient) sinks: # A queryable index of every transfer landing in a watched address. incoming_index: type: postgres from: incoming_transfers schema: public table: incoming_transfers secret_name: MY_POSTGRES primary_key: id ``` * **`sources.base_transfers`** — the curated `base.erc20_transfers` dataset. `start_at: latest` processes only new transfers going forward. Use `start_at: earliest` to backfill history first. * **`transforms.watched_addresses`** — the dynamic table. `schema: public` creates `public.watched_addresses` with a `value` (primary key) column and an `updated_at` timestamp. (If you omit `schema`, the dynamic table defaults to a `streamling` schema — setting it explicitly keeps everything in one place.) * **`transforms.incoming_transfers`** — filters the stream with `dynamic_table_check('watched_addresses', recipient)`, which returns `true` when `recipient` exists in the dynamic table. Change `recipient` to `sender` (or check both) to track outgoing transfers too. * **`sinks.incoming_index`** — writes matches to `public.incoming_transfers`. `primary_key: id` makes writes idempotent upserts. Store and match addresses in **lowercase**. Curated dataset addresses are already lowercased, and inserting your watched addresses with `lower(...)` (Step 4) keeps the match consistent. ## Step 3: Deploy Validate, then apply: ```bash theme={null} goldsky turbo validate address-watcher.yaml goldsky turbo apply address-watcher.yaml ``` The dynamic table starts empty, so nothing matches yet — that's expected. Watch data flow through the pipeline with live inspect: ```bash theme={null} goldsky turbo inspect address-watcher.yaml -n incoming_transfers ``` ## Step 4: Add addresses to watch The dynamic table is just a Postgres table. Connect any SQL client to the same database and insert the addresses you want to watch. Changes take effect within a second or two — no redeploy. ```sql theme={null} -- Watch a single address (example placeholder address) INSERT INTO public.watched_addresses (value) VALUES (lower('0x1111111111111111111111111111111111111111')); -- Watch several at once INSERT INTO public.watched_addresses (value) VALUES (lower('0x1111111111111111111111111111111111111111')), (lower('0x2222222222222222222222222222222222222222')), (lower('0x3333333333333333333333333333333333333333')) ON CONFLICT (value) DO NOTHING; ``` To stop watching an address, delete it: ```sql theme={null} DELETE FROM public.watched_addresses WHERE value = lower('0x1111111111111111111111111111111111111111'); ``` Inspect the current set at any time: ```sql theme={null} SELECT * FROM public.watched_addresses ORDER BY updated_at DESC; ``` ## Step 5: Query the index Once addresses are being watched and matching transfers arrive, query the Postgres table directly: ```sql theme={null} -- Most recent assets received across all watched addresses SELECT recipient, token, amount, block_time, transaction_hash FROM public.incoming_transfers ORDER BY block_time DESC LIMIT 20; -- Everything a specific address has received SELECT token, amount, block_time, transaction_hash FROM public.incoming_transfers WHERE recipient = lower('0x1111111111111111111111111111111111111111') ORDER BY block_time DESC; ``` ## Get a webhook when funds land To be notified the moment a watched address receives funds, add a [webhook sink](/turbo-pipelines/sinks/webhook). It POSTs each matching transfer to your endpoint. You can run it **alongside** the Postgres sink in the same pipeline. First, create an auth secret for your endpoint: ```bash theme={null} goldsky secret create # Choose "httpauth", then set a header (e.g. Authorization) and value (e.g. "Bearer ") ``` Then add the sink under `sinks:` in `address-watcher.yaml`: ```yaml theme={null} sinks: incoming_index: type: postgres from: incoming_transfers schema: public table: incoming_transfers secret_name: MY_POSTGRES primary_key: id # Fire one webhook request per transfer as soon as it's detected. arrival_alerts: type: webhook from: incoming_transfers url: https://api.example.com/asset-arrived secret_name: MY_WEBHOOK_AUTH one_row_per_request: true ``` Redeploy: ```bash theme={null} goldsky turbo apply address-watcher.yaml ``` Each request body is a single JSON object for the matched transfer: ```json theme={null} { "id": "...", "token": "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913", "sender": "0x...", "recipient": "0x1111111111111111111111111111111111111111", "amount": "1000000", "block_time": "2026-07-09T18:36:00Z", "transaction_hash": "0x..." } ``` The webhook sink guarantees **at-least-once** delivery and retries `5xx`/timeout responses with backoff. Your endpoint should be idempotent — dedupe on `id` (or `transaction_hash`) in case a batch is retried. ## Optional: track a running balance per address The queryable index above logs every transfer that landed. If you also want to answer "which of these addresses currently hold funds?", you can optionally add a **running balance per watched address** using the [`postgres_aggregate` sink](/turbo-pipelines/sinks/postgres-aggregate), which maintains a live sum in a database trigger. The example below is a self-contained pipeline you can adapt — add its transform and sink to `address-watcher.yaml`, or run it on its own. Take the parts you need. It tracks the net balance each watched address holds of a single token (USDC on Base). It counts credits (incoming) as positive and debits (outgoing) as negative, so the balance reflects deposits minus withdrawals. ```yaml balance-watcher.yaml theme={null} name: balance-watcher resource_size: s sources: base_transfers: type: dataset dataset_name: base.erc20_transfers version: 1.2.0 start_at: latest transforms: watched_addresses: type: dynamic_table backend_type: Postgres backend_entity_name: watched_addresses schema: public secret_name: MY_POSTGRES # Emit one signed balance-change row per watched address involved in a # USDC transfer: +amount when it receives, -amount when it sends. # `amount` is a 256-bit integer, so use the I256 helpers (a plain CAST # to DECIMAL is not supported on 256-bit columns). balance_changes: type: sql primary_key: entry_id sql: | SELECT concat(id, '-credit') AS entry_id, lower(recipient) AS account, to_i256(amount) AS delta FROM base_transfers -- 0x8335...2913 is the USDC token contract on Base WHERE lower(address) = lower('0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913') AND dynamic_table_check('watched_addresses', recipient) UNION ALL SELECT concat(id, '-debit') AS entry_id, lower(sender) AS account, i256_neg(to_i256(amount)) AS delta FROM base_transfers -- 0x8335...2913 is the USDC token contract on Base WHERE lower(address) = lower('0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913') AND dynamic_table_check('watched_addresses', sender) sinks: balances: type: postgres_aggregate from: balance_changes schema: public landing_table: usdc_balance_log agg_table: usdc_balances primary_key: entry_id secret_name: MY_POSTGRES group_by: account: type: text aggregate: balance: from: delta fn: sum type: numeric(78, 0) ``` Each transfer produces two distinct rows — a `-credit` entry for the recipient and a `-debit` entry for the sender — so `entry_id` (not the raw transfer `id`) is the primary key. Reusing `id` would make the two rows collide in the aggregate's landing table. Query which watched addresses currently hold a balance: ```sql theme={null} SELECT account, balance FROM public.usdc_balances WHERE balance > 0 ORDER BY balance DESC; ``` Balances here are in the token's **base units** (USDC has 6 decimals, so `1000000` = 1 USDC). Divide by `10^decimals` when displaying. The `sum` aggregation supports insert/delete streams but **not** updates — see the [`postgres_aggregate` caveats](/turbo-pipelines/sinks/postgres-aggregate#supported-aggregation-functions). ## Coverage and limitations Add more sources (`ethereum.erc20_transfers`, `arbitrum.erc20_transfers`, and so on) and `UNION ALL` them in the SQL transform, tagging each with a `chain` column. See [multi-chain monitoring](/turbo-pipelines/sources/evm#guide-multi-chain-monitoring). The same `watched_addresses` table can back all chains. Some chains expose dedicated balance datasets (for example `solana.native_balances`, `stellar_mainnet.balances`). For EVM chains, derive balances from the transfer stream as shown above, or query a single address on demand via [Edge RPC `eth_getBalance`](/edge-rpc/evm/methods/eth_getBalance). ## Related * [Dynamic Tables](/turbo-pipelines/transforms/dynamic-tables) — full reference for the watched-address table * [EVM Sources](/turbo-pipelines/sources/evm) — available datasets and fields * [PostgreSQL sink](/turbo-pipelines/sinks/postgres) and [PostgreSQL aggregation sink](/turbo-pipelines/sinks/postgres-aggregate) * [Webhook sink](/turbo-pipelines/sinks/webhook) Can't find what you're looking for? Reach out to us at [support@goldsky.com](mailto:support@goldsky.com) for help.