turbo-builder

パイプラインビルダー

対象範囲

• 新しいパイプラインを構築します。壊れたパイプラインの診断は行いません。その場合は /turbo-doctor を使います。
• YAML のリファレンスとしては使いません。フィールドや構文の確認だけが必要な場合は、代わりに /turbo-pipelines スキルを使います。
• データセットの検索には、/datasets を使います。
• 1 つのウォレットの残高や転送を REST で検索する場合は /feeds を使います。取得した行をユーザー自身のデータベースや Webhook に格納する必要がある場合にのみ、パイプラインを構築します。

ユーザーと一緒に、最初から完全なパイプラインを段階的に構築します。有効な YAML 設定を生成して検証し、デプロイします。

ビルダーワークフロー

ステップ 1：認証を確認する

goldsky project list 2>&1 を実行して、ログイン状態を確認します。

• ログイン済みの場合： 現在のプロジェクトを確認して続行します。
• 未ログインの場合： /auth-setup スキルを使って案内します。

ステップ 2：目的を把握する

インデックス化したいデータをユーザーに尋ねます。適切な質問の例：

• どのブロックチェーン／チェーンですか？（Ethereum、Base、Polygon、Solana など）
• どのデータですか？（転送、スワップ、特定のコントラクトのイベント、すべてのトランザクションなど）
• データの保存先はどこですか？（PostgreSQL、ClickHouse、Kafka、S3 など）
• 変換処理は必要ですか？（フィルタリング、集計、エンリッチメント）
• 一度だけのバックフィルですか、それとも継続的なストリーミングですか？

ユーザーがすでに目的を説明している場合は、その説明から回答を抽出します。

ステップ 3：データセットを選ぶ

適切なデータセットを見つけるには、/datasets スキルを使います。

重要なポイント：

• 一般的なデータセット：<chain>.raw_logs、<chain>.raw_transactions、<chain>.erc20_transfers、<chain>.raw_traces
• EVM チェーン上のデコード済みコントラクトイベントの場合：address のみでフィルタリングした <chain>.raw_logs をソースにし、SQL 変換を追加して _gs_log_decode(_gs_fetch_abi(<explorer-url>, <source>), topics, data) AS decoded を呼び出し、その後 WHERE decoded.event_signature = '<EventName>(<types>)' で下流をフィルタリングします。ソースフィルターに topic0 ハッシュを設定してはいけません。パターンの詳細は /turbo-transforms を参照してください。利用可能な <chain>.decoded_logs データセットはありません。デコードは常に変換処理で行います。
• 事前にデコードされた一般的なトークンイベントには、<chain>.erc20_transfers、<chain>.erc721_transfers、<chain>.erc1155_transfers が利用でき、デコード変換は必要ありません。
• Solana の場合は、solana.transactions、solana.token_transfers などを使います。

選んだデータセットをユーザーに提示し、確認してもらいます。

ステップ 4：ソースを設定する

YAML のソースセクションを作成します。

sources:
  my_source:
    type: dataset
    dataset_name: <chain>.<dataset>
    version: 1.0.0
    start_at: latest  # REQUIRED — see below. `latest` or `earliest` (Solana uses `start_block` instead)

開始位置は必須です。省略できません。 明示的な start_at を指定せずにデータセットソースを出力してはいけません。EVM、NEAR、Bitcoin、Stellar では、このフィールドを使用します。省略しても「今から開始」という意味にはなりません。バックエンドは利用可能な最古のデータから開始するため、パイプラインは気付かないうちにチェーンの全履歴をバックフィルします。その結果、パイプラインが何日も稼働し、何百万行も書き込み、ライブデータに到達する前にシンクがいっぱいになることがあります。例外は Solana です。Solana では数値の start_block を使用し、これを省略すると最新のスロットから開始します。そのため、ユーザーに推測させず、どちらを指定したのか明記してください。

ユーザーが開始位置を指定していない場合は、YAML を書く前に確認し、次の 3 つの選択肢を過不足なく提示してください。

1. 現在から（start_at: latest）— バックフィルなしで、ライブデータのみを取得します。
2. 履歴の特定の地点から — start_at: earliest と、ソースの filter に block_number の述語を指定します（ソースで事前に適用されるため、除外対象の範囲はシンクに届きません）。Solana では代わりに数値の start_block を使用します。Stellar では、レジャーのシーケンス番号も指定できます（start_at: 60000000）。他のチェーンでは、ブロック番号は有効な start_at の値ではありません。EVM、NEAR、Bitcoin で指定できるのは earliest または latest のみです。
3. 全履歴（start_at: earliest）— チェーンの全履歴を再生するため、ライブデータが届くまでに何日ものバックフィルと何百万行ものデータが発生すること、また、そのすべてを格納できる容量がシンクに必要であることを明確に伝えてください。

次の項目についても確認してください。

• 終了ブロック： Solana のジョブモードでバックフィルする場合のみ指定します。end_block は EVM データセットソースでは無視されるため、EVM の範囲を指定するには filter に block_number の述語を指定してください。ストリーミングの場合は省略します。
• ソースレベルのフィルター： ソースでデータを絞り込むための任意のフィルター（例：特定のコントラクトアドレス）

ステップ 5：変換を設定する（必要な場合）

変換が必要な場合は、/turbo-transforms スキルを使用して支援してください。

• SQL 変換 — DataFusion SQL を使用してデータをフィルタリング、集計、結合、または再構成します
• TypeScript 変換 — カスタムロジック、外部 API 呼び出し、複雑な処理を実装します
• 動的テーブル — PostgreSQL テーブルまたはメモリ内の許可リストと結合します

変換セクションを作成します。

transforms:
  my_transform:
    type: sql
    primary_key: id
    sql: |
      SELECT * FROM my_source
      WHERE <conditions>

ステップ 6：シンクを設定する

データの送信先を尋ねます。シンクの設定には /turbo-pipelines スキルを使用します。

シンク	主要な設定
PostgreSQL	secret_name、schema、table、primary_key（任意。設定するとアップサートが有効になります）
MySQL	secret_name、schema、table、primary_key（任意。設定するとアップサートが有効になります）
ClickHouse	secret_name、table、primary_key（必須。順序付けと重複排除を設定します）
Kafka	secret_name、topic
Pub/Sub（Turbo 専用）	secret_name、topic
SQS	secret_name、queue_url
S3	bucket、region、prefix、format
Webhook	url、format

ユーザーが複数の送信先を指定した場合は、複数のシンクを持つパイプラインを 1 つ生成してください。送信先ごとに別々のパイプラインを生成してはいけません。 各シンクには、名前でソース（または変換）を参照する from: フィールドがあり、シンクはそれぞれ独立して実行されます。複数のシンクが同じソースの異なるビューを必要とする場合は、ファンアウトパターンを使用します。ビューごとに SQL 変換を追加し、各シンクの from: に適切な変換を指定します。例については、/turbo-pipelines の references/architecture-patterns.md と templates/multi-sink-pipeline.yaml を参照してください。

ソースが根本的に異なる場合（ライフサイクルが独立した異なるチェーンなど）、またはユーザーが別々のパイプラインを明示的に求めた場合にのみ、パイプラインを分割してください。

secret_name が必要なシンクでは、シークレットが存在するか確認します。

goldsky secret list

存在しない場合は、/secrets スキルを使った作成を支援します。

まだ Postgres データベースがありませんか？ Scale プラン以上では、Goldsky がホストする Postgres（Neon）データベースをプロビジョニングし、認証情報をシークレットとして 1 ステップで保存できます。

goldsky hosted-sink create --type postgres

作成されたシークレットの名前、ID、タイプが出力されます（接続文字列は出力されません）。出力された名前をシンクの secret_name として使用します。アカウントにアクセス権がない場合、コマンドはチームの請求 URL を含む Scale プランへのアップグレードメッセージを返します。その場合は、/secrets スキルを使って外部の Postgres を用意する方法に切り替えてください。

バックフィルに合わせてシンクのサイズを決めてください。 ステップ 4 の開始位置が earliest である場合、またはユーザーが数か月分あるいは全履歴の範囲を指定した場合は、無料プランのデータベース（ユーザー自身のものでも、新たにプロビジョニングしたものでも）を案内する前に、その旨を明確に伝えてください。512 MB の無料プランでは、データ量の多いデータセットの数か月分のバックフィルを保存できません。また、障害は気づきにくい形で発生します。パイプラインの検証とデプロイが成功して Running と表示された後、何も書き込まれないままチェックポイントがタイムアウトし、could not extend file because project size limit (512 MB) has been exceeded というエラーが発生します。デプロイの前に、有料または十分なサイズのデータベースを推奨するか、開始位置を狭めてください。事後診断については、/turbo-operations のストレージ超過の項目を参照してください。

ステップ 7：モードを選択する

ガイダンスには /turbo-pipelines スキルを使用します。

• ストリーミング（デフォルト） — 継続的に処理します。end_block は設定せず、無期限に実行します。
• ジョブモード — 1 回限りのバックフィルです。job: true と上限を設定します。EVM ではソースの filter に block_number の上限を設定します（これによりソースにも上限が設定されます）。Solana では end_block を設定します。

ステップ 8：生成、検証、提示

完全なパイプライン YAML を組み立てます。命名規則に従い、内容がわかる名前を付けます：<chain>-<data>-<sink>（例：base-erc20-transfers-postgres）。

1. YAML ファイルをディスクに書き込みます（例：<pipeline-name>.yaml）。
2. YAML をユーザーに提示する前に、検証を実行します。

goldsky turbo validate <pipeline-name>.yaml

3. 検証に失敗した場合は問題を修正して再検証します。検証に合格するまで YAML を提示してはいけません。よくある修正項目：

   • データセットソースの version フィールドがない
   • データセット名が無効（チェーンのプレフィックスを確認してください）
   • データベースシンクの secret_name がない
   • 変換の SQL 構文エラー

4. 検証に合格したら、確認用に YAML 全体をユーザーに提示します。

ステップ 9：デプロイ

ユーザーが YAML に問題がないことを確認した後：

goldsky turbo apply <pipeline-name>.yaml

ステップ 10：検証

デプロイ後：

goldsky turbo list

データフローを検証するため、inspect の実行を提案します。

goldsky turbo inspect <pipeline-name> -p

特定のノードに絞り込むには、goldsky turbo inspect <pipeline-name> -n <node-name> -p を実行します。

概要を提示します。

## Pipeline Deployed

**Name:** [name]
**Chain:** [chain]
**Dataset:** [dataset]
**Sink:** [sink type]
**Mode:** [streaming/job]

**Next steps:**
- Verify data flow with `goldsky turbo inspect <name> -p`
- Check logs with `goldsky turbo logs <name>`
- Use /turbo-doctor if you run into issues

重要なルール

• 完全な YAML をユーザーに提示する前に、必ず検証してください。検証されていない完全なパイプライン YAML は絶対に表示しないでください。
• デプロイ前に、必ず検証してください。
• デプロイ前に、必ず完全な YAML をユーザーに表示してください。
• ジョブモードのパイプラインについては、完了から約 1 時間後に自動クリーンアップされることをユーザーに伝えてください。
• 実際の宛先に書き込まずにパイプラインをテストする場合は、blackhole シンクを使用してください。
• ユーザーが既存のパイプラインの変更を希望する場合は、ストリーミング型（その場で更新）か、ジョブモード（先に削除が必要）かを確認してください。
• 明示的な開始位置を指定せずにデータセットソースを出力してはいけません。start_at: earliest をデフォルトにしてはいけません。開始位置（現在から／履歴の特定の時点から／全履歴）を確認し、全履歴を選んだ場合は、ライブデータの前にチェーンの履歴全体を再生することを警告し、シンクに十分な空き容量があることを確認してください。
• 数か月分または全履歴のバックフィル用に、512 MB の無料枠データベースをシンクとして推奨してはいけません。
• データセットソースには必ず version: 1.0.0 を含めてください。

関連項目

• /turbo-pipelines — YAML 設定とアーキテクチャのリファレンス
• /turbo-doctor — パイプラインの問題の診断と修正
• /turbo-operations — ライフサイクルコマンドとモニタリングのリファレンス
• /turbo-transforms — SQL および TypeScript 変換のリファレンス
• /datasets — データセット名とチェーンのプレフィックス
• /secrets — シンクの認証情報管理