Sink to Cloudflare Pipeline and R2 Bucket

Run the following command to create a new R2 bucket called sequin-r2-bucket:

npx wrangler r2 bucket create sequin-r2-bucket

Create a pipeline connected to the R2 bucket you just created:

npx wrangler pipelines create sequin-pg-sink --r2-bucket sequin-r2-bucket --batch-max-seconds 5 --compression none

This will create a new pipeline called sequin-pg-sink with a simple configuration to get you started. You’ll see the following output:

Id:    {{unique-id}}
Name:  sequin-pg-sink
Sources:
HTTP:
    Endpoint:        https://{{unique-id}}.pipelines.cloudflare.com
    Authentication:  off
    Format:          JSON
Worker:
    Format:  JSON
Destination:
Type:         R2
Bucket:       sequin-r2-bucket
Format:       newline-delimited JSON
Compression:  NONE
Batch hints:
    Max bytes:     100 MB
    Max duration:  5 seconds
    Max records:   10,000,000

🎉 You can now send data to your pipeline!

Copy down the Endpoint URL, you’ll need this to setup the Sequin Webhook Sink.

In the Sequin console, navigate to the Sink page and click “Create Sink”.

Then, select “Webhook” as the sink type.

Select the the table(s) you want to sink (i.e. public.users).

The Cloudflare pipeline expects JSON objects wrapped in a single array. To achieve this, you’ll need to add a transform to the sink.

Skip down to the “Transform” section and click the + Create new transform button.

In the “New Function” page that opens, name your function cloudflare-pipeline-transform with the type set to Transform function. Then, add the following code to the function:

def transform(action, record, changes, metadata) do
 [%{
     action: action,
     record: record,
     changes: changes,
     metadata: metadata
 }]
end

This function will wrap each JSON object in a single array as required by the Cloudflare pipeline.

Click “Create Function” to save the function.

Navigate back to the Webhook Sink form, click the refresh button in the “Transform” section and select the cloudflare-pipeline-transform function you just created.

In the “Webhook Sink configuration” section, turn off batching.

Skip down to the “HTTP Endpoint” section, click New HTTP Endpoint and in the new window, name the endpoint (e.g. cloudflare-pipeline-endpoint) and enter the url for your pipeline:

https://{{unique-id}}.pipelines.cloudflare.com

No need to add any headers or authentication right now.

Then, go back to your Webhook Sink form, refresh the list of endpoints, and select the endpoint you just created.

Give your webhook sink a name (e.g. cloudflare-pipeline-sink) and click “Save” to save the webhook sink.

Create a new record in the table. For example, you can run the following SQL command to create a new user in the database:

insert into users (first_name, last_name, email) values ('Eric', 'Goldman', 'eric@sequinstream.com');

Using the Cloudflare dashboard, you can check the R2 bucket for the new user:

And if you download the object, you’ll see the change as a single JSON object:

{
"record": {
    "created_at": "2025-06-05T02:04:58.577494Z",
    "email": "eric@sequinstream.com",
    "first_name": "Eric",
    "id": "666a0182-cde7-4fcf-8fc2-2561536a35bf",
    "last_name": "Goldman",
    "updated_at": "2025-06-05T02:04:58.577494Z"
},
"metadata": {
    "database_name": "withered_feather_4",
    "commit_idx": 0,
    "commit_lsn": 35003912,
    "commit_timestamp": "2025-06-05T02:04:58.584143Z",
    "table_name": "users",
    "table_schema": "public",
    "transaction_annotations": null,
    "idempotency_key": "MzUwMDM5MTI6MA==",
    "consumer": {
    "id": "c1905648-968c-4d73-8a1e-6cb2eb59a6a3",
    "name": "cloudflare-pipeline-sink",
    "annotations": {}
    }
},
"action": "insert",
"changes": null
}