create-connections

接続の作成

目的

選択したストリームをソースから組織のデータウェアハウスに取り込むため、connections/<name>.yaml に Connection YAML を作成します。各接続は単独で完結するドキュメントです。revos apply はこれを読み取り、API と照合して差分を反映します。

難しいのは YAML の形式ではなく、ストリームの選択と同期モードの選択です。ほとんどのソースでは、プロジェクトで実際に必要とする数をはるかに上回るストリームが公開されています。特にデータベースではその傾向が顕著です。慎重に選び、ユーザーに確認してから作成してください。

---

前提条件

• ソースがすでにサーバー上に存在している必要があります（revos sources list で id を取得します）。まだ作成されていないソースからの取り込みをユーザーが希望している場合は、いったん中断し、まず revos sources create を実行するよう伝えてください。ソースの設定（コネクターの選択、認証情報、OAuth）は RevOS UI で行います。UI から戻ったことをユーザーが確認したら、続行する前に revos connections list --json と revos sources list --json を再実行してください。UI によって新しいソース用の接続が自動作成されている場合があります。一致する接続がすでに存在する場合は、重複を作成するのではなく、revos pull で取得することを提案してください。
• revos auth status で認証済みであることを確認します。未認証の場合は、revos auth login を実行するようユーザーに依頼してください。

---

ユーザー確認ポイント

このスキルでは、明示的な確認が有益な判断を2つ行います。必ず確認してください。ストリームや同期モードを誤ると、データウェアハウスのストレージを無駄にし、再読み込みに時間がかかります。

確認ポイント 1：ストリームの選択

ストリームを調べ、ユーザーにユースケースを尋ねた後、各ストリームについて1行の理由を添えて候補を提示します。次に進む前に、確認または修正を待ってください。

確認ポイント 2：同期モードとキーの選択

選択した各ストリームの同期モード、カーソル、主キーを決めた後、表を提示して、ファイルを書き込む前に確認します。ほとんどのユーザーはざっと目を通して承認しますが、ストリームを full_refresh_overwrite に変更したり、カーソルを変更したりする場合もあります。

---

ワークフロー

フェーズ 1：ソースの特定

まず revos connections list --json を実行し、既存の接続を確認します。ユーザーが Web UI から戻った直後の場合は、この再取得が不可欠です。UI によって接続が自動作成されている場合があります。ユーザーが示した目的に一致する接続が見つかったら、重複を作成しないよう、先にその接続を提示してください。

次に revos sources list --json を実行し、サーバー側の name と照合します（一覧性の高い表で確認するには revos sources list を実行します）。ソースの id を記録します。この値を Connection YAML の spec.source.id に設定します。

ソースが指定されていない場合は、ユーザーにどのソースを使うか尋ねてください。id がない状態では先に進まないでください。

フェーズ 2：ストリームの確認

revos sources list-streams <id> --json

次のような形のオブジェクト配列が返されます。

{
  "streamName": "customers",
  "streamnamespace": "public",
  "syncModes": ["full_refresh_overwrite", "incremental_deduped_history", ...],
  "defaultCursorField": ["updated_at"],
  "sourceDefinedCursorField": false,
  "sourceDefinedPrimaryKey": [["id"]],
  "propertyFields": [["id"], ["email"], ["updated_at"], ...]
}

レスポンス全体を保存してください。後続のフェーズで参照します。syncModes 配列には、プロジェクトの送信先でそのストリームに対して有効な同期モードが絞り込まれています。この一覧にないモードは提案しないでください。

レスポンスの形式について、知っておくべき点が2つあります。

• データベースソースでは、インクリメンタル同期モードをサポートしていても、defaultCursorField が空のままであることがよくあります。 すべての列を propertyFields で公開し、列を選択できるようにしています。データベースソースでは propertyFields を使う想定にしてください。SaaS ソースでは通常、適切なフィールドが defaultCursorField に事前設定されています。
• データベースでは streamnamespace が設定されます（例：public、dbo）。一方、ほとんどの SaaS ソースでは設定されません。検出レスポンスに streamnamespace が含まれている場合は、YAML に namespace: の行を設定する必要があります。含まれていないストリームでは、その行を省略してください。

フェーズ 3：ユースケースについて尋ねる

ストリームを提案する前に、選択の基準となる短い質問を1つ尋ねます。例：

• 「この接続で何を分析したいですか？ 例：顧客あたりの売上、サポートチケットの傾向、マーケティングファネル。」
• 「ここではソースのどの部分が重要ですか？ 営業パイプライン、財務記録、製品の利用状況などです。」

自由回答形式にします。ユーザーの回答は1文で十分です。目的は、明らかに関係のないストリーム（監査ログ、内部キュー、システムテーブル）を除外し、ビジネスエンティティを優先することです。ユーザーが最初のリクエストですでに目的を示している場合に限り、この質問を省略します。

フェーズ 4：ストリームのサブセットを提案する（チェックポイント 1）

次のルールを順番に適用します。

1. ユーザーの目的で明確に必要とされていない限り、技術系／システム系のストリームを除外します。 次のいずれかに一致する名前：pg_、information_schema、_airbyte、temp_、tmp_、audit_、system_、migration、schema_migrations で始まるもの。または _history、_log、_audit、_archive で終わるものです。慎重に判断し、迷った場合は含めて、その旨を明記します。

2. データベースの場合（多数のストリームがあり、public、dbo、sales などのネームスペースがあることで判別できる、postgres／mysql／mssql／mongodb 形式のソース）：ストリームの30～80％を除外することを想定します。プロジェクトのユースケースで、運用テーブルをすべて必要とすることはほとんどありません。ユーザーの目的に合致する名前のストリーム（売上分析なら orders、customers、products、サポートなら tickets、contacts、messages）を優先します。

3. SaaS ソースの場合（少数のフラットなストリーム一覧で、ネームスペースがなく、companies、deals、tickets、engagements のような名前）：基本的に、主要なビジネスエンティティをすべて含めます。明らかに不要なもの（*_metadata、*_history、*_changelog）のみを除外します。

4. 不確かな点は正直に伝えます。 ストリーム名から内容を判断できない場合は、その旨を伝えます。理由を捏造してはいけません。

ユーザーが一覧を確認しやすいよう、提案を表として提示します。

Proposed streams (12 of 47):

  customers         core entity — revenue analysis needs this
  orders            transactions table; cursor: updated_at
  order_items       order line items; needed to break revenue by SKU
  products          dimension table for orders/order_items
  ...

Dropped (35): pg_stat_*, _airbyte_internal_*, audit_*, schema_migrations,
              user_sessions (not in scope for revenue analysis), ...

「よさそうですか？ 追加または削除したいものはありますか？」と尋ね、ユーザーの返答を待ちます。「users を追加して」や「products は不要なので削除して」などの返答に応じて調整します。

フェーズ 5：ストリームごとの同期モード、カーソル、主キーを決定する

選択した各ストリームについて、選択はソースがサポートする内容（syncModes）と、ソースが提示するフィールドによって決まります。使用する同期モードは次の 4 つです。

同期モード	必要なもの	使用する場面
incremental_deduped_history	カーソル と 主キー	デフォルトとして最適です。ソースがカーソル（例：updated_at）と主キーを提示しています。新規または変更された行のみを取得し、主キーで重複を統合します。
incremental_append	カーソルのみ	ソースにカーソルはありますが、主キーがありません（イベントストリーム、追記専用ログなど）。行は蓄積され、重複排除は行われません。
full_refresh_overwrite_deduped	主キーのみ	ソースに主キーはありますが、利用可能なカーソルがありません。同期のたびに宛先テーブルを置き換え、主キーで重複排除します。小規模から中規模のテーブルに適しています。
full_refresh_overwrite	不要	小規模なディメンションテーブル（1 万行未満）、または他の方法が使えない場合に使用します。同期のたびに上書きします。

ストリームごとの決定アルゴリズム：

1. ソースがサポートする最も強力なモードを、次の優先順位で選びます：incremental_deduped_history → incremental_append → full_refresh_overwrite_deduped → full_refresh_overwrite。ストリームの syncModes 配列にないモードはスキップします。
2. カーソルが必要なモードでは、空でなければ defaultCursorField を使用します。それ以外の場合は、propertyFields の中から最初に見つかったタイムスタンプらしいフィールド（updated_at、modified_at、last_modified、_updated、timestamp）を確認します。適切なものがなければ、フルリフレッシュモードに切り替えます。
3. 主キーが必要なモードでは、空でなければ sourceDefinedPrimaryKey を使用します。それ以外の場合は、propertyFields から id に相当するフィールド（id、<entity>_id、uuid、key）を探します。適切なものがなければ、incremental_append または full_refresh_overwrite に切り替えます。

提示する際は簡潔な表記を使用します（チェックポイント 2）：

Stream             Sync mode                       Cursor        Primary key
customers          incremental_deduped_history     updated_at    id
orders             incremental_deduped_history     updated_at    id
order_items        incremental_append              created_at    —
products           full_refresh_overwrite          —             —

「同期モードは適切ですか？変更はありますか？」と尋ねます。よくある変更には、大規模なテーブルを full_refresh_overwrite のみにする（小規模または静的な場合のみ）ことや、カーソルフィールドの切り替えがあります。

フェーズ 6：YAML を記述する

異なるルールで選ぶ名前が 2 つあります。

• metadata.name — ローカル IaC 用のスラッグです。ファイル名に使いやすいよう、小文字、英数字、ハイフンのみを使用します。YAML ファイル（connections/<metadata.name>.yaml）の名前となり、他のリソースがこの接続を参照するときにも使われます。ソースとスコープに基づいて付けてください：hubspot-sales、postgres-prod-revenue、stripe-billing。
• spec.name — RevOS UI に表示される、人が読みやすいラベルです。短く、文頭のみ大文字にし、スペースは使用できますが、アンダースコアは使わないでください。UI をざっと確認するチームメイトが開かなくても分かるよう、この接続が同期する内容を表してください。適切な例：「HubSpot sales pipeline」、「Postgres prod → revenue tables」、「Stripe billing & subscriptions」などです。ここでスラッグを再利用するのは避けてください。hubspot-sales より HubSpot sales pipeline のほうが spec.name に適しています。

指定したパスにすでにファイルがある場合は、上書きするか、別の名前を選ぶかを確認してください。

テンプレート：

apiVersion: revos/v1
kind: Connection
metadata:
  name: hubspot-sales
spec:
  name: HubSpot sales pipeline
  source: { id: <id from `revos sources list`> }
  schedule: { units: 24, timeUnit: hours }
  status: active
  streams:
    - name: customers
      namespace: public # omit if the stream has no namespace
      syncMode: incremental_deduped_history
      cursorField: [updated_at]
      primaryKey: [[id]]

注意事項：

• source.id は Source のサーバー ID です（例：src_abc123）。Sources は IaC ではありません。revos sources list で ID を取得してください。
• primaryKey はリストのリストで、内側の各リストが 1 つの PK 列のパスを表し、ネストされたキーをサポートします。ほとんどの場合は [[id]] です。
• cursorField はパスセグメントのフラットなリストです。トップレベルのフィールドは [updated_at]、ネストされたフィールドは [meta, modified_at] です。
• full_refresh_overwrite の場合は cursorField と primaryKey を省略してください。full_refresh_overwrite_deduped の場合は cursorField のみを省略してください。CLI の検証では、適用時に必須フィールドが不足していると拒否されるため、設定を適切に保ってください。
• metadata.id や spec.prefix は設定しないでください。どちらも初回作成時に revos apply が入力します。

ストリームマッパー（streams[].mappers）は、行が BigQuery に格納される前に適用されるサーバー側の変換です。PII のハッシュ化、列名の変更、フィールドの削除、行のフィルタリング、値の暗号化などを行います。デフォルトでは、マッパーを使わずにクリーンな同期を行ってください。次のいずれかに該当する場合は、references/mappers.md を読み込んでください。

• ユーザーがマスキング、ハッシュ化、PII、列名の変更、列の削除、行のフィルタリング、暗号化に言及した場合。
• 選択したストリームの propertyFields に、明らかに機密性の高い列（email、phone、ssn、ip_address、氏名を構成する項目の組み合わせ、決済関連フィールド）が含まれている場合。その場合はリファレンスを読み込んだうえで、フィールド名と理由を示してマッパーを積極的に提案し、追加する前にユーザーの確認を求めてください。マッパーを黙って追加しないでください。誤ったフィールドをマスキングすると分析が破損します。

フェーズ 7：ローカルで検証する

revos diff

CLI は zod を使って YAML を解析し、どのような変更が行われるかを報告します。次の点を確認してください。

• 解析エラー — スキーマによる拒否（必須フィールドの不足、無効な同期モードなど）。修正して再実行してください。
• ドリフトレポート — 新しい接続が単一の create エントリとして表示されるはずです。

revos diff の結果に問題がなければ、次のように伝えてください：「接続 YAML を connections/<slug>.yaml に書き込み、検証しました。revos apply を実行してサーバー上に作成してください。」

検証に失敗した場合は、エラーをそのまま共有し、成功を宣言する前に修正してください。

---

よくある落とし穴

• ソースがサポートしていない同期モードを選ぶ。 常にフェーズ 2 のストリームの syncModes 配列と選択内容を照合してください。API はサーバー側で検証しますが、ローカルで事前に確認するのが得策です。
• サーバー ID が必要な場所にソース名を使う。 spec.source.id はサーバー ID（例：src_abc123）であり、表示名やスラッグではありません。必ず revos sources list から取得してください。
• full_refresh_* モードで cursorField を設定する。 CLI の検証では余分なフィールドが許容されますが、不要な情報です。選択したモードに適用されないフィールドは省略してください。
• すべてのストリームを提案する。 特にデータベースの場合は注意してください。提案を見た後でユーザーが「すべて」と言った場合は問題ありませんが、まずは厳選したリストを提示してください。
• フィールドをでっち上げる。 検出レスポンス（defaultCursorField、sourceDefinedPrimaryKey、または propertyFields）に表示されている値のみを cursorField と primaryKey に使用してください。