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

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

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

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

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

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

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

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

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)