compose-dividend-distribution

构建：Compose 股息分配（corporate-actions）

在用户自己的 Goldsky 账户下搭建 corporate-actions 分发器。它会为代币化公司行动（股息、票息、返利、空投）按比例向 N 位持有人付款，确保操作幂等且持久，并留下防篡改的链上审计轨迹。其亮点在于：Compose 将 Goldsky Turbo 编排为按需运行的临时子程序。 声明活动后，会启动一个一次性的 job-mode Turbo 管道，在运营方提供的记录区块处对份额代币持有人进行快照；Compose 会等待其完成，为每位持有人通过 Gas 赞助钱包付款，然后删除该管道。无需持续运行索引。

一个 HTTP 任务（declare_campaign）驱动整个生命周期：声明 → 托管 USDC → 启动快照管道 → 轮询 → 计算按比例分配的金额 → 并发向最多 25 位持有人付款 → 验证 escrowRemaining == 0 → 删除管道。重新 POST 相同的 campaignId，即可在任何故障后顺利恢复；合约是判断“该持有人是否已收到付款”的唯一事实来源，因此从结构上杜绝了重复付款。

此模板仅提供股息／corporate-actions 应用特有的内容——工作原理及其源代码。建议的路径使用 Base Sepolia 上共享且无需许可的演示合约（MockUSDC 的 mint 开放，活动的 declare() 开放），因此无需部署任何内容。

步骤 0a — 先加载基础技能

在做任何事情之前——在回答、提问、搭建文件或运行任何命令之前——先加载此模板依赖的两项基础技能：

1. Skill(compose) — 常驻的 Compose 指南：黄金规则（绝不擅自假设用户应用的任何情况；不确定时先提问）和通用构建指导。
2. Skill(compose-reference) — 清单／字段／API 参考；编写任何 compose.yaml 或任务文件之前都应查阅。

此模板特意省略了这些规则和参考资料——要正确构建就必须了解它们，而且本文不会重复说明。两项技能都加载完成后再继续。

模式检测

根据你可用的工具选择模式：

• 有 deployComposeApp 工具可用（Goldsky Web 应用聊天机器人）。 此示例完全在应用内完成部署。作业模式 Turbo 流水线在运行时通过 Turbo API（在 src/lib/turbo.ts 中通过应用内 ctx.fetch POST 请求）启动，而非在部署时预配，因此 deployComposeApp 可以顺利部署应用。应用内流程：先进行 Step 0b 应用名称访谈（应用名称是第一个问题），然后根据下方的 The app (full source) 在内存中搭建以下文件，并将它们传递给 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 和 src/lib/turbo.ts。采用推荐的共享合约路径时，无需部署任何内容，因此排除三个 contracts/*.sol 源文件（共享路径不部署任何内容）；采用自行部署路径时，也要搭建 .sol 文件，并对 MockUSDC、ShareToken（传入 holder/amount 数组）和 DistributionCampaign 使用 deployContract。在 src/lib/constants.ts 中填入四个合约 CONFIG 值（三个已部署地址以及 shareTokenDeployBlock），然后调用 deployComposeApp。GOLDSKY_PROJECT_KEY 密钥是用户的最后一步：应用内部署会跳过密钥验证，因此即使没有该密钥，deployComposeApp 也会成功，但在用户于 Compose 应用的控制面板中添加密钥并从控制面板重新部署之前，应用都无法运行，因为密钥会在部署时写入 pod，不会热加载。绝不要尝试在聊天中设置密钥；按设计，这里没有相应工具。注意：聊天机器人没有 writeContract 工具，因此无法向 corp-actions-operator 钱包铸造测试用 MockUSDC；声明活动前，用户必须通过 CLI（goldsky compose writeContract）或控制面板运行 Step 4 的铸币操作（任务会在声明时自行执行 USDC approve，但操作员钱包必须先持有 USDC）。
• 有 Bash 可用（本地 CLI / 编码 Agent）： 直接执行以下步骤，解析输出，并将捕获的值替换到后续命令中。
• 两者都没有（纯参考问答）： 说明应用的功能和生命周期；只有在用户要求逐步指导时，才一次输出一条命令，并让用户粘贴输出。请引导他们运行 npx skills add goldsky-io/goldsky-agent，以便在本地通过 Bash 使用该工具。

不可妥协事项

• 默认指向共享、无需许可的演示合约——无需部署。 MockUSDC 开放了 mint，DistributionCampaign 开放了 declare()，因此任何人都可以在这些合约上运行活动。默认情况下，共享演示合约部署在 Base Sepolia 上；src/lib/constants.ts 还列出了已知的 Base 主网部署地址，你可以切换到这些地址（在主网上会产生实际 Gas 费用）。请在说明中告知用户，这些合约仅用于演示和快速入门，不适用于生产环境。
• 一个项目 API 密钥即可完成所有工作。 它有三种用途：作为 goldsky compose deployContract / deploy / writeContract 命令的 GOLDSKY_API_TOKEN 环境变量（首选，可避免密钥出现在 argv 和 shell 历史记录中）；作为 GOLDSKY_PROJECT_KEY 密钥的值（以便运行中的应用启动、轮询和删除 Turbo 管道）；以及作为 Step 5 HTTP 任务的 $GOLDSKY_TOKEN bearer（也可以使用从同一项目生成的单独 Compose API 令牌）。在 Goldsky 控制面板的 Settings > API Keys 中生成此密钥。没有此密钥，应用将无法运行。
• recordBlock 必须 <= currentBlock，且应超过最终确认区块（例如 currentBlock - 32）。快照回溯到过去——它是确定谁能获得付款的截止点。不支持将记录区块设为未来区块。
• 未经事先展示确切命令并获得明确确认，绝不要运行 goldsky compose deployContract（自行部署）、goldsky compose deploy、goldsky compose secret set、git push 或 gh repo create。
• 此设计支持续跑——绝不必担心重复付款。 使用相同的 campaignId 再次 POST，会推动现有活动继续执行。每个持有者的链上 isPaid() 检查，以及合约的 require(!paid[id][holder]) 防护，意味着 Compose 可以随时崩溃或重启，完全不会有重复付款的风险。
• 此示例无法在缺少 Turbo 管道基础设施的本地/dev Compose 集群中运行。 它会部署到真实的 Goldsky（app.goldsky.com），用户本来也会在那里运行它。

应用（完整源代码）

这是完整的分红应用。请原样创建这些文件（Step 0b 会通过 degit 将它们写入磁盘；应用内流程则会根据下方代码块在内存中创建这些文件）。src/lib/constants.ts 中的共享 Base Sepolia 演示合约意味着推荐路径无需部署任何内容。只需编辑 src/lib/constants.ts，即可接入你自己的合约地址（Step 1 Branch B），或切换到 Base 主网。

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

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,
    CONFIG.campaignContract,
    "campaigns(bytes32) view returns (address,address,address,uint256,uint256,uint256,bool,bool)",
    [campaign.onChainId],
  );
  const escrowRemaining = c[4];
  console.log(
    `[${campaign.userId}] maybeMarkComplete: escrowRemaining=${escrowRemaining}`,
  );
  if (escrowRemaining > 0n) return; // not done; caller can re-drive to keep paying

  await terminalCleanup(context, campaign);
  await campaigns.setById(campaign.rowId, {
    ...campaign,
    status: "complete",
    completedAt: Date.now(),
  });
  console.log(`[${campaign.userId}] complete: paid ${payouts.length} holders`);
}

async function markFailed(
  context: TaskContext,
  campaigns: Awaited<ReturnType<TaskContext["collection"]>>,
  campaign: Campaign,
  reason: string,
) {
  await terminalCleanup(context, campaign);
  await campaigns.setById(campaign.rowId, {
    ...campaign,
    status: "failed",
    failedAt: Date.now(),
    failureReason: reason,
  });
  console.log(`[${campaign.userId}] failed: ${reason}`);
}

/**
 * Belt-and-suspenders cleanup. Turbo auto-deletes successful job-mode
 * pipelines ~1h after completion, but errored jobs stay around forever
 * unless we DELETE them. We always run both regardless of terminal status
 * so the user's account stays clean across many demo runs.
 *
 * Order matters: DELETE the pipeline first (releases the sink writer's
 * connection), THEN drop the tables.
 */
async function terminalCleanup(context: TaskContext, campaign: Campaign) {
  await deletePipeline(context, campaign.pipelineName).catch(() => {});
  const transfers = aggTableName(campaign.userId);
  await dropCampaignTables(context, transfers).catch(() => {});
}

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

src/lib/turbo.ts

import type { TaskContext } from "compose";
import {
  aggTableName,
  CONFIG,
  pipelineName,
} from "./constants";
import type { Hex } from "./types";

/**
 * HTTP client for Goldsky's Turbo pipeline API. Same endpoints the `goldsky`
 * CLI uses, talking directly to api.goldsky.com over public ingress (the
 * compose pod has outbound network for this).
 *
 * Auth is a project API token surfaced as `context.env.GOLDSKY_PROJECT_KEY`
 * — declared in compose.yaml's `secrets:` block, set once via
 * `goldsky compose secret set GOLDSKY_PROJECT_KEY --value "$GOLDSKY_PROJECT_KEY"`.
 */

const API_BASE = "https://api.goldsky.com/api/v1";

function authHeaders(env: Record<string, string>): Record<string, string> {
  const key = env.GOLDSKY_PROJECT_KEY;
  if (!key) {
    throw new Error(
      "GOLDSKY_PROJECT_KEY missing — declare it under `secrets:` in compose.yaml " +
        "and set its value with `goldsky compose secret set GOLDSKY_PROJECT_KEY`",
    );
  }
  return {
    Authorization: `Bearer ${key}`,
    "Content-Type": "application/json",
  };
}

export type TurboState =
  | "running"
  | "starting"
  | "paused"
  | "stopped"
  | "error"
  | "completed"
  | "unknown";

interface StateResponse {
  status?: string;
  state?: string;
  errors?: unknown;
  success?: boolean;
  error?: string;
}

/**
 * Build the job-mode pipeline definition that snapshots holders of a single
 * ERC-20 at a specific block.
 *
 * Why `erc20_transfers` and not `logs`?
 *   `job: true` requires every source to be a "hybrid source" (one that
 *   supports a bounded backfill). On Base mainnet, `logs` is NOT hybrid;
 *   `erc20_transfers` is. (Pipeline that tried `base.logs` failed with
 *   `job_mode is enabled but the following source(s) do not support it`.)
 *
 * Why `to_u256(amount)` before the cast?
 *   `erc20_transfers.amount` is `FixedSizeBinary(32)` and DataFusion has no
 *   direct cast from FixedSizeBinary to DOUBLE/DECIMAL. `to_u256` is the
 *   canonical bridge — it returns a numeric type that casts cleanly.
 *   (See the failing `corp-actions-5a06330a7f49ce61` pod which tripped on
 *   `Unsupported CAST from FixedSizeBinary(32) to Float64`.)
 *
 * Pieces:
 *   - source: `base.erc20_transfers` with `start_at: earliest` (required for
 *     hybrid-source / job-mode), narrowed by a filter that bounds both ends
 *     of the scan: `lower(address) = <token> AND block_number BETWEEN
 *     <deployBlock> AND <recordBlock>`. Per Jeff: a filter-level lower+upper
 *     block range is dramatically faster than a source-level `end_block`
 *     because the planner can prune partitions/files before scanning.
 *   - transform: split each Transfer into `+amount` at `recipient` and
 *     `-amount` at `sender` (skip zero-address sender for mints)
 *   - sink: postgres_aggregate sums per-account deltas into a per-campaign
 *     agg table so concurrent campaigns can't collide on SUM
 */
export function buildSnapshotPipeline(input: {
  campaignId: string;
  shareToken: Hex;
  recordBlock: bigint;
}): Record<string, unknown> {
  const transfersTable = aggTableName(input.campaignId);
  const tokenLower = input.shareToken.toLowerCase();
  const startBlock = CONFIG.shareTokenDeployBlock;
  const endBlock = Number(input.recordBlock);

  return {
    resource_size: "s",
    job: true,
    sources: {
      transfers: {
        type: "dataset",
        dataset_name: `${CONFIG.turboChain}.erc20_transfers`,
        version: "1.2.0",
        // `start_at` MUST be 'earliest' for hybrid-source / job-mode
        // semantics. Setting it to a block number makes the source
        // non-hybrid and Turbo refuses to run with job:true. The
        // block-range filter below carries the actual scan bounds.
        start_at: "earliest",
        filter:
          `lower(address) = '${tokenLower}' ` +
          `AND block_number BETWEEN ${startBlock} AND ${endBlock}`,
      },
    },
    // The v1 pipelines API requires a `transforms` object even when there's
    // nothing to do — leave it empty.
    transforms: {},
    // Why no SQL transform?
    //   We tried — extensively. `amount` arrives as `FixedSizeBinary(32)` at
    //   the DataFusion layer and there is no in-pipeline conversion to a
    //   numeric/decimal that the planner accepts. Every UDF route either
    //   silently passes binary through (invalid Utf8 downstream),
    //   misinterprets hex as decimal, or hits "Unsupported CAST from
    //   FixedSizeBinary(32) to <T>". The Postgres sink, however, has its
    //   own binary→numeric mapping. So we sink the raw Transfer rows and
    //   compute balances via a SQL aggregate over Postgres at read time
    //   (see `getHolders` in lib/db.ts). One more network round-trip per
    //   campaign in exchange for not fighting DataFusion.
    sinks: {
      transfers_sink: {
        type: "postgres",
        from: "transfers",
        schema: "public",
        table: transfersTable,
        primary_key: "id",
        secret_name: "CORPORATE_ACTIONS",
      },
    },
  };
}

export async function createSnapshotPipeline(
  ctx: TaskContext,
  input: {
    campaignId: string;
    shareToken: Hex;
    recordBlock: bigint;
  },
): Promise<{ name: string }> {
  const name = pipelineName(input.campaignId);
  const definition = buildSnapshotPipeline(input);

  const body = {
    name,
    resource_size: "s",
    description: `Holder snapshot for campaign ${input.campaignId} at block ${input.recordBlock}`,
    definition,
  };

  console.log(
    `[turbo] POST /pipelines: name=${name}, ` +
      `block_range=[${CONFIG.shareTokenDeployBlock}, ${input.recordBlock}]`,
  );
  try {
    const res = await ctx.fetch(`${API_BASE}/pipelines`, {
      method: "POST",
      headers: authHeaders(ctx.env),
      body,
    });
    console.log(`[turbo] POST /pipelines response: ${JSON.stringify(res).slice(0, 500)}`);
  } catch (err) {
    const msg = err instanceof Error ? err.message : String(err);
    console.log(`[turbo] POST /pipelines threw: ${msg}`);
    throw err;
  }
  return { name };
}

/**
 * Read raw upstream pipeline state.
 *
 * NB: we use `/state` (proxied unchanged from streamling) NOT `/status`. The
 * v1 `/status` endpoint runs a status normalizer that maps `completed` →
 * `"UNKNOWN"`, which would silently break job-mode termination detection.
 */
export async function getPipelineState(
  ctx: TaskContext,
  name: string,
): Promise<TurboState> {
  try {
    // Use `/state` (proxied to streamling-agent) NOT `/pipelines/<name>` or
    // `/status` — those return CACHED registry data that lags k8s by minutes
    // (or forever for job-mode pipelines that were cleaned up). `/state`
    // queries the actual k8s deployment, so it's the only reliable source
    // for "did the pipeline finish".
    const res = await ctx.fetch<StateResponse>(
      `${API_BASE}/pipelines/${encodeURIComponent(name)}/state`,
      { method: "GET", headers: authHeaders(ctx.env) },
    );

    // streamling-agent returns `{success: false, error: "...not found"}`
    // when the k8s deployment is gone — which for a job-mode pipeline means
    // it completed and was cleaned up. The caller (driveSnapshot) treats
    // this as "unknown" + uses the per-campaign table's row count as the
    // actual success signal.
    if (res?.success === false) {
      const errMsg = String(res?.error ?? "").toLowerCase();
      if (/not found|missing|deployments?\b/i.test(errMsg)) return "unknown";
    }

    const raw = (res?.status ?? res?.state ?? "unknown").toString().toLowerCase();
    console.log(`[turbo] /state(${name}) raw=${raw}`);
    if (
      raw === "completed" || raw === "complete" ||
      raw === "succeeded" || raw === "success" ||
      raw === "finished" ||
      raw === "paused" || raw === "stopped"
    ) {
      return "completed";
    }
    if (raw === "running" || raw === "starting" || raw === "deploying") {
      return "running";
    }
    if (raw === "error" || raw === "failed") return "error";
    return "unknown";
  } catch (err) {
    const msg = err instanceof Error ? err.message : String(err);
    // Direct 404 (pipeline registry entry has been deleted) — same as the
    // "deployment not found" case above: treat as unknown, let the caller
    // decide based on the table contents.
    if (/404|not found/i.test(msg)) return "unknown";
    throw err;
  }
}

export async function deletePipeline(
  ctx: TaskContext,
  name: string,
): Promise<void> {
  try {
    await ctx.fetch(
      `${API_BASE}/pipelines/${encodeURIComponent(name)}`,
      { method: "DELETE", headers: authHeaders(ctx.env) },
    );
  } catch (err) {
    const msg = err instanceof Error ? err.message : String(err);
    // Idempotent: a 404 just means the pipeline was already cleaned up. Any
    // other error is real and the caller should know about it.
    if (!/404|not found/i.test(msg)) throw err;
  }
}

contracts/DistributionCampaign.sol

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.20;

interface IERC20 {
    function transfer(address to, uint256 value) external returns (bool);
    function transferFrom(address from, address to, uint256 value) external returns (bool);
}

/**
 * @title DistributionCampaign
 * @notice On-chain audit-trail + double-pay guard for tokenized corporate-action distributions
 *         (dividends, coupons, rebates, rewards). Operator-supplied per-holder amounts; the
 *         contract enforces escrow caps and one-time payment per holder.
 *
 * @dev Many campaigns can coexist in a single deployed instance; each is keyed by a canonical
 *      `id = keccak256(operator, userId)` to namespace user-supplied IDs across operators.
 *
 *      Reverts use string `require` messages (not custom errors) so off-chain callers can
 *      decode them via the standard `Error(string)` selector.
 *
 *      Demo-grade. Not audited. Anyone can `declare()` a campaign with their own funds; the
 *      operator address is recorded and only the operator can `pay()` or `seal()` that campaign.
 */
contract DistributionCampaign {
    struct Campaign {
        address operator;        // wallet that called declare(); only caller for pay/seal
        address payToken;        // ERC-20 used for payouts (e.g. MockUSDC)
        address shareToken;      // tokenized equity reference (informational; for audit reconciliation)
        uint256 totalAmount;     // initial escrow pulled from operator at declare time
        uint256 escrowRemaining; // decremented per pay(); refunded on seal()
        uint256 recordBlock;     // snapshot block at declare time (audit reference)
        bool    declared;        // set on declare; prevents re-declare under same id
        bool    sealed_;         // terminal; no more pay() allowed
    }

    /// @notice campaigns[canonicalId]
    mapping(bytes32 => Campaign) public campaigns;

    /// @notice paid[canonicalId][holder] = amount (0 means unpaid)
    mapping(bytes32 => mapping(address => uint256)) public paid;

    event CampaignDeclared(
        bytes32 indexed id,
        address indexed operator,
        address payToken,
        address shareToken,
        uint256 totalAmount,
        uint256 recordBlock
    );
    event HolderPaid(
        bytes32 indexed id,
        address indexed holder,
        address payToken,         // included so auditors don't need a separate getCampaign call
        uint256 amount,
        uint256 sharesAtSnapshot  // operator-supplied; lets auditors recompute pro-rata independently
    );
    event CampaignSealed(bytes32 indexed id, uint256 refunded);

    /// @notice Derive canonical id from operator + user-supplied id. Anyone can call this view to
    ///         compute the id off-chain before tx submission.
    function canonicalId(address operator, bytes32 userId) public pure returns (bytes32) {
        return keccak256(abi.encodePacked(operator, userId));
    }

    /// @notice Open a new campaign. Pulls `totalAmount` of `payToken` from msg.sender atomically.
    ///         The record block is `block.number` of the declare tx — the off-chain indexer
    ///         must wait for that block + finality depth before reading the snapshot.
    /// @param userId      operator-supplied identifier; canonical id = keccak256(operator, userId)
    /// @param payToken    ERC-20 used for payouts
    /// @param shareToken  reference share token (informational; recorded in events)
    /// @param totalAmount total escrow to pull from msg.sender via transferFrom
    /// @return id         canonical id
    function declare(
        bytes32 userId,
        address payToken,
        address shareToken,
        uint256 totalAmount
    ) external returns (bytes32 id) {
        id = canonicalId(msg.sender, userId);
        Campaign storage c = campaigns[id];
        require(!c.declared, "AlreadyDeclared");
        require(totalAmount > 0, "ZeroAmount");

        c.operator = msg.sender;
        c.payToken = payToken;
        c.shareToken = shareToken;
        c.totalAmount = totalAmount;
        c.escrowRemaining = totalAmount;
        c.recordBlock = block.number;
        c.declared = true;

        require(IERC20(payToken).transferFrom(msg.sender, address(this), totalAmount), "TransferFromFailed");

        emit CampaignDeclared(id, msg.sender, payToken, shareToken, totalAmount, block.number);
    }

    /// @notice Pay one holder for one campaign. Idempotent — reverts `AlreadyPaid` on duplicate.
    /// @dev Strict checks-effects-interactions: paid[] and escrowRemaining are updated BEFORE the
    ///      external transfer to defeat any reentrancy hook in a swapped-in pay token.
    function pay(
        bytes32 id,
        address holder,
        uint256 amount,
        uint256 sharesAtSnapshot
    ) external {
        Campaign storage c = campaigns[id];
        require(c.declared, "NotDeclared");
        require(!c.sealed_, "AlreadySealed");
        require(msg.sender == c.operator, "NotOperator");
        require(paid[id][holder] == 0, "AlreadyPaid");
        require(c.escrowRemaining >= amount, "InsufficientEscrow");
        require(amount > 0, "ZeroAmount");

        // Effects
        paid[id][holder] = amount;
        c.escrowRemaining -= amount;

        // Interaction
        require(IERC20(c.payToken).transfer(holder, amount), "TransferFailed");

        emit HolderPaid(id, holder, c.payToken, amount, sharesAtSnapshot);
    }

    /// @notice Close out a campaign and return any unpaid escrow to the operator.
    function seal(bytes32 id) external {
        Campaign storage c = campaigns[id];
        require(c.declared, "NotDeclared");
        require(!c.sealed_, "AlreadySealed");
        require(msg.sender == c.operator, "NotOperator");

        uint256 refund = c.escrowRemaining;
        c.escrowRemaining = 0;
        c.sealed_ = true;

        if (refund > 0) {
            require(IERC20(c.payToken).transfer(c.operator, refund), "TransferFailed");
        }
        emit CampaignSealed(id, refund);
    }

    /// @notice Read the full campaign struct.
    function getCampaign(bytes32 id) external view returns (Campaign memory) {
        return campaigns[id];
    }

    /// @notice True if `holder` has been paid for `id`.
    function isPaid(bytes32 id, address holder) external view returns (bool) {
        return paid[id][holder] != 0;
    }
}

contracts/MockUSDC.sol

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.20;

/**
 * @title MockUSDC
 * @notice 6-decimal mock stablecoin with permissionless mint, suitable for testnet demos only.
 * @dev Permissionless `mint` is intentional — anyone can fund any wallet for the demo.
 *      Do NOT use this contract in any production setting.
 */
contract MockUSDC {
    string public constant name = "Mock USDC";
    string public constant symbol = "mUSDC";
    uint8  public constant decimals = 6;

    uint256 public totalSupply;
    mapping(address => uint256) public balanceOf;
    mapping(address => mapping(address => uint256)) public allowance;

    event Transfer(address indexed from, address indexed to, uint256 value);
    event Approval(address indexed owner, address indexed spender, uint256 value);

    error InsufficientBalance();
    error InsufficientAllowance();

    /// @notice Permissionless mint — anyone can mint to anyone. Demo only.
    function mint(address to, uint256 value) external {
        balanceOf[to] += value;
        totalSupply += value;
        emit Transfer(address(0), to, value);
    }

    function transfer(address to, uint256 value) external returns (bool) {
        _transfer(msg.sender, to, value);
        return true;
    }

    function approve(address spender, uint256 value) external returns (bool) {
        allowance[msg.sender][spender] = value;
        emit Approval(msg.sender, spender, value);
        return true;
    }

    function transferFrom(address from, address to, uint256 value) external returns (bool) {
        uint256 a = allowance[from][msg.sender];
        if (a < value) revert InsufficientAllowance();
        if (a != type(uint256).max) {
            allowance[from][msg.sender] = a - value;
        }
        _transfer(from, to, value);
        return true;
    }

    function _transfer(address from, address to, uint256 value) internal {
        uint256 b = balanceOf[from];
        if (b < value) revert InsufficientBalance();
        unchecked { balanceOf[from] = b - value; }
        balanceOf[to] += value;
        emit Transfer(from, to, value);
    }
}

contracts/ShareToken.sol

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.20;

/**
 * @title ShareToken
 * @notice Minimal ERC-20 representing tokenized equity shares.
 *         Pre-mints to a list of holders in the constructor.
 *
 * @dev Demo-grade: no permits, no transfer hooks, no ERC-2612.
 */
contract ShareToken {
    string public constant name = "Example Issuer Shares";
    string public constant symbol = "EIS";
    uint8  public constant decimals = 18;

    uint256 public totalSupply;
    mapping(address => uint256) public balanceOf;
    mapping(address => mapping(address => uint256)) public allowance;

    event Transfer(address indexed from, address indexed to, uint256 value);
    event Approval(address indexed owner, address indexed spender, uint256 value);

    error LengthMismatch();
    error InsufficientBalance();
    error InsufficientAllowance();

    constructor(address[] memory holders, uint256[] memory amounts) {
        if (holders.length != amounts.length) revert LengthMismatch();
        uint256 supply;
        for (uint256 i = 0; i < holders.length; i++) {
            balanceOf[holders[i]] += amounts[i];
            supply += amounts[i];
            emit Transfer(address(0), holders[i], amounts[i]);
        }
        totalSupply = supply;
    }

    function transfer(address to, uint256 value) external returns (bool) {
        _transfer(msg.sender, to, value);
        return true;
    }

    function approve(address spender, uint256 value) external returns (bool) {
        allowance[msg.sender][spender] = value;
        emit Approval(msg.sender, spender, value);
        return true;
    }

    function transferFrom(address from, address to, uint256 value) external returns (bool) {
        uint256 a = allowance[from][msg.sender];
        if (a < value) revert InsufficientAllowance();
        if (a != type(uint256).max) {
            allowance[from][msg.sender] = a - value;
        }
        _transfer(from, to, value);
        return true;
    }

    function _transfer(address from, address to, uint256 value) internal {
        uint256 b = balanceOf[from];
        if (b < value) revert InsufficientBalance();
        unchecked { balanceOf[from] = b - value; }
        balanceOf[to] += value;
        emit Transfer(from, to, value);
    }
}

Step 0b — 创建示例项目

仅将 corporate-actions 示例拉取到一个全新目录（不包含 Git 历史记录）：

npx -y degit goldsky-io/documentation-examples/compose/corporate-actions#6abe62878cb92f2569538ba9572b049ed5949a01 dividend-distribution
cd dividend-distribution

如果无法使用 npx degit，请改用稀疏克隆：

git clone --filter=blob:none --sparse https://github.com/goldsky-io/documentation-examples.git
cd documentation-examples && git checkout 6abe62878cb92f2569538ba9572b049ed5949a01 && git sparse-checkout set compose/corporate-actions && cd compose/corporate-actions

如果用户已经克隆了该示例，则跳过此步骤并进入该目录。

进行任何钱包或合约操作之前，先询问应用名称——这是访谈中的第一个问题： “应用应该叫什么？（建议使用 corporate-actions）”。名称之后很难更改：它会限定具名钱包（例如 corp-actions-operator），也会参与每次 deployContract 的 CREATE2 盐值（Step 1 Branch B 会直接从 compose.yaml 读取应用名称），因此必须现在确定。用户犹豫不决时，接受默认名称 corporate-actions。degit 脚手架在 compose.yaml 中预设了 name: "corporate-actions"——继续之前，请将此 name: 改为用户选定的名称。

预检

goldsky CLI 和身份验证检查是标准的 Compose 预检（参见 /compose 和 /auth-setup）。如果 goldsky compose --version 输出 compose CLI is not installed，请运行 goldsky compose install（在稳定通道上可幂等执行），然后再次检查。Dividend 专属检查：

1. Compose CLI 版本 — 运行 goldsky compose --version（输出 goldsky compose 0.8.1）。如果版本低于 0.8.1，或者 goldsky compose deployContract / writeContract 是未知命令，请先提出运行 goldsky compose update，并在继续前再次检查。以下部署（步骤 1 分支 B）和铸造（步骤 4）步骤依赖这些命令。
2. 项目 API 密钥 - 一个密钥即可完成所有操作（参见不可妥协事项）：GOLDSKY_API_TOKEN 环境变量（优先使用，避免密钥出现在 argv 和 shell 历史记录中），用于 deployContract / deploy / writeContract、GOLDSKY_PROJECT_KEY 密钥，以及步骤 5 的 $GOLDSKY_TOKEN bearer。身份验证：在 shell 中导出 GOLDSKY_API_TOKEN=<project API key> 一次，或使用已登录的 goldsky login 会话。优先级：--token > GOLDSKY_API_TOKEN > ~/.goldsky/auth_token。在控制面板 (https://app.goldsky.com) 的 Settings > API Keys 中生成。请让用户准备好密钥；不要将其回显。
3. node + npm — 运行 npm --version，然后运行 npm install（应用已打包 viem）。
4. foundry — 运行 cast --version / forge --version。仅步骤 6 的 cast call 验证需要；部署或铸造不需要（deployContract 在 CLI 中编译；步骤 4 的铸造是由 Gas 赞助的 writeContract）。

步骤 1 — 合约

分支 A — 复用共享演示合约（推荐）。 无需部署。将 src/lib/constants.ts 中的 CONFIG 保持为上文显示的共享 Base Sepolia 地址。跳至步骤 2。

分支 B - 部署自己的合约。 在应用目录中运行这些命令（它们会读取 compose.yaml 获取应用名称）。三次部署都通过 Base Sepolia 上由 Gas 赞助的 Compose 钱包执行（无需注资 EOA），并使用 GOLDSKY_API_TOKEN 环境变量进行身份验证（未通过身份验证运行时会报错："Please run goldsky login, set GOLDSKY_API_TOKEN, or pass --token to the command."）。根据确认规则，可以一次性提出是否批准全部三项（“全部部署吗？”），而不是分三次确认。MockUSDC 和 DistributionCampaign 不需要构造函数参数；ShareToken 接收 (address[] holders, uint256[] amounts)，参数由 scripts/seed-holders.json 构建：

PAY_TOKEN=$(goldsky compose deployContract contracts/MockUSDC.sol --chain-id 84532 --json | jq -r .address)

# Build the two array tokens from scripts/seed-holders.json.
# Lowercase the holder addresses: the shipped seed-holders.json uses mixed-case
# (EIP-55) addresses that viem rejects on checksum, so ascii_downcase bypasses
# checksum validation (same resolved value forge accepts). Amounts are left as-is.
HOLDERS="$(jq -r '"[" + ([.holders[].address | ascii_downcase] | join(",")) + "]"' scripts/seed-holders.json)"
AMOUNTS="$(jq -r '"[" + ([.holders[].amount] | join(",")) + "]"' scripts/seed-holders.json)"
OUT=$(goldsky compose deployContract contracts/ShareToken.sol --chain-id 84532 \
  --constructor-args "$HOLDERS" "$AMOUNTS" --json)
SHARE_TOKEN=$(jq -r .address <<<"$OUT"); DEPLOY_BLOCK=$(jq -r .deployBlock <<<"$OUT")

CAMPAIGN_CONTRACT=$(goldsky compose deployContract contracts/DistributionCampaign.sol --chain-id 84532 --json | jq -r .address)

--json 会抑制 ASCII 艺术；命令失败时会以状态码 1 退出，并在 stderr 输出 {"error":true,"code":"...","message":"..."}，因此如果捕获内容为空，请检查 stderr。

将三个地址以及 shareTokenDeployBlock（ShareToken 部署区块，即上方捕获的 $DEPLOY_BLOCK）复制到 src/lib/constants.ts 中的 CONFIG（payToken、shareToken、campaignContract、shareTokenDeployBlock）。若要改为在 Base 主网上运行，请传入 --chain-id 8453，并将 chain / turboChain 设为 base。（如果你更倾向于这种方式，scripts/deploy.sh 仍可作为基于 forge/EOA 的替代方案。）deployContract 会输出 Run 'compose codegen' to generate typed contract classes.——此处可以安全地跳过；本应用使用带签名字符串的原始 wallet.writeContract / readContract，从不导入生成的合约类。

步骤 2 — 设置项目密钥

运行中的应用会使用此密钥来启动 / 轮询 / 删除 Turbo 流水线：

# export GOLDSKY_PROJECT_KEY once in your shell (or source it from a chmod-600 .env);
# never paste the literal key into a command or the transcript
goldsky compose secret set GOLDSKY_PROJECT_KEY --value "$GOLDSKY_PROJECT_KEY"

（脚手架生成的仓库中的 compose.yaml 注释和 README 仍指向普通的项目密钥命令——这是过时内容；请使用上面的 Compose 密钥命令，否则部署会失败，并报错："The following secrets referenced in the manifest do not exist or do not belong to this app".）

步骤 3 — 部署 Compose 应用

goldsky compose deploy

Compose-cloud 会自动配置托管的 Neon Postgres 数据库，并创建一个名为 CORPORATE_ACTIONS、指向该数据库的项目密钥；作业模式的管道会将快照写入该数据库。首次部署可能需要 1-2 分钟。请留意 Deployed compose app: <the chosen app name>（例如 corporate-actions）以及 HTTP 任务 URL。

步骤 4 — 向 operator 钱包铸造 MockUSDC

declare_campaign 任务本身会通过名为 corp-actions-operator 的 Compose 赞助 Gas 钱包（sponsorGas: true）在链上执行 USDC approve + declare，因此该 Compose 钱包必须持有 USDC。approve 会在任务声明时执行；无需手动操作。你只需向钱包铸造 USDC，铸币也会赞助 Gas——无需已注资的 EOA，也无需水龙头。

(a) 获取 operator 钱包地址。 wallet create 可在应用部署前后运行，因此任何时候都可以执行：COMPOSE_WALLET=$(goldsky compose wallet create corp-actions-operator --json | jq -r .address)。此操作是幂等的，如果钱包已存在，则会返回现有钱包。（如果托管 Postgres 尚不存在，它可能会报告 persisted: false；该地址仍是运行时会解析到的地址。）

(b) 向该钱包铸造 USDC（赞助 Gas）。 MockUSDC.mint 是开放的，因此 writeContract 发送方无关紧要；Gas 会获得赞助。请铸造充足的数量——1,000,000 mUSDC = 1000000000000（6 位小数），足以支持多场活动：

goldsky compose writeContract --chain-id 84532 --to $PAY_TOKEN \
  --function "mint(address,uint256)" --args $COMPOSE_WALLET 1000000000000

$PAY_TOKEN 是 src/lib/constants.ts 中的 CONFIG.payToken——在共享演示中，它是 0x8ec24F07F08745fc3D979336AA81d4Dc73f3D9DE。如果是自行部署的路径，请将你自己的 MockUSDC 地址用作 $PAY_TOKEN。

步骤 5 — 声明活动

选择一个超过最终确认区块的记录区块，然后发送 POST 请求。$GOLDSKY_TOKEN 是 HTTP 任务的 bearer token——也就是此处其他操作中使用的同一个项目 API 密钥（也可以使用从同一项目生成的单独 Compose API 令牌；两者都可在设置 > API 密钥中获取）：

RECORD_BLOCK=$(( $(curl -s -X POST https://sepolia.base.org -H 'content-type: application/json' -d '{"jsonrpc":"2.0","id":1,"method":"eth_blockNumber","params":[]}' | jq -r .result) - 32 ))

goldsky compose callTask declare_campaign "{\"campaignId\":\"0x000000000000000000000000000000000000000000000000000000000000c0a1\",\"recordBlock\":$RECORD_BLOCK,\"totalAmount\":\"10000000000\"}"

callTask 默认以已部署的应用为目标，并向完全相同的路径发送 POST 请求；它会自行解析身份验证信息（从你的 goldsky login 会话或 GOLDSKY_API_TOKEN 中获取）。失败时它会以 1 退出，并返回 {"error":true,"code":"...","message":"..."}；而旧的 curl 在收到 404 时会以 0 退出。

这会声明一笔 10,000 mUSDC 的分配。请求会保持打开约 10-30 秒，Compose 会在此期间生成持有者快照、计算按比例分配，并在一个批次中触发 25 次 pay() 调用。响应正文包含活动的最终状态（正常情况下为 complete；如果需要再次调用 drive，则为 paying——只需使用相同的 campaignId 重新发送 POST 请求）。

步骤 6 — 在链上验证

$ON_CHAIN_ID 来自步骤 5 的 POST 响应正文中的 onChainId 字段——合约按 onChainId = keccak256(encodePacked(operatorWallet, campaignId)) 计算该值（任务会将 userId 设为 campaignId）。getCampaign(bytes32) 会返回整个 Campaign 结构体，因此请声明完整的 8 字段元组返回签名，以便解码字段（没有单独的 escrowRemaining getter）：

cast call 0xA8e58573B1e10908b63d12B603aCF9C784BF904E \
  "getCampaign(bytes32)(address,address,address,uint256,uint256,uint256,bool,bool)" \
  $ON_CHAIN_ID --rpc-url https://sepolia.base.org

返回字段依次为：operator, payToken, shareToken, totalAmount, escrowRemaining, recordBlock, declared, sealed_。所有 25 位持有者都收到付款后，escrowRemaining（第 5 个值）恰好为 0。完整的审计记录可查看合约的 HolderPaid 事件。

设置密钥（最后一步）

应用已部署，但在设置 GOLDSKY_PROJECT_KEY 密钥并使用该密钥（重新）部署应用之前，应用不会运行。正在运行的应用会使用该密钥来启动、轮询和删除作业模式的 Turbo 管道（参见 src/lib/turbo.ts）。

• 应用内（聊天机器人）：应用内部署会跳过密钥验证，因此即使没有密钥，deployComposeApp 也会成功。部署后，在 Compose 应用的控制面板中进入应用的 Secrets 页面，添加 GOLDSKY_PROJECT_KEY，然后从控制面板重新部署，让 Pod 获取该密钥。仅设置密钥并不会让应用开始工作；密钥会在部署时写入 Pod，不会热重载，因此必须重新部署。切勿尝试通过聊天设置密钥；这是有意设计的，聊天中没有相应工具。
• **CLI：**在第 2 步设置该密钥（必须在 CLI 部署前设置，CLI 会验证并将其写入部署）。如果跳过了这一步：运行 goldsky compose secret set GOLDSKY_PROJECT_KEY --value "$GOLDSKY_PROJECT_KEY" --redeploy（该变量应已在 shell 中导出，切勿直接粘贴字面值），即可设置密钥并触发重新部署，让 Pod 一步获取该密钥。

故障排除

• **重新部署后，对 compose.yaml 或源文件的修改没有生效。**本地 .compose/ 捆绑包缓存已过期。运行 rm -rf .compose/，然后重新部署。
• **出现 error: Unknown command "deployContract". Did you mean command "deploy"?（或 writeContract 变体）。**Compose CLI 版本过旧。运行 goldsky compose update，确认 goldsky compose --version 输出 0.8.1 或更高版本，然后重试（参见预检步骤 1）。
• 应用启动管道时报错／pipelines API 返回 401 或 403。GOLDSKY_PROJECT_KEY 密钥缺失或错误。使用有效的项目 API 密钥重新创建该密钥（第 2 步），然后重新部署。
• **快照始终无法完成／活动卡在 snapshotting 状态。**确认 recordBlock <= currentBlock 且 >= shareTokenDeployBlock，并确认 CONFIG.shareToken / shareTokenDeployBlock 与要分发的代币一致。管道会筛选 block_number BETWEEN <deployBlock> AND <recordBlock>。
• **declare() 回滚。**确保 corp-actions-operator Compose 钱包持有足够的 MockUSDC，以满足 totalAmount 的要求（任务会在调用 declare() 时自行执行 approve；在第 4 步向钱包铸造更多代币）。payToken 不是合约级设置项，而是每次 declare() 调用的参数，来源于 src/lib/constants.ts 中的 CONFIG.payToken。如果代币不正确，请在该文件中修改 CONFIG.payToken，使其与已铸造／持有的代币一致，然后重新部署。
• **deployContract 拒绝部署："This contract has already been deployed with this app..."。**相同的合约、构造函数参数和应用名称会生成相同的 CREATE2 地址，因此 CLI 会在发起交易前拒绝部署，且不会输出 Deploy Block。可选方案：更改构造函数参数、更改源代码或使用不同的应用名称。新名称不能只在大小写或 - / _ 上有所不同：名称会被规范化（转为小写，将 [-_]+ 替换为 -），因此 my-app、My_App 和 my__app 会发生冲突，部署会返回 409。这里的两个无参合约（MockUSDC、DistributionCampaign）不适用更改构造函数参数／源代码的方案，唯一的方案是使用不同的应用名称。ShareToken 接受 (address[],uint256[])，因此可通过更改其构造函数参数来解决。⚠ 重命名应用会改变所有后续部署地址，并且会与 compose deploy 的应用名称（所选名称，例如 corporate-actions）冲突，因此只要适用，优先更改构造函数参数。
• **Campaign 返回 paying（而不是 complete）。**此次操作中有些 pay() 调用未成功。使用相同的 campaignId 重新发送 POST 请求；链上 isPaid() 会跳过已付款的持有人，并继续执行。
• **没有托管数据库／快照行。**确认部署已创建 CORPORATE_ACTIONS 密钥（compose-cloud 会自动执行此操作）；如果没有，请在 shell 中设置 GOLDSKY_API_TOKEN 后重新部署。

不应执行的操作

• 不要将共享的 Base Sepolia 合约用作生产目标——它们无需许可即可使用（任何人都可以铸造 mUSDC 或声明活动）。
• 不要将此应用部署到缺少 Turbo pipeline 基础设施的本地／开发 Compose 集群——快照步骤会卡住。请部署到真实的 Goldsky。
• 不要在命令行中输入 GOLDSKY_PROJECT_KEY，以免它留在 Shell 历史记录中；compose secret set 调用除外。也不要提交它。
• 不要将 recordBlock 设为低于 shareTokenDeployBlock 的值，也不要使用未来的区块——快照的定义就是向后回溯。
• 不要手动编辑 src/lib/constants.ts 中每个活动的 pipeline／表名称（pipelineName / aggTableName）——它们根据 campaignId 派生，必须保持稳定，才能正常恢复。

相关内容

• /compose — 从头开始构建新的／自定义 Compose 应用，或说明 Compose 是什么。
• /compose-reference — 清单、CLI、TaskContext API、钱包、Gas 赞助、代码生成。
• /compose-doctor — 诊断并修复损坏的 Compose 应用。
• /turbo-pipelines / /turbo-operations — 作业模式的 pipeline 结构与生命周期（用于自定义快照）。
• /auth-setup — goldsky login 操作指南。