Configuration

{
  "fields": [
    { "name": "event_id", "type": "string", "required": true },
    { "name": "amount", "type": "float64", "required": false }
  ]
}

npx wrangler basin pipelines setup   # creates stream + sink + pipeline, optionally bucket + catalog

# 1. Stream
npx wrangler basin pipelines streams create my_stream --schema-file schema.json

# 2. Sink — Basin Catalog (Iceberg). Creates the namespace + table.
npx wrangler basin pipelines sinks create my_sink \
  --type basin-catalog \
  --bucket my-bucket --namespace my_namespace --table my_table \
  --catalog-token $API_TOKEN \
  --compression zstd --roll-interval 300

# 2b. Sink — R2 raw Parquet (alternative)
npx wrangler basin pipelines sinks create my_sink \
  --type r2 --bucket my-bucket --format parquet \
  --path analytics/events --partitioning "year=%Y/month=%m/day=%d" \
  --access-key-id $KEY --secret-access-key $SECRET

# 3. Pipeline (SQL connects stream → sink)
npx wrangler basin pipelines create my_pipeline \
  --sql "INSERT INTO my_sink SELECT * FROM my_stream"

# Stream
curl -X POST "$BASE_URL/streams" -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" -d '{
    "name": "my_stream",
    "http": {"enabled": true, "authentication": false},
    "schema": {"fields": [{"name": "event_id", "type": "string", "required": true}]}
  }'

# Sink — NOTE REST field names differ from CLI flags (see table)
curl -X POST "$BASE_URL/sinks" -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" -d '{
    "name": "my_sink", "type": "r2_data_catalog",
    "config": {"bucket": "my-bucket", "namespace": "my_namespace",
               "table_name": "my_table", "token": "'$API_TOKEN'",
               "rolling_policy": {"interval_seconds": 300}},
    "format": {"type": "parquet"}
  }'

# Pipeline
curl -X POST "$BASE_URL/pipelines" -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name": "my_pipeline", "sql": "INSERT INTO my_sink SELECT * FROM my_stream;"}'

// wrangler.jsonc
{ "pipelines": [ { "stream": "<STREAM_ID>", "binding": "MY_STREAM" } ] }

resource "cloudflare_pipeline_stream" "my_stream" {
  account_id     = var.cloudflare_account_id
  name           = "my_stream"
  format         = { type = "json" }
  schema         = { fields = [{ name = "value", type = "json", required = true }] }
  http           = { enabled = true, authentication = false, cors = {} }
  worker_binding = { enabled = false }
}

resource "cloudflare_pipeline_sink" "my_sink" {
  account_id = var.cloudflare_account_id
  name       = "my_sink"
  type       = "r2_data_catalog"
  format     = { type = "parquet" }
  schema     = { fields = [] }
  config     = {
    account_id = var.cloudflare_account_id
    bucket     = cloudflare_r2_bucket.pipeline_bucket.name
    table_name = "my_table"
    token      = var.catalog_token
  }
}

resource "cloudflare_pipeline" "my_pipeline" {
  account_id = var.cloudflare_account_id
  name       = "my_pipeline"
  sql        = "INSERT INTO ${cloudflare_pipeline_sink.my_sink.name} SELECT * FROM ${cloudflare_pipeline_stream.my_stream.name}"
}