compose-dividend-distribution について
トークン保有者へ保有比率に応じてUSDCを支払う、企業アクション配布アプリを構築するGoldsky Composeスキルです。Turboのスナップショットパイプラインをオンデマンドで起動し、再開可能かつ冪等な処理とオンチェーン監査記録を備えます。
主な機能
- 01配当・クーポンの按分支払い
- 02オンデマンドのTurbo保有者スナップショット
- 03再開可能で冪等なキャンペーン処理
- 04ガス代を負担するウォレットでの支払い
- 05オンチェーンの支払い監査記録
活用シーン
- 01トークン保有者への配当分配
- 02クーポンやリベートのキャンペーン実施
- 03Base Sepoliaでのテストキャンペーン
インストール
コマンドをコピーしてターミナルで実行します。
$
npx skills add https:/インストール後、Claude Code で/compose-dividend-distributionを入力して読み込まれたか確認します。
詳細
Build: 配当分配（corporate-actions）
ユーザー自身のGoldskyアカウントで、corporate-actionsディストリビューターを構築します。トークン化されたコーポレートアクション（配当、クーポン、リベート、エアドロップ）について、N人の保有者に持分比率に応じて、冪等かつ永続的に支払い、改ざん検知可能なオンチェーン監査証跡を残します。注目すべき点は、ComposeがGoldsky Turboをオンデマンドの一時的なサブルーチンとしてオーケストレーションすることです。 キャンペーンを宣言すると、オペレーターが指定した記録ブロック時点のシェアトークン保有者をスナップショットする、1回限りのjob-mode Turboパイプラインが起動します。Composeは完了を待ち、ガス代をスポンサーするウォレット経由で各保有者に支払い、その後パイプラインを削除します。常時稼働するインデックス処理はありません。
1つのHTTPタスク（declare_campaign）が、宣言 → USDCのエスクロー → スナップショットパイプラインの起動 → ポーリング → 持分比率に基づく金額の計算 → 最大25人の保有者への同時支払い → escrowRemaining == 0の確認 → パイプラインの削除、というライフサイクル全体を実行します。同じcampaignIdを再度POSTすると、どの段階で失敗していても問題なく再開できます。「この保有者に支払済みかどうか」の唯一の情報源はコントラクトであるため、二重払いは構造上起こり得ません。
このテンプレートで提供するのは、配当／corporate-actionsアプリに固有の内容（仕組みとソース）のみです。推奨される方法では、Base Sepolia上の共有・パーミッションレスなデモコントラクト（MockUSDCのmintは誰でも実行可能、キャンペーンのdeclare()も誰でも実行可能）を使用するため、デプロイは不要です。
ステップ0a — まず基本スキルを読み込む
何よりも先に、回答する、質問する、ファイルを用意する、コマンドを実行するといったことを始める前に、このテンプレートが依存する2つの基本スキルを読み込んでください。
Skill(compose)— Composeの常時参照ガイドです。黄金律（アプリについてユーザーに代わって決めつけず、不明な点は質問する）と、ビルド全般のガイダンスが記載されています。Skill(compose-reference)— マニフェスト／フィールド／APIのリファレンスです。compose.yamlやタスクファイルを書く前に参照してください。
このテンプレートでは、これらのルールとリファレンスを意図的に省略しています。正しく構築するために必須であり、ここでは繰り返しません。両方を読み込むまで先に進まないでください。
モードの判定
利用可能なツールからモードを選択してください。
deployComposeAppツールが利用可能（Goldsky webapp chatbot）の場合。 この例はアプリ内ですべてデプロイします。ジョブモードの Turbo パイプラインはデプロイ時にプロビジョニングされるのではなく、実行時に Turbo API 経由で起動されます（src/lib/turbo.ts内のアプリ内ctx.fetchPOST）。そのため、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。推奨される共有コントラクトの方法ではデプロイ対象がないため、3 つのcontracts/*.solソースは除外します（共有方法では何もデプロイしません）。独自にデプロイする方法では、.solファイルも生成し、MockUSDC、ShareToken（保有者と数量の配列を渡す）、DistributionCampaign に対してdeployContractを使用します。src/lib/constants.tsの 4 つのコントラクト CONFIG 値（デプロイされた 3 つのアドレスとshareTokenDeployBlock）を設定してから、deployComposeAppを呼び出します。GOLDSKY_PROJECT_KEYシークレットはユーザーが最後に設定します。アプリ内デプロイではシークレットの検証がスキップされるため、これがなくてもdeployComposeAppは成功しますが、ユーザーが Compose アプリのダッシュボードでシークレットを追加し、ダッシュボードから再デプロイして Pod に反映させるまではアプリは動作しません（シークレットはデプロイ時に Pod に組み込まれ、ホットリロードされません）。チャットからシークレットを設定しようとしてはいけません。設計上、そのためのツールはありません。注：チャットボットにはwriteContractツールがないため、corp-actions-operatorウォレットにテスト用 MockUSDC を発行できません。キャンペーンを宣言する前に、ユーザーは CLI（goldsky compose writeContract）またはダッシュボードから Step 4 のミントを実行する必要があります（タスクは宣言時に USDC のapproveを実行しますが、その前にオペレーターのウォレットが USDC を保有している必要があります）。Bashが利用可能（ローカル CLI／コーディングエージェント）の場合。 以下の手順を直接実行し、出力を解析して、取得した値を後続のコマンドに代入します。- どちらも利用できない場合（参照 Q&A のみ）。 アプリの機能とライフサイクルを説明します。手順を尋ねられた場合に限り、一度に 1 つのコマンドを出力し、ユーザーに実行結果を貼り付けてもらいます。Bash を使ってローカルで実行するには、
npx skills add goldsky-io/goldsky-agentを案内してください。
必須事項
- 共有の、誰でも利用できるデモコントラクトを参照した状態で提供されます。デプロイは不要です。 MockUSDC には誰でも実行できる
mintがあり、DistributionCampaign には誰でも実行できるdeclare()があるため、誰でもこれらを使ってキャンペーンを実行できます。共有デモはデフォルトで Base Sepolia にあります。src/lib/constants.tsには切り替え可能な既知の Base メインネットのデプロイ先も記載されています（メインネットでは実際にガス代がかかります）。これらはデモおよび利用開始用であり、本番環境向けではないことを、ユーザーに文章で伝えてください。 - プロジェクト API キー 1 つですべてを行えます。 次の 3 通りで使用します。
goldsky compose deployContract／deploy／writeContractではGOLDSKY_API_TOKEN環境変数として使用します（こちらを推奨します。キーが argv やシェル履歴に残りません）。実行中のアプリが Turbo パイプラインを起動／ポーリング／削除できるよう、GOLDSKY_PROJECT_KEYシークレットの値として使用します。さらに Step 5 の HTTP タスクでは$GOLDSKY_TOKENベアラーとして使用します（または同じプロジェクトから発行した別の 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 パイプラインのインフラがないローカル／開発用 Compose クラスターでは動作しません。 実際の Goldsky（app.goldsky.com）に対してデプロイします。ユーザーが実行する環境もここです。
アプリ（全ソース）
これは完全な配当アプリです。以下のファイルをそのままスキャフォールドしてください（Step 0b では degit を使ってディスクに書き込みます。アプリ内のフローでは、以下のブロックからメモリー上にスキャフォールドします）。src/lib/constants.ts にある共有 Base Sepolia デモコントラクトを使うため、推奨パスではデプロイは不要です。独自のコントラクトアドレスを設定する（Step 1 Branch B）か、Base メインネットに切り替える場合にのみ、src/lib/constants.ts を編集してください。
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 — 例をスキャフォールドする
企業アクションの例だけを新しいディレクトリに取得します（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
ユーザーがすでに例をクローンしている場合は、この手順をスキップして、そのディレクトリに cd してください。
ウォレットやコントラクトに関する手順に進む前に、アプリ名を尋ねてください。これが最初の質問です。 「アプリ名は何にしますか？（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 を実行してから、もう一度確認してください（stable チャンネルでは冪等です）。Dividend 固有の確認事項：
- Compose CLI のバージョン —
goldsky compose --versionを実行します（goldsky compose 0.8.1と表示されます）。バージョンが0.8.1より古い場合、またはgoldsky compose deployContract／writeContractが不明なコマンドとして扱われる場合は、続行する前にgoldsky compose updateを提案し、もう一度確認してください。以下のデプロイ（ステップ 1 のブランチ B）とミント（ステップ 4）では、これらのコマンドが必要です。 - プロジェクト API キー - キーは全工程で 1 つだけ使います（必須事項を参照）。
deployContract／deploy／writeContract用のGOLDSKY_API_TOKEN環境変数（キーが argv やシェル履歴に残らないため推奨）、GOLDSKY_PROJECT_KEYシークレット、およびステップ 5 の$GOLDSKY_TOKENbearer です。認証方法：シェルで一度GOLDSKY_API_TOKEN=<project API key>をエクスポートするか、有効なgoldsky loginセッションを使います。優先順位：--token>GOLDSKY_API_TOKEN>~/.goldsky/auth_token。ダッシュボード（https://app.goldsky.com）の Settings > API Keys で生成します。ユーザーに準備してもらい、キーを表示しないでください。 node+npm—npm --versionを実行してから、npm installを実行します（アプリにはviemがバンドルされています）。foundry—cast --version／forge --versionを実行します。必要なのはステップ 6 のcast callによる検証のみです。デプロイやミントには不要です（deployContractは CLI 内でコンパイルし、ステップ 4 のミントではガス代をスポンサーするwriteContractを使います）。
ステップ 1 — コントラクト
ブランチ A — 共有のデモ用コントラクトを再利用する（推奨）。 デプロイは不要です。src/lib/constants.ts の CONFIG は、上記の共有 Base Sepolia アドレスのままにします。ステップ 2 に進んでください。
ブランチ B - 独自にデプロイする。 アプリのディレクトリから次のコマンドを実行します（アプリ名の取得には compose.yaml を読み込みます）。3 つすべてのデプロイは、Base Sepolia 上のガス代をスポンサーする Compose ウォレットを通じて行われ（資金を入金した EOA は不要）、GOLDSKY_API_TOKEN 環境変数で認証されます（未認証で実行すると、"Please run goldsky login, set GOLDSKY_API_TOKEN, or pass --token to the command." というエラーになります）。確認ルールに従い、3 回に分けて承認を求める代わりに、3 つまとめて「すべてデプロイしますか？」と提案してかまいません。MockUSDC と DistributionCampaign にはコンストラクター引数はありません。ShareToken は scripts/seed-holders.json から作成する (address[] holders, uint256[] amounts) を受け取ります。
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 を確認してください。
3 つのアドレスと shareTokenDeployBlock（ShareToken の Deploy Block。上記で $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. と表示します。ここでは安全にスキップできます。このアプリではシグネチャ文字列を使った raw の 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" というエラーが発生してデプロイに失敗します。）
Step 3 — Compose アプリをデプロイする
goldsky compose deploy
Compose-cloud はホストされた Neon Postgres DB を自動でプロビジョニングし、そこを参照する CORPORATE_ACTIONS というプロジェクトシークレットを作成します。ジョブモードのパイプラインは、その DB にスナップショットを書き込みます。初回のデプロイには 1〜2 分かかる場合があります。Deployed compose app: <the chosen app name>（例：corporate-actions）と HTTP タスク URL が表示されることを確認してください。
Step 4 — オペレーターウォレットに MockUSDC をミントする
declare_campaign タスク自体が、ガス代をスポンサーする Compose ウォレット corp-actions-operator（sponsorGas: true）を使って、オンチェーンで USDC の approve と declare を実行します。そのため、この Compose ウォレットは USDC を保有している必要があります。approve は宣言時にタスク内で実行されるため、手動で行う必要はありません。ウォレットに USDC をミントするだけでよく、ミントのガス代もスポンサーされます。資金を入れた EOA もフォーセットも不要です。
(a) オペレーターウォレットのアドレスを取得します。 wallet create はアプリのデプロイ前後どちらでも実行できるため、いつでも実行できます：COMPOSE_WALLET=$(goldsky compose wallet create corp-actions-operator --json | jq -r .address)。この操作は冪等で、ウォレットがすでに存在する場合は既存のウォレットを返します。（ホストされた Postgres がまだ存在しない場合は persisted: false と報告されることがありますが、ランタイムが解決するアドレスは同じです。）
(b) ウォレットに USDC をミントします（ガス代スポンサーあり）。 MockUSDC.mint はオープンなので、writeContract の送信者は関係なく、ガス代はスポンサーされます。余裕をもってミントしてください。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 として使ってください。
Step 5 — キャンペーンを宣言する
ファイナリティを過ぎたレコードブロックを選び、POST します。$GOLDSKY_TOKEN は HTTP タスク用のベアラートークンです。ここで他の場所と同じプロジェクト API キーを使うか（または同じプロジェクトから発行した別の Compose API トークンを使います）、どちらも Settings > API Keys から取得できます：
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 から）。失敗時には {"error":true,"code":"...","message":"..."} を返して終了コード 1 で終了します。以前の curl では 404 でも終了コード 0 でした。
これにより、10,000 mUSDC の分配が宣言されます。Compose が保有者のスナップショットを取得し、比例配分を計算して、25 件の pay() 呼び出しを 1 回のバッチで実行する間、リクエストは約 10〜30 秒開いたままになります。レスポンスボディには最終的なキャンペーン状態が含まれます（正常終了時は complete、追加の drive 呼び出しが必要な場合は paying です。その場合は同じ campaignId を再度 POST してください）。
Step 6 — オンチェーンで確認する
$ON_CHAIN_ID は Step 5 の POST レスポンスボディにある onChainId フィールドの値です。コントラクトはこれを onChainId = keccak256(encodePacked(operatorWallet, campaignId)) として計算します（タスクでは userId = campaignId を設定します）。getCampaign(bytes32) は Campaign 構造体全体を返すため、フィールドをデコードするには 8 フィールドすべてを含むタプルの戻り値シグネチャを指定してください（escrowRemaining の個別ゲッターはありません）：
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 のデプロイでは検証され、組み込まれるため、デプロイ前に設定しておく必要があります）。設定を忘れた場合は、
goldsky compose secret set GOLDSKY_PROJECT_KEY --value "$GOLDSKY_PROJECT_KEY" --redeployを実行します（シェルでエクスポートした値を使用し、文字列をそのまま貼り付けないでください）。これにより、シークレットが設定され、再デプロイが開始されるため、1 回の手順で 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-operatorCompose ウォレットに、totalAmountに対して十分な MockUSDC があることを確認してください（タスクはdeclareの実行時にapproveも行います。ステップ 4 でウォレットに追加発行してください）。payTokenはコントラクトレベルの設定ではありません。src/lib/constants.tsのCONFIG.payTokenを参照する、declare()ごとの引数です。トークンが間違っている場合は、そこでCONFIG.payTokenを編集して、追加発行したトークン、または保有しているトークンに合わせてから再デプロイしてください。deployContractが「This contract has already been deployed with this app...」というエラーで拒否されます。 同一のコントラクト、コンストラクター引数、アプリ名を使うと、同じ CREATE2 アドレスになるため、CLI はトランザクションの実行前に拒否し、Deploy Blockは出力されません。対処法は、コンストラクター引数を変更する、ソースを変更する、または別のアプリ名を使うことです。新しい名前は、大文字・小文字だけ、または-と_だけが異なる名前にはできません。名前は正規化されるため（小文字に変換し、[-_]+を-に変換）、my-app、My_App、my__appは衝突し、デプロイは 409 を返します。ここで使う引数なしの 2 つのコントラクト（MockUSDC、DistributionCampaign）では、コンストラクター引数とソースによる対処はできません。別のアプリ名を使う以外に方法はありません。ShareToken は(address[],uint256[])を受け取るため、こちらではコンストラクター引数を変更できます。⚠ アプリ名を変更すると、以降のデプロイアドレスがすべて変わり、compose deployのアプリ名（選択したアプリ名。例：corporate-actions）と競合します。そのため、適用できる場合はコンストラクター引数による対処を優先してください。- キャンペーンが
completeではなくpayingを返します。 この処理で一部のpay()呼び出しが反映されませんでした。同じcampaignIdで再度 POST してください。支払い済みの保有者はオンチェーンのisPaid()によってスキップされ、実行は続きから再開されます。 - ホスト型 DB／スナップショットの行がありません。 デプロイによって
CORPORATE_ACTIONSシークレットが作成されたことを確認してください（compose-cloud が自動的に作成します）。作成されていない場合は、シェルでGOLDSKY_API_TOKENを設定して再デプロイしてください。
してはいけないこと
- Base Sepolia の共有コントラクトを本番環境のターゲットとして使用しないでください。許可なく利用できるため、誰でも mUSDC をミントしたり、キャンペーンを宣言したりできます。
- Turbo パイプラインのインフラがないローカル／開発用 Compose クラスターにはデプロイしないでください。スナップショットの手順が停止します。実際の Goldsky に対してデプロイしてください。
GOLDSKY_PROJECT_KEYを、compose secret setの呼び出し以外でシェルの履歴に残るコマンドラインに記述しないでください。また、コミットしないでください。recordBlockをshareTokenDeployBlockより小さくしないでください。また、未来のブロックを指定しないでください。スナップショットは定義上、過去を振り返るものです。src/lib/constants.tsのキャンペーンごとのパイプライン名／テーブル名（pipelineName／aggTableName）を手作業で編集しないでください。これらはcampaignIdから導出されており、再開を機能させるには安定した状態を保つ必要があります。
関連情報
/compose— Compose アプリをゼロから構築する方法、または Compose の説明です。/compose-reference— マニフェスト、CLI、TaskContext API、ウォレット、ガスのスポンサー提供、コード生成について説明します。/compose-doctor— 破損した Compose アプリの診断と修正を行います。/turbo-pipelines／/turbo-operations— スナップショットをカスタマイズする場合に、ジョブモードのパイプラインの構成とライフサイクルを確認できます。/auth-setup—goldsky loginの手順です。
コメント
Markdown 対応