ストリームへの書き込み

クライアント向けアプリケーションや外部システムから、Worker バインディング または HTTP エンドポイントを使って、ストリームにイベントを送信します。

Worker バインディングを使うと、API トークンや認証情報を管理せずに、Workers からストリームへ安全にデータを送れます。

Wrangler ファイルに、ストリームを指すパイプラインバインディングを追加します。

{
	"pipelines": [
		{
			"binding": "STREAM",
			"stream": "<STREAM_ID>"
		}
	]
}

[[pipelines]]
binding = "STREAM"
stream = "<STREAM_ID>"

パイプラインバインディングは、ストリームへデータを送るメソッドを公開します。

JSON シリアライズ可能なレコードの配列をストリームへ送信します。レコードの取り込みが確認されると解決する Promise を返します。

export default {
	async fetch(request, env, ctx) {
		const events = await request.json();

		await env.STREAM.send(events);

		return new Response("Events sent");
	},
};

export default {
	async fetch(request, env, ctx): Promise<Response> {
		const events = await request.json<Record<string, unknown>[]>();

		await env.STREAM.send(events);

		return new Response("Events sent");
	},
} satisfies ExportedHandler<Env>;

ストリームにスキーマが定義されている場合、wrangler types を実行すると、パイプラインバインディング向けにスキーマ固有の TypeScript 型が生成されます。汎用の Pipeline<PipelineRecord> ではなく、オートコンプリートとコンパイル時の型チェックが使える名前付きレコード型になります。詳細は wrangler types のドキュメント を参照してください。

wrangler types を実行すると、生成された worker-configuration.d.ts に、Cloudflare 名前空間内の名前付きレコード型が含まれます。型名はバインディング名ではなくストリーム名から作られ、PascalCase に Record サフィックスを付けます。

次は、ecommerce_stream という名前のストリームに対して、worker-configuration.d.ts に生成される型の例です。

declare namespace Cloudflare {
	type EcommerceStreamRecord = {
		user_id: string;
		event_type: string;
		product_id?: string;
		amount?: number;
	};
	interface Env {
		STREAM: import("cloudflare:pipelines").Pipeline<Cloudflare.EcommerceStreamRecord>;
	}
}

wrangler types は、次の場合に汎用の Pipeline<PipelineRecord> 型へフォールバックします。

• 未認証: 型付きパイプラインバインディングを使うには、wrangler login を実行します。
• ストリームが見つからない: Wrangler 設定のストリーム ID が、既存のストリームと一致しません。
• 非構造化ストリーム: スキーマなしで作成されたストリームです。

各ストリームは、外部アプリケーション、ブラウザー、HTTP リクエストを出せる任意のシステムからデータを取り込むための、任意の HTTP エンドポイントを提供します。

HTTP エンドポイントは次の形式です。

https://{stream-id}.ingest.cloudflare.com

ダッシュボードの Pipelines > Streams、または Wrangler CLI でストリーム ID かストリーム名を指定して、ストリームのエンドポイント URL を確認できます。

npx wrangler pipelines streams get <STREAM_NAME_OR_ID>

イベントを JSON 配列として POST リクエストで送信します。

curl -X POST https://{stream-id}.ingest.cloudflare.com \
  -H "Content-Type: application/json" \
  -d '[
    {
      "user_id": "12345",
      "event_type": "purchase",
      "product_id": "widget-001",
      "amount": 29.99
    }
  ]'

ストリームで認証が有効な場合は、Authorization ヘッダーに API トークンを含めます。

curl -X POST https://{stream-id}.ingest.cloudflare.com \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -d '[{"event": "test"}]'

API トークンには Workers Pipeline Send 権限が必要です。詳細は、API トークンの作成 を参照してください。

ストリームは、設定に応じて検証の扱いが異なります。

• 構造化ストリーム: イベントは、定義済みスキーマのフィールドと型に一致する必要があります。
• 非構造化ストリーム: 有効な JSON 構造なら受け付けます。データは単一の value 列に保存されます。

構造化ストリームでは、イベントがスキーマ定義と一致することを確認してください。無効なイベントは受け付けられますが、破棄されます。破棄を避けるため、送信前にデータを検証してください。Worker バインディングを使う場合は、wrangler types を実行して 型付きパイプラインバインディング を生成すると、スキーマ違反をコンパイル時に検出できます。破棄されたイベントの監視とスキーマ検証の問題の切り分けには、ユーザーエラーメトリクス を照会できます。