# Compose Dividend Distribution

> Build and deploy the Goldsky Compose corporate-actions / dividend-distribution example under the user's own account — a durable, idempotent distributor that pays N token holders pro-rata for a tokenized corporate action (dividend, coupon, rebate, airdrop) with an on-chain audit trail. The interesting bit: Compose orchestrates a Goldsky Turbo job-mode pipeline as an ephemeral subroutine — declaring a campaign spawns a one-shot pipeline that snapshots share-token holders at a record block, waits for it, pays each holder via a gas-sponsored wallet, then deletes the pipeline. Triggers on: 'dividend distribution', 'pay dividends onchain', 'distribute dividends to shareholders', 'corporate actions distributor', 'pro-rata payout to token holders', 'airdrop pro-rata by balance', 'cap table distribution', 'set up / deploy the dividend / corporate-actions example'. Ships pointed at shared permissionless demo contracts on Base Sepolia so there's nothing to deploy. Deploys fully in-app or via the CLI; the only runtime re

- Skill: `goldsky-io/compose-dividend-distribution` (Agent Skill)
- Install (CLI): `npx skillmds@latest add goldsky-io/compose-dividend-distribution`
- Raw SKILL.md: https://api.skillmd.com/api/skills/goldsky-io/compose-dividend-distribution/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Marketing & Growth
- Author: goldsky-io (https://skillmd.com/u/goldsky-io)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/goldsky-io/compose-dividend-distribution

---


# Build: Compose dividend distribution (corporate-actions)

Stand up the corporate-actions distributor under the user's own Goldsky account. It pays N holders pro-rata for a tokenized corporate action — dividend, coupon, rebate, airdrop — idempotently and durably, with a tamper-evident on-chain audit trail. The interesting bit: **Compose orchestrates Goldsky Turbo as an ephemeral, on-demand subroutine.** Declaring a campaign spawns a one-shot [job-mode](https://docs.goldsky.com/turbo-pipelines/job-mode) Turbo pipeline that snapshots share-token holders at the operator-supplied record block; Compose waits for it to finish, pays each holder via a gas-sponsored wallet, then deletes the pipeline. No always-on indexing.

One HTTP task (`declare_campaign`) drives the whole lifecycle: declare → escrow USDC → spawn snapshot pipeline → poll → compute pro-rata → pay up to 25 holders concurrently → verify `escrowRemaining == 0` → delete the pipeline. Re-POSTing the same `campaignId` resumes cleanly after any failure; the contract is the sole source of truth for "did this holder get paid?", so double-pays are structurally impossible.

This template supplies only what's specific to the dividend/corporate-actions app — how it works and its source. The recommended path uses **shared, permissionless demo contracts on Base Sepolia** (open `mint` on MockUSDC, open `declare()` on the campaign), so there's nothing to deploy.

## Step 0a — Load the base skills first

**Before anything else — before you answer, ask a question, scaffold a file, or run any command — load the two base skills this template depends on:**

1. **`Skill(compose)`** — the always-on Compose guide: the golden rules (never assume anything about the app on the user's behalf; ask when unsure) and general build guidance.
2. **`Skill(compose-reference)`** — the manifest / field / API reference; consult before writing any `compose.yaml` or task file.

This template deliberately omits those rules and that reference — they are **required** to build correctly and are not repeated here. Do not proceed until both are loaded.

## Mode Detection

Pick the mode from the tools available to you:

- **A `deployComposeApp` tool is available (Goldsky webapp chatbot).** This example deploys fully in-app. The job-mode Turbo pipeline is spawned at runtime via the Turbo API (an in-app `ctx.fetch` POST in `src/lib/turbo.ts`), not provisioned at deploy, so `deployComposeApp` deploys the app fine. In-app flow: run the Step 0b app-name interview first (the app name is the FIRST question), then scaffold these files in-memory from **The app (full source)** below and pass them to `deployComposeApp`: `compose.yaml`, `src/tasks/declare-campaign.ts`, `src/lib/constants.ts`, `src/lib/types.ts`, `src/lib/math.ts`, `src/lib/normalize.ts`, `src/lib/db.ts`, `src/lib/driver.ts`, and `src/lib/turbo.ts`. On the recommended shared-contract path there is nothing to deploy, so exclude the three `contracts/*.sol` sources (the shared path deploys nothing); on the deploy-your-own path also scaffold the `.sol` files and use `deployContract` for MockUSDC, ShareToken (passing the holder/amount arrays), and DistributionCampaign. Wire the four contract CONFIG values in `src/lib/constants.ts` (the three deployed addresses plus `shareTokenDeployBlock`), then call `deployComposeApp`. The `GOLDSKY_PROJECT_KEY` secret is the user's LAST step: the in-app deploy skips secret validation, so `deployComposeApp` succeeds without it, but the app won't run until the user adds the secret in the Compose app's dashboard **and redeploys from the dashboard** so the pod picks it up (secrets are baked into the pod at deploy, not hot-reloaded). NEVER attempt to set a secret from chat; there is no tool, by design. Note: the chatbot has no `writeContract` tool, so it cannot mint test MockUSDC to the `corp-actions-operator` wallet; before declaring a campaign the user must run the Step 4 mint via the CLI (`goldsky compose writeContract`) or the dashboard (the task does the USDC `approve` itself at declare time, but the operator wallet must hold the USDC first).
- **`Bash` is available (local CLI / coding agent):** execute the steps below directly, parse output, and substitute captured values into later commands.
- **Neither (pure reference Q&A):** explain what the app does and the lifecycle; only if asked for step-by-step help, output one command at a time and have the user paste output back. Point them at `npx skills add goldsky-io/goldsky-agent` to run it locally with Bash.

## Non-negotiables

- **Ships pointed at shared, permissionless demo contracts — nothing to deploy.** MockUSDC has an open `mint` and DistributionCampaign has an open `declare()`, so anyone can run a campaign on them. The shared demos exist on Base Sepolia by default; `src/lib/constants.ts` also lists known Base mainnet deployments you can swap to (real gas applies on mainnet). Tell the user, in prose, these are demos/getting-started only, not production.
- **One project API key does the whole job.** It's used three ways: the `GOLDSKY_API_TOKEN` env var (preferred, keeps the key out of argv and shell history) on `goldsky compose deployContract` / `deploy` / `writeContract`, the value of the `GOLDSKY_PROJECT_KEY` secret (so the running app can spawn / poll / delete Turbo pipelines), **and** as the `$GOLDSKY_TOKEN` bearer for the Step 5 HTTP task (or a separate Compose API token minted from the same project). Generate it in the Goldsky dashboard under **Settings > API Keys**. The app won't run without the secret.
- **`recordBlock` must be `<= currentBlock`** and should be past finality (e.g. `currentBlock - 32`). The snapshot is backwards-looking — it's the cutoff for who gets paid. Future-dated record blocks are out of scope.
- **Never run `goldsky compose deployContract` (deploy-your-own), `goldsky compose deploy`, `goldsky compose secret set`, `git push`, or `gh repo create` without showing the exact command first and getting explicit confirmation.**
- **Resumable by design — never worry about double-pay.** Re-POSTing the same `campaignId` drives the existing campaign forward. A per-holder on-chain `isPaid()` check plus the contract's `require(!paid[id][holder])` guard mean Compose can crash/restart at any point with zero risk of double-paying.
- **This example does not run in a local/dev Compose cluster without Turbo pipeline infra.** It deploys against real Goldsky (app.goldsky.com), which is where a user runs it anyway.

## The app (full source)

This is the complete dividend app. Scaffold these files verbatim (Step 0b writes them to disk via `degit`; the in-app flow scaffolds them in-memory from the blocks below). The shared Base Sepolia demo contracts in `src/lib/constants.ts` mean the recommended path has nothing to deploy. Only edit `src/lib/constants.ts` to wire in your own contract addresses (Step 1 Branch B) or swap to Base mainnet.

### `compose.yaml`

```yaml
name: "corporate-actions"
api_version: "stable"

# POSTGRES_CONNECTION_STRING is auto-injected at deploy time by compose-cloud.
# A Goldsky-project secret named CORPORATE_ACTIONS is created alongside it,
# referencing the same Neon DB. The job-mode Turbo pipelines that
# declare_campaign spawns (see src/lib/turbo.ts) write share-balance
# snapshots back into that DB.
secrets:
  # Project API key used to spawn / poll / delete Turbo pipelines from
  # inside declare_campaign. Set once:
  #   goldsky compose secret set GOLDSKY_PROJECT_KEY --value "$GOLDSKY_PROJECT_KEY"
  - GOLDSKY_PROJECT_KEY

tasks:
  - path: "./src/tasks/declare-campaign.ts"
    name: "declare_campaign"
    triggers:
      - type: "http"
        authentication: "auth_token"
    retry_config:
      max_attempts: 1
      initial_interval_ms: 500
      backoff_factor: 1
```

### `src/tasks/declare-campaign.ts`

```typescript
import type { TaskContext } from "compose";
import { encodePacked, keccak256 } from "viem";

import { CONFIG } from "../lib/constants";
import { driveCampaign } from "../lib/driver";
import { isHexBytes32 } from "../lib/normalize";
import { createSnapshotPipeline } from "../lib/turbo";
import type { Campaign, DeclareParams, Hex } from "../lib/types";

/**
 * HTTP trigger.
 *
 *   POST {
 *     "campaignId":  "0x<32 bytes hex>",  // operator-supplied id, unique per operator
 *     "recordBlock": 24500000,            // snapshot point; must be <= chain head
 *     "totalAmount": "10000000000"        // 10,000 mUSDC (6 decimals)
 *   }
 *
 *   1. Validate `recordBlock` is in the past (or current). Future-dated record
 *      blocks aren't supported in this demo — they're a real corp-action feature
 *      (record dates often look forward) but out of scope here.
 *   2. Approve the campaign contract for `totalAmount` of MockUSDC.
 *   3. Call `DistributionCampaign.declare(...)` — pulls escrow atomically.
 *   4. Spawn a job-mode Turbo pipeline to snapshot holders of `shareToken`
 *      from the share-token deploy block up to `recordBlock`. Per-campaign
 *      sink tables avoid cross-campaign aggregate contamination.
 *   5. Drive the campaign through snapshot → paying → complete inline,
 *      polling the pipeline at STATE_POLL_INTERVAL_MS until done. The
 *      whole lifecycle finishes in this single HTTP request.
 *
 * Idempotent on `campaignId`: a second POST with the same id resumes the
 * existing campaign (drives it forward if non-terminal) instead of
 * re-declaring.
 */
export async function main(context: TaskContext, params?: DeclareParams) {
  const { evm, collection } = context;
  if (!params) throw new Error("POST body required");

  const userId = params.campaignId;
  if (!isHexBytes32(userId)) {
    throw new Error("campaignId must be a 0x-prefixed 32-byte hex string");
  }
  if (typeof params.recordBlock !== "number" || params.recordBlock <= 0) {
    throw new Error("recordBlock must be a positive integer");
  }
  const totalAmount = BigInt(params.totalAmount);
  if (totalAmount <= 0n) throw new Error("totalAmount must be positive");

  const campaigns = await collection<Campaign>("campaigns", [
    { path: "status", type: "text" },
  ]);

  const rowId = userId.toLowerCase();
  const existing = await campaigns.getById(rowId);
  if (existing) {
    // Resume an in-flight campaign — keep driving it forward. Terminal
    // states (complete/failed) just return without doing anything.
    await driveCampaign(context, campaigns, existing);
    const fresh = (await campaigns.getById(rowId)) ?? existing;
    return responseFor(fresh, "resumed");
  }

  // --- recordBlock <= currentBlock ---
  // Resolved against the chain's public RPC via context.fetch (only fetch
  // path that's --allow-net'd in this child process).
  const chain = evm.chains[CONFIG.chain];
  const currentBlock = await getCurrentBlock(context, chain.rpcUrls.default.http[0]);
  const recordBlock = BigInt(params.recordBlock);
  if (recordBlock > currentBlock) {
    throw new Error(
      `recordBlock ${recordBlock} > currentBlock ${currentBlock}; ` +
        `future-dated record blocks are out of scope for this demo`,
    );
  }

  const wallet = await evm.wallet({
    name: "corp-actions-operator",
    sponsorGas: true,
  });

  // --- approve + declare on-chain ---
  await wallet.writeContract(
    chain,
    CONFIG.payToken,
    "approve(address,uint256)",
    [CONFIG.campaignContract, totalAmount.toString()],
  );

  const { hash } = await wallet.writeContract(
    chain,
    CONFIG.campaignContract,
    "declare(bytes32,address,address,uint256)",
    [userId, CONFIG.payToken, CONFIG.shareToken, totalAmount.toString()],
  );

  // canonicalId matches the contract's keccak256(operator, userId).
  const onChainId = keccak256(
    encodePacked(["address", "bytes32"], [wallet.address, userId as Hex]),
  );

  // --- spawn the snapshot pipeline ---
  // If this fails AFTER declare(), the operator can recover escrow with
  // DistributionCampaign.seal(). The campaign row is not written, so
  // there's nothing to drive forward.
  const pipeline = await createSnapshotPipeline(context, {
    campaignId: userId,
    shareToken: CONFIG.shareToken,
    recordBlock,
  });

  const campaign: Campaign = {
    rowId,
    userId: rowId as Hex,
    onChainId,
    shareToken: CONFIG.shareToken,
    payToken: CONFIG.payToken,
    totalAmount: totalAmount.toString(),
    recordBlock: recordBlock.toString(),
    declareTxHash: hash as Hex,
    pipelineName: pipeline.name,
    status: "snapshotting",
    createdAt: Date.now(),
  };
  await campaigns.setById(rowId, campaign);

  // Drive the campaign through snapshot → paying → complete inline. If
  // anything throws, the partial state is preserved and the operator can
  // re-POST the same campaignId to resume. Re-throw to surface failure.
  await driveCampaign(context, campaigns, campaign);
  const final = (await campaigns.getById(rowId)) ?? campaign;
  return responseFor(final, "declared");
}

function responseFor(c: Campaign, source: "declared" | "resumed") {
  return {
    status: c.status,
    source,
    userId: c.userId,
    onChainId: c.onChainId,
    pipelineName: c.pipelineName,
    declareTxHash: c.declareTxHash,
    failureReason: c.status === "failed" ? c.failureReason : undefined,
  };
}

async function getCurrentBlock(
  ctx: TaskContext,
  rpcUrl: string,
): Promise<bigint> {
  const res = await ctx.fetch<{ result?: string; error?: { message: string } }>(
    rpcUrl,
    {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: { jsonrpc: "2.0", id: 1, method: "eth_blockNumber", params: [] },
    },
  );
  if (res?.error) throw new Error(`eth_blockNumber: ${res.error.message}`);
  if (!res?.result) throw new Error("eth_blockNumber returned no result");
  return BigInt(res.result);
}
```

### `src/lib/constants.ts`

```typescript
import type { Hex } from "./types";

/**
 * Single-chain demo on Base Sepolia. Each declaration spawns its own
 * job-mode Turbo pipeline, so there's nothing chain-specific to configure
 * beyond the deployed contract addresses below.
 *
 * Defaults to Base Sepolia so the demo costs no real gas and uses the shared
 * permissionless contracts below (open mint on MockUSDC, open declare() on the
 * campaign). To run on Base mainnet instead, swap in these values (real gas
 * applies):
 *   chain: "base", turboChain: "base",
 *   shareToken:        "0xE05Ceb3E269029E3bab46E35515e8987060D1027",
 *   payToken (MockUSDC): "0x02D9Df62B7AED15739D638B92BAcEA2ce4Cb3d70",
 *   campaignContract:  "0x81051f77ea167b631Dd7F40ac414A9F9344Fb162",
 *   shareTokenDeployBlock: 45654954,
 *
 * Update after running `scripts/deploy.sh`.
 */
export const CONFIG = {
  chain:      "baseSepolia" as const,   // evm.chains[chain] key (camelCase)
  turboChain: "base_sepolia",           // Turbo dataset prefix (snake_case network slug)
  shareToken:       "0x713e0749a9Fe480322990913850e81b0F4F4dc0d" as Hex,
  payToken:         "0x8ec24F07F08745fc3D979336AA81d4Dc73f3D9DE" as Hex,  // MockUSDC (permissionless mint)
  campaignContract: "0xA8e58573B1e10908b63d12B603aCF9C784BF904E" as Hex,  // permissionless: anyone can declare()
  // Block at which `shareToken` was deployed. Job-mode forces
  // `start_at: earliest`, so we can't anchor the source there directly;
  // instead this is used as the lower bound in the snapshot pipeline's
  // SQL filter (`block_number BETWEEN <deploy> AND <record>`), which lets
  // the planner prune all pre-deploy blocks before scanning. Per Jeff: a
  // filter-level block range is meaningfully faster than a source-level
  // `end_block` alone.
  shareTokenDeployBlock: 42275958,
};

/**
 * Concurrent pay() calls. Bounded by the gas-sponsored bundler's throughput
 * (~1-5 userOps/sec/sender). Set high enough that the demo's full
 * 25-holder snapshot fires in a single batch.
 */
export const CONCURRENCY = 25;

/**
 * State-poll cadence while waiting for the Turbo job-mode snapshot to
 * finish. With Jeff's filter-level block range the snapshot finishes in
 * ~5-10s, so we poll fast (2s) so `declare_campaign` can drive the campaign
 * end-to-end inline before returning.
 */
export const STATE_POLL_INTERVAL_MS = 2_000;

/**
 * Hard cap on snapshot-poll iterations per drive call. Set high so we wait
 * out the snapshot in-line for any realistic case; pathological hangs still
 * eventually fall through to the cron path.
 */
export const MAX_POLLS_PER_TICK = 100;  // 100 × 2s = ~3.3 minutes

/**
 * The Turbo pipeline writes into per-campaign tables to avoid cross-campaign
 * SUM contamination in the `postgres_aggregate` sink.
 *
 *   share_balances_<id>      — agg table (account, balance)
 *   share_transfer_log_<id>  — landing table (truncated per checkpoint)
 *
 * `id` is a 16-char slice of campaignId — stable, unique, fits in Postgres'
 * 63-char identifier limit.
 */
export function pipelineId(campaignId: string): string {
  return campaignId.toLowerCase().replace(/^0x/, "").slice(0, 16);
}

export function pipelineName(campaignId: string): string {
  return `corp-actions-${pipelineId(campaignId)}`;
}

export function aggTableName(campaignId: string): string {
  return `share_balances_${pipelineId(campaignId)}`;
}
```

### `src/lib/types.ts`

```typescript
export type Hex = `0x${string}`;

/**
 * Campaign lifecycle:
 *
 *   snapshotting → paying → complete
 *                  ↘ failed
 *
 *   - snapshotting: a job-mode Turbo pipeline is running, indexing Transfer
 *     events of `shareToken` from chain genesis up to `recordBlock`.
 *   - paying: the pipeline has emitted the snapshot to Postgres; the cron is
 *     pro-rata paying out per holder.
 *   - complete: every holder is paid on-chain. Pipeline has been deleted.
 *   - failed: the pipeline errored. Pipeline has been deleted; the campaign
 *     row stays around for postmortem (escrow can be recovered via
 *     `DistributionCampaign.seal()`).
 */
export type CampaignStatus = "snapshotting" | "paying" | "complete" | "failed";

export interface DeclareParams {
  campaignId: string;   // bytes32 hex string — operator-supplied id
  recordBlock: number;  // snapshot point; must be <= chain head at declare time
  totalAmount: string;  // bigint as string (USDC has 6 decimals)
}

export interface Campaign {
  rowId: string;            // = userId (lowercased) — collection unique key
  userId: Hex;              // operator-supplied campaignId, lowercased
  onChainId: Hex;           // keccak256(operator, userId)
  shareToken: Hex;          // resolved server-side from constants
  payToken: Hex;            // resolved server-side (MockUSDC)
  totalAmount: string;
  recordBlock: string;      // snapshot block, recorded both on-chain and here
  declareTxHash: Hex;
  pipelineName: string;     // unique per campaign; used for /state polls and DELETE
  status: CampaignStatus;
  createdAt: number;
  snapshotCompletedAt?: number;
  completedAt?: number;
  failedAt?: number;
  failureReason?: string;
  // Persisted payouts so the holder/amount table survives terminalCleanup
  // (which drops the per-campaign Postgres tables). Populated when the
  // driver transitions to "paying" and the pro-rata is computed; bigints
  // serialised as decimal strings so the row round-trips through JSON.
  payouts?: PersistedPayout[];
}

export interface PersistedPayout {
  holder: Hex;
  sharesAtSnapshot: string;  // bigint as text
  amount: string;            // bigint as text (USDC, 6 decimals)
  payTxHash?: Hex;           // captured per-batch in drivePayouts
}

export interface Holder {
  address: Hex;
  balance: bigint;
}

export interface Payout {
  holder: Hex;
  amount: bigint;
  sharesAtSnapshot: bigint;
}
```

### `src/lib/math.ts`

```typescript
import type { Holder, Payout } from "./types";
import { normalizeAddr } from "./normalize";

/**
 * Pro-rata payout calculator.
 *
 * Integer division floors each holder's share, leaving a remainder. To make the
 * sum equal `totalAmount` exactly, the remainder is added to the LAST holder's
 * payout. This rounding direction is documented and stable: holders are sorted
 * by address ascending so "last" is deterministic across runs.
 *
 * @param holders     non-empty list of holders with bigint balances
 * @param totalAmount total escrow to distribute (bigint)
 * @param totalSupply sum of all holder balances (bigint)
 */
export function proRata(
  holders: Holder[],
  totalAmount: bigint,
  totalSupply: bigint,
): Payout[] {
  if (holders.length === 0) return [];
  if (totalSupply === 0n) {
    throw new Error("totalSupply must be positive");
  }

  // Sort by address ascending so "last holder" is deterministic.
  const sorted = [...holders].sort((a, b) =>
    a.address.toLowerCase() < b.address.toLowerCase() ? -1 : 1,
  );

  const payouts: Payout[] = [];
  let allocated = 0n;
  for (let i = 0; i < sorted.length; i++) {
    const h = sorted[i];
    const isLast = i === sorted.length - 1;
    const amount = isLast
      ? totalAmount - allocated // last holder absorbs floor remainder
      : (h.balance * totalAmount) / totalSupply;
    payouts.push({
      holder: normalizeAddr(h.address),
      amount,
      sharesAtSnapshot: h.balance,
    });
    allocated += amount;
  }
  return payouts;
}
```

### `src/lib/normalize.ts`

```typescript
import type { Hex } from "./types";

const BYTES32_RE = /^0x[a-fA-F0-9]{64}$/;
const ADDR_RE = /^0x[a-fA-F0-9]{40}$/;

/**
 * Normalize an EVM address to lowercase 0x-hex.
 * Apply at every boundary: Postgres reads, collection keys, contract args.
 */
export function normalizeAddr(s: string): Hex {
  if (!ADDR_RE.test(s)) {
    throw new Error(`invalid address: ${s}`);
  }
  return s.toLowerCase() as Hex;
}

export function isHexBytes32(s: string): boolean {
  return BYTES32_RE.test(s);
}
```

### `src/lib/db.ts`

```typescript
import type { TaskContext } from "compose";
import type { Hex, Holder } from "./types";

/**
 * Query Neon's HTTP `/sql` endpoint via compose's IPC-routed `context.fetch`.
 *
 * Why not the `@neondatabase/serverless` driver?
 *   The compose-task child process is compiled WITHOUT `--allow-net`. Raw TCP
 *   AND `globalThis.fetch` from the task code both error with `EPERM`. The ONLY
 *   permitted egress path is `context.fetch`, which IPCs into the host process
 *   (which has `--allow-net`).
 *
 * `POSTGRES_CONNECTION_STRING` is auto-injected by compose-cloud. The Turbo
 * job-mode pipelines that this app spawns write into the same Neon DB via
 * the auto-created `CORPORATE_ACTIONS` project secret.
 */

interface NeonRow {
  [key: string]: string | number | boolean | null;
}
interface NeonResponse {
  rows?: NeonRow[];
}

function getConnectionString(): string {
  const url = Deno.env.get("POSTGRES_CONNECTION_STRING");
  if (!url) {
    throw new Error("POSTGRES_CONNECTION_STRING not set");
  }
  // Tack on params that bust Neon pool stickiness:
  //   - target_session_attrs=read-write → routes to primary, not a replica
  //   - application_name=<unique>      → defeats pool slot stickiness so each
  //                                      HTTP query gets a freshly-spawned
  //                                      backend connection (which sees
  //                                      committed writes, not a stale snapshot)
  // We've measured ~145s read lag without these; with them the read should
  // see writes within seconds.
  const u = new URL(url);
  u.searchParams.set("target_session_attrs", "read-write");
  u.searchParams.set(
    "application_name",
    `corp-actions-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
  );
  return u.toString();
}

/** Derive Neon's HTTP `/sql` URL from a Postgres connection string. */
function neonHttpUrl(connStr: string): string {
  const u = new URL(connStr);
  // Use hostname (not host) so we drop the :5432 postgres port; HTTPS goes
  // to 443. Replace the first dotted segment with "api." per the official
  // @neondatabase/serverless transformation.
  const apiHost = u.hostname.replace(/^[^.]+\./, "api.");
  return `https://${apiHost}/sql`;
}

export async function neonQuery(
  ctx: TaskContext,
  query: string,
  params: unknown[] = [],
): Promise<NeonRow[]> {
  const connStr = getConnectionString();
  const url = neonHttpUrl(connStr);
  const res = await ctx.fetch<NeonResponse>(url, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "Neon-Connection-String": connStr,
      "Neon-Raw-Text-Output": "true",
      "Neon-Array-Mode": "false",
    },
    body: { query, params },
  });
  return res?.rows ?? [];
}

/**
 * Compute holder balances for a campaign's snapshot.
 *
 * Each campaign's pipeline writes raw Transfer rows to its own
 * `share_balances_<id>` table (no in-pipeline aggregation — see the rant in
 * `lib/turbo.ts` about FixedSizeBinary handling). The aggregate runs here
 * as a Postgres SQL: every transfer credits the recipient and debits the
 * sender (skipping the zero-address sender for mints), summed per account.
 *
 * Postgres handles binary→numeric coercion at write time — `amount` lands
 * as `numeric(78,0)` which we can cast to text and parse as a JS bigint
 * without precision loss.
 */
export async function getHolders(
  ctx: TaskContext,
  table: string,
): Promise<Holder[]> {
  // `table` is a constructed identifier from pipelineId() — alphanumeric +
  // underscore only — so direct interpolation is safe. (Postgres prepared
  // statements don't support parameterising table names anyway.)
  const rows = await neonQuery(
    ctx,
    `SELECT account, SUM(delta)::text AS balance
       FROM (
         SELECT lower(recipient) AS account, amount AS delta
           FROM "${table}"
         UNION ALL
         SELECT lower(sender) AS account, -amount AS delta
           FROM "${table}"
          WHERE lower(sender) != '0x0000000000000000000000000000000000000000'
       ) ledger
      GROUP BY account
     HAVING SUM(delta) > 0
      ORDER BY account ASC`,
  );
  return rows.map((r) => ({
    address: String(r.account).toLowerCase() as Hex,
    balance: BigInt(String(r.balance)),
  }));
}

/**
 * Number of Transfer rows in the per-campaign table, or `null` if the table
 * doesn't exist yet.
 *
 * Why count rather than just check existence?
 *   The Postgres sink creates the table on the very first checkpoint —
 *   even an "empty epoch" with zero matching rows commits, which creates
 *   the schema. So `aggTableExists` can return `true` while the pipeline
 *   is still mid-scan and hasn't reached blocks where the share token's
 *   Transfers live. Counting rows distinguishes "sink initialized" from
 *   "sink has actually delivered data".
 */
export async function aggTableRowCount(
  ctx: TaskContext,
  aggTable: string,
): Promise<number | null> {
  // Beefy diagnostic version: schema-qualified count, planner-stat count,
  // physical table size, schema list, and connection identity. Designed
  // to triangulate a Neon read-after-write visibility lag we've been
  // seeing where count(*) returns 0 for ~2 minutes after the pipeline
  // commits ~10 rows.
  try {
    const diag = await neonQuery(
      ctx,
      `SELECT
         (SELECT count(*)::text FROM public."${aggTable}") AS n_public,
         (SELECT pg_table_size('public."${aggTable}"')::text) AS bytes,
         pg_is_in_recovery()::text AS in_recovery,
         pg_last_xact_replay_timestamp()::text AS last_replay,
         now()::text AS now_ts,
         (SELECT EXTRACT(epoch FROM (now() - pg_last_xact_replay_timestamp()))::text) AS replay_lag_s,
         pg_backend_pid()::text AS pid,
         (SELECT setting FROM pg_settings WHERE name = 'application_name') AS app_name`,
    );
    const r = diag[0] ?? {};
    const n = Number(r.n_public ?? 0);
    console.log(
      `[db] "${aggTable}" n=${r.n_public} bytes=${r.bytes} ` +
        `recovery=${r.in_recovery} replay_lag_s=${r.replay_lag_s} ` +
        `last_replay=${r.last_replay} now=${r.now_ts} ` +
        `pid=${r.pid} app=${r.app_name}`,
    );
    return n;
  } catch (err) {
    const msg = err instanceof Error ? err.message : String(err);
    if (/does not exist|undefined.relation|relation .* does not exist/i.test(msg)) {
      console.log(`[db] "${aggTable}" → table missing`);
      return null;
    }
    console.log(`[db] "${aggTable}" threw: ${msg}`);
    throw err;
  }
}

/**
 * Drop the per-campaign table. Called on terminal cleanup so the user's
 * Neon DB doesn't accumulate orphaned tables across many campaigns.
 *
 * MUST be called AFTER the pipeline has been DELETE-ed — Turbo's sink writer
 * holds a connection, and dropping while it's still active is racing.
 */
export async function dropCampaignTables(
  ctx: TaskContext,
  transfersTable: string,
): Promise<void> {
  await neonQuery(ctx, `DROP TABLE IF EXISTS "${transfersTable}"`);
}
```

### `src/lib/driver.ts`

```typescript
import type { TaskContext } from "compose";

import {
  aggTableName,
  CONCURRENCY,
  CONFIG,
  MAX_POLLS_PER_TICK,
  STATE_POLL_INTERVAL_MS,
} from "./constants";
import { aggTableRowCount, dropCampaignTables, getHolders } from "./db";
import { proRata } from "./math";
import { deletePipeline, getPipelineState } from "./turbo";
import type { Campaign, Payout, PersistedPayout } from "./types";

/**
 * Drive a single campaign through the snapshot → paying → complete state
 * machine. Called inline by declare_campaign; can be called repeatedly to
 * resume a stuck campaign.
 *
 *   - status="snapshotting": poll the campaign's job-mode pipeline state at
 *     STATE_POLL_INTERVAL_MS up to MAX_POLLS_PER_TICK iterations until it
 *     transitions to `completed` (or its k8s deployment auto-cleans up
 *     after a successful run, which we infer from `unknown` + agg table
 *     having rows). On `error` → flip to "failed", drop the per-campaign
 *     tables, delete the pipeline.
 *
 *   - status="paying": read the snapshot from the per-campaign agg table,
 *     compute pro-rata, pay each holder via DistributionCampaign.pay() with
 *     bounded concurrency. The contract's `paid[id][holder]` mapping is the
 *     sole source of truth for "did this holder get paid?" — re-read on
 *     every drive call, so a pod kill mid-batch is recovered cleanly.
 *     When `escrowRemaining == 0` → mark complete, delete pipeline,
 *     drop tables.
 *
 *   - status="complete" or "failed": no-op. Terminal.
 */
export async function driveCampaign(
  context: TaskContext,
  campaigns: Awaited<ReturnType<TaskContext["collection"]>>,
  campaign: Campaign,
) {
  if (campaign.status === "snapshotting") {
    await driveSnapshot(context, campaigns, campaign);
    return;
  }
  if (campaign.status === "paying") {
    await drivePayouts(context, campaigns, campaign);
    return;
  }
}

async function driveSnapshot(
  context: TaskContext,
  campaigns: Awaited<ReturnType<TaskContext["collection"]>>,
  campaign: Campaign,
) {
  const aggTable = aggTableName(campaign.userId);

  for (let i = 0; i < MAX_POLLS_PER_TICK; i++) {
    const state = await getPipelineState(context, campaign.pipelineName);

    if (state === "error") {
      await markFailed(
        context,
        campaigns,
        campaign,
        "pipeline entered error state",
      );
      return;
    }

    // The Postgres sink commits the table on its FIRST checkpoint — even
    // an empty epoch creates the schema. So we have to count rows, not
    // just check the table exists, or a brief `/state` 404 mid-scan can
    // race the driver into transitioning to `paying` while the pipeline
    // is still scanning ahead of the share token's deploy block.
    //
    // Two paths to "snapshot ready":
    //   1. Pipeline reports completed/paused/stopped AND the table has
    //      ≥1 row.
    //   2. Pipeline state is `unknown` (404 from the auto-cleanup path
    //      that follows a successful job-mode run) AND the table has
    //      ≥1 row.
    //
    // If state is terminal but the table is empty, that's a structural
    // failure (or a token with no holders, which for a corp-action is
    // also operationally a failure).
    const rowCount = await aggTableRowCount(context, aggTable);
    console.log(
      `[${campaign.userId}] poll i=${i} state=${state} rowCount=${rowCount}`,
    );
    const sawTerminalState = state === "completed";

    if (sawTerminalState && (rowCount === null || rowCount === 0)) {
      await markFailed(
        context,
        campaigns,
        campaign,
        `pipeline completed but ${aggTable} has no rows ` +
          `(token may have no transfers, or pipeline failed silently)`,
      );
      return;
    }

    const haveRows = rowCount !== null && rowCount > 0;
    const looksAutoCleaned = state === "unknown" && haveRows;

    if ((sawTerminalState && haveRows) || looksAutoCleaned) {
      console.log(`[${campaign.userId}] snapshot completed → paying`);
      const updated: Campaign = {
        ...campaign,
        status: "paying",
        snapshotCompletedAt: Date.now(),
      };
      await campaigns.setById(campaign.rowId, updated);
      // Don't re-read from the collection here — same Neon pool-stickiness
      // bug we hit on user tables means setById's write may not be visible
      // to an immediate getById on the same connection. We have the new
      // value in-memory; pass it through directly.
      await drivePayouts(context, campaigns, updated);
      return;
    }

    // running / starting / unknown-without-rows → keep waiting
    if (i < MAX_POLLS_PER_TICK - 1) {
      await sleep(STATE_POLL_INTERVAL_MS);
    }
  }
  console.log(
    `[${campaign.userId}] snapshot still in-flight after ${MAX_POLLS_PER_TICK} polls; ` +
      `re-call declare_campaign with the same id to resume`,
  );
}

async function drivePayouts(
  context: TaskContext,
  campaigns: Awaited<ReturnType<TaskContext["collection"]>>,
  campaign: Campaign,
) {
  console.log(`[${campaign.userId}] drivePayouts: start`);
  const aggTable = aggTableName(campaign.userId);
  const holders = await getHolders(context, aggTable);
  console.log(`[${campaign.userId}] drivePayouts: holders=${holders.length}`);
  if (holders.length === 0) {
    // The snapshot completed (the agg table exists) but contains zero rows.
    // For a corporate-action distribution this is always a failure — either
    // the pipeline pod silently failed before writing data, or the operator
    // declared against a token with no holders. Surface it; the operator can
    // recover escrow via DistributionCampaign.seal().
    await markFailed(
      context,
      campaigns,
      campaign,
      "snapshot returned 0 holders (pipeline may have failed to index)",
    );
    return;
  }

  const totalSupply = holders.reduce((s, h) => s + h.balance, 0n);
  const payouts = proRata(holders, BigInt(campaign.totalAmount), totalSupply);

  // Persist payouts onto the campaign row so the holder/amount table
  // survives terminalCleanup (which drops the per-campaign Postgres
  // table). Useful for audit + for operator UIs reading campaign state
  // after the per-campaign tables have been cleaned up.
  if (!campaign.payouts) {
    const persisted: PersistedPayout[] = payouts.map((p) => ({
      holder: p.holder,
      sharesAtSnapshot: p.sharesAtSnapshot.toString(),
      amount: p.amount.toString(),
    }));
    campaign.payouts = persisted;
    await campaigns.setById(campaign.rowId, campaign);
  }

  const wallet = await context.evm.wallet({
    name: "corp-actions-operator",
    sponsorGas: true,
  });
  const chain = context.evm.chains[CONFIG.chain];

  // Filter to unpaid holders by reading on-chain state. The contract is the
  // sole source of truth — if the pod was killed mid-batch on a previous
  // call, the already-paid holders show up here as paid and we skip them.
  const unpaid: Payout[] = [];
  for (const p of payouts) {
    const isAlreadyPaid = await wallet.readContract(
      chain,
      CONFIG.campaignContract,
      "isPaid(bytes32,address)",
      [campaign.onChainId, p.holder],
    );
    if (!isAlreadyPaid) unpaid.push(p);
  }

  console.log(
    `[${campaign.userId}] drivePayouts: unpaid=${unpaid.length}/${payouts.length}`,
  );
  if (unpaid.length === 0) {
    await maybeMarkComplete(context, campaigns, campaign, payouts);
    return;
  }

  // Bounded concurrency. Promise.allSettled so one revert doesn't break the
  // whole batch — the contract's `AlreadyPaid` guard means duplicates are
  // safe even when we're optimistic about parallel state.
  const txByHolder = new Map<string, string>();
  for (let i = 0; i < unpaid.length; i += CONCURRENCY) {
    const batch = unpaid.slice(i, i + CONCURRENCY);
    console.log(`[${campaign.userId}] drivePayouts: sending batch ${batch.length}`);
    const results = await Promise.allSettled(
      batch.map((p) => payOne(wallet, chain, campaign, p)),
    );
    results.forEach((res, idx) => {
      if (res.status === "fulfilled" && res.value) {
        txByHolder.set(batch[idx].holder.toLowerCase(), res.value);
      }
    });
  }

  // Stitch tx hashes back into the persisted payouts so audit tooling
  // can deep-link each holder row to the actual pay() tx on basescan.
  if (txByHolder.size && campaign.payouts) {
    let mutated = false;
    for (const p of campaign.payouts) {
      const tx = txByHolder.get(p.holder.toLowerCase());
      if (tx && !p.payTxHash) {
        p.payTxHash = tx as `0x${string}`;
        mutated = true;
      }
    }
    if (mutated) await campaigns.setById(campaign.rowId, campaign);
  }

  console.log(`[${campaign.userId}] drivePayouts: batches done, checking escrow`);
  // Re-read on-chain state to decide if we're done.
  await maybeMarkComplete(context, campaigns, campaign, payouts);
}

async function payOne(
  wallet: Awaited<ReturnType<TaskContext["evm"]["wallet"]>>,
  chain: TaskContext["evm"]["chains"][keyof TaskContext["evm"]["chains"]],
  campaign: Campaign,
  { holder, amount, sharesAtSnapshot }: Payout,
): Promise<string | null> {
  try {
    const { hash } = await wallet.writeContract(
      chain,
      CONFIG.campaignContract,
      "pay(bytes32,address,uint256,uint256)",
      [
        campaign.onChainId,
        holder,
        amount.toString(),
        sharesAtSnapshot.toString(),
      ],
    );
    return hash;
  } catch (err) {
    const msg = err instanceof Error ? err.message : String(err);
    if (/AlreadyPaid/.test(msg)) return null;       // contract guard absorbed a race
    if (/AlreadySealed|InsufficientEscrow/.test(msg)) {
      console.log(`[${campaign.userId}] terminal pay failure for ${holder}: ${msg}`);
      return null;
    }
    console.log(`[${campaign.userId}] transient pay failure for ${holder}: ${msg}`);
    return null;
  }
}

async function maybeMarkComplete(
  context: TaskContext,
  campaigns: Awaited<ReturnType<TaskContext["collection"]>>,
  campaign: Campaign,
  payouts: Payout[],
) {
  if (campaign.status === "complete") return;

  // Source of truth for "is this campaign fully distributed" is the
  // contract's `escrowRemaining`. A single read, atomic.
  //
  // The previous implementation looped N `isPaid()` reads instead — which
  // looked correct but had a real failure mode: with sponsored gas + a
  // cluster of pay() txs, individual RPC nodes can return a stale `false`
  // for an isPaid that's actually true on chain. One stale read kept the
  // campaign in `paying` forever and re-fired pay() (silently absorbed by
  // the AlreadyPaid guard, but noisy in operator logs). escrowRemaining=0
  // is a single signal that's already resolved by the contract's
  // checks-effects-interactions on every pay().
  const wallet = await context.evm.wallet({
    name: "corp-actions-operator",
    sponsorGas: true,
  });
  const chain = context.evm.chains[CONFIG.chain];
  const c = await wallet.readContract<
    readonly [
      `0x${string}`, `0x${string}`, `0x${string}`,
      bigint, bigint, bigint,
      boolean, boolean,
    ]
  >(
    chain,
    CONFI

…(truncated)