comi-logic/cfw-autumn
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #841 from useautumn/feat/tinybird-cdc-sequin

Browse Source
This commit is contained in:
amianthus
2026-02-26 21:58:58 +00:00
committed by GitHub
parent 16a6c7f012 fa2d3adeb3
commit ac60e2af4f
36 changed files with 2029 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

255
others/sequin/tinybird_mega_transform.ex Normal file
View File

@@ -0,0 +1,255 @@
def transform(action, record, changes, metadata) do
ts = metadata.commit_timestamp
lsn = metadata.commit_lsn
ikey = metadata.idempotency_key
cdc = %{__action: action, __commit_timestamp: ts, __commit_lsn: lsn, __idempotency_key: ikey}
di = fn v -> if v == nil, do: nil, else: Decimal.to_integer(v) end
df = fn v -> if v == nil, do: nil, else: Decimal.to_float(v) end
bi = fn v -> if v, do: 1, else: 0 end
r = record
Map.merge(
cdc,
case metadata.table_name do
"customers" ->
%{
internal_id: r["internal_id"],
org_id: r["org_id"],
created_at: di.(r["created_at"]),
name: r["name"],
id: r["id"],
email: r["email"],
fingerprint: r["fingerprint"],
metadata: r["metadata"] |> JSON.encode!(),
env: r["env"],
processor: r["processor"] |> JSON.encode!(),
processors: (r["processors"] || %{}) |> JSON.encode!(),
send_email_receipts: bi.(r["send_email_receipts"])
}
"invoices" ->
%{
id: r["id"],
created_at: di.(r["created_at"]),
product_ids: r["product_ids"] || [],
internal_product_ids: r["internal_product_ids"] || [],
internal_customer_id: r["internal_customer_id"],
internal_entity_id: r["internal_entity_id"],
stripe_id: r["stripe_id"],
status: r["status"],
hosted_invoice_url: r["hosted_invoice_url"],
total: df.(r["total"]),
currency: r["currency"],
discounts: r["discounts"] |> JSON.encode!(),
items: r["items"] |> JSON.encode!()
}
"organizations" ->
%{
id: r["id"],
slug: r["slug"],
name: r["name"],
logo: r["logo"],
created_at_ts: if(r["createdAt"] == nil, do: nil, else: to_string(r["createdAt"])),
metadata: r["metadata"],
default_currency: r["default_currency"] || "usd",
stripe_connected: bi.(r["stripe_connected"]),
created_at: di.(r["created_at"]),
onboarded: bi.(r["onboarded"]),
deployed: bi.(r["deployed"])
}
"customer_products" ->
%{
id: r["id"],
internal_customer_id: r["internal_customer_id"],
internal_product_id: r["internal_product_id"],
internal_entity_id: r["internal_entity_id"],
created_at: di.(r["created_at"]),
status: r["status"],
processor: r["processor"] |> JSON.encode!(),
canceled: bi.(r["canceled"]),
canceled_at: di.(r["canceled_at"]),
ended_at: di.(r["ended_at"]),
starts_at: di.(r["starts_at"]),
options: (r["options"] || []) |> JSON.encode!(),
product_id: r["product_id"],
free_trial_id: r["free_trial_id"],
trial_ends_at: di.(r["trial_ends_at"]),
collection_method: r["collection_method"] || "charge_automatically",
subscription_ids: r["subscription_ids"] || [],
scheduled_ids: r["scheduled_ids"] || [],
quantity: if(r["quantity"] == nil, do: 1.0, else: Decimal.to_float(r["quantity"])),
is_custom: bi.(r["is_custom"]),
customer_id: r["customer_id"],
entity_id: r["entity_id"],
billing_version: r["billing_version"],
api_version: df.(r["api_version"]),
api_semver: r["api_semver"]
}
"customer_entitlements" ->
%{
id: r["id"],
customer_product_id: r["customer_product_id"],
entitlement_id: r["entitlement_id"],
internal_customer_id: r["internal_customer_id"],
internal_entity_id: r["internal_entity_id"],
internal_feature_id: r["internal_feature_id"],
unlimited: bi.(r["unlimited"]),
balance: if(r["balance"] == nil, do: 0.0, else: Decimal.to_float(r["balance"])),
created_at: di.(r["created_at"]),
next_reset_at: di.(r["next_reset_at"]),
usage_allowed: bi.(r["usage_allowed"]),
adjustment: df.(r["adjustment"]),
additional_balance:
if(r["additional_balance"] == nil,
do: 0.0,
else: Decimal.to_float(r["additional_balance"])
),
entities: (r["entities"] || %{}) |> JSON.encode!(),
expires_at: di.(r["expires_at"]),
cache_version: r["cache_version"] || 0,
customer_id: r["customer_id"],
feature_id: r["feature_id"]
}
"customer_prices" ->
%{
id: r["id"],
created_at: di.(r["created_at"]),
price_id: r["price_id"],
options: r["options"] |> JSON.encode!(),
internal_customer_id: r["internal_customer_id"],
customer_product_id: r["customer_product_id"]
}
"replaceables" ->
%{
id: r["id"],
cus_ent_id: r["cus_ent_id"],
created_at: r["created_at"],
from_entity_id: r["from_entity_id"],
delete_next_cycle: bi.(r["delete_next_cycle"])
}
"rollovers" ->
%{
id: r["id"],
cus_ent_id: r["cus_ent_id"],
balance: if(r["balance"] == nil, do: 0.0, else: Decimal.to_float(r["balance"])),
expires_at: di.(r["expires_at"]),
usage: if(r["usage"] == nil, do: 0.0, else: Decimal.to_float(r["usage"])),
entities: (r["entities"] || %{}) |> JSON.encode!()
}
"entitlements" ->
%{
id: r["id"],
created_at: di.(r["created_at"]),
internal_feature_id: r["internal_feature_id"],
internal_product_id: r["internal_product_id"],
is_custom: bi.(r["is_custom"]),
allowance_type: r["allowance_type"],
allowance: df.(r["allowance"]),
interval: r["interval"],
interval_count:
if(r["interval_count"] == nil, do: 1.0, else: Decimal.to_float(r["interval_count"])),
carry_from_previous: bi.(r["carry_from_previous"]),
entity_feature_id: r["entity_feature_id"],
org_id: r["org_id"],
feature_id: r["feature_id"],
usage_limit: df.(r["usage_limit"]),
rollover: r["rollover"] |> JSON.encode!()
}
"free_trials" ->
%{
id: r["id"],
created_at: di.(r["created_at"]),
internal_product_id: r["internal_product_id"],
duration: r["duration"] || "day",
length: df.(r["length"]),
unique_fingerprint: bi.(r["unique_fingerprint"]),
is_custom: bi.(r["is_custom"]),
card_required: bi.(r["card_required"])
}
"entities" ->
%{
id: r["id"],
org_id: r["org_id"],
created_at: di.(r["created_at"]),
internal_id: r["internal_id"],
internal_customer_id: r["internal_customer_id"],
env: r["env"],
name: r["name"],
deleted: bi.(r["deleted"]),
internal_feature_id: r["internal_feature_id"],
feature_id: r["feature_id"]
}
"features" ->
%{
internal_id: r["internal_id"],
org_id: r["org_id"],
created_at: di.(r["created_at"]),
env: r["env"],
id: r["id"],
name: r["name"],
type: r["type"],
config: r["config"] |> JSON.encode!(),
display: r["display"] |> JSON.encode!(),
archived: bi.(r["archived"]),
event_names: r["event_names"] || []
}
"prices" ->
%{
id: r["id"],
org_id: r["org_id"],
internal_product_id: r["internal_product_id"],
config: r["config"] |> JSON.encode!(),
created_at: di.(r["created_at"]),
billing_type: r["billing_type"],
tier_behavior: r["tier_behavior"],
is_custom: bi.(r["is_custom"]),
entitlement_id: r["entitlement_id"],
proration_config: r["proration_config"] |> JSON.encode!()
}
"products" ->
%{
internal_id: r["internal_id"],
id: r["id"],
name: r["name"],
description: r["description"],
org_id: r["org_id"],
created_at: di.(r["created_at"]),
env: r["env"],
is_add_on: bi.(r["is_add_on"]),
is_default: bi.(r["is_default"]),
group: r["group"],
version: if(r["version"] == nil, do: 1.0, else: Decimal.to_float(r["version"])),
processor: r["processor"] |> JSON.encode!(),
base_variant_id: r["base_variant_id"],
archived: bi.(r["archived"])
}
"subscriptions" ->
%{
id: r["id"],
org_id: r["org_id"],
stripe_id: r["stripe_id"],
stripe_schedule_id: r["stripe_schedule_id"],
created_at: di.(r["created_at"]),
metadata: (r["metadata"] || %{}) |> JSON.encode!(),
usage_features: r["usage_features"] || [],
env: r["env"],
current_period_start: di.(r["current_period_start"]),
current_period_end: di.(r["current_period_end"])
}
end
)
end

24
others/sequin/tinybird_routing.ex Normal file
View File

@@ -0,0 +1,24 @@
def route(action, record, changes, metadata) do
datasource = case metadata.table_name do
"customers" -> "customers"
"invoices" -> "invoices"
"organizations" -> "organizations"
"customer_products" -> "customer_products"
"customer_entitlements" -> "customer_entitlements"
"customer_prices" -> "customer_prices"
"replaceables" -> "replaceables"
"rollovers" -> "rollovers"
"entitlements" -> "entitlements"
"free_trials" -> "free_trials"
"entities" -> "entities"
"subscriptions" -> "subscriptions"
"features" -> "features"
"prices" -> "prices"
"products" -> "products"
end
%{
method: "POST",
endpoint_path: "?name=#{datasource}&format=json"
}
end

68
server/tinybird/copies/customer_entitlements_backfill.pipe Normal file
View File

@@ -0,0 +1,68 @@
DESCRIPTION >
Chunked backfill of customer_entitlements from Postgres into customer_entitlements.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE raw
SQL >
%
SELECT
id,
customer_product_id,
entitlement_id,
internal_customer_id,
internal_entity_id,
internal_feature_id,
unlimited,
balance,
created_at,
next_reset_at,
usage_allowed,
adjustment,
additional_balance,
entities,
expires_at,
cache_version,
customer_id,
feature_id
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'customer_entitlements',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
AND balance < 1e15
AND additional_balance < 1e15
NODE migrate
SQL >
SELECT
assumeNotNull(id) AS id,
customer_product_id,
coalesce(entitlement_id, '') AS entitlement_id,
coalesce(internal_customer_id, '') AS internal_customer_id,
internal_entity_id,
coalesce(internal_feature_id, '') AS internal_feature_id,
toUInt8(if(coalesce(unlimited, false), 1, 0)) AS unlimited,
toFloat64OrZero(toString(coalesce(balance, toDecimal128(0, 19)))) AS balance,
created_at::Int64 AS created_at,
if(next_reset_at IS NULL, NULL, next_reset_at::Int64) AS next_reset_at,
toUInt8(if(coalesce(usage_allowed, false), 1, 0)) AS usage_allowed,
toFloat64OrNull(toString(adjustment)) AS adjustment,
toFloat64OrZero(toString(coalesce(additional_balance, toDecimal128(0, 19)))) AS additional_balance,
ifNull(toString(entities), '{}') AS entities,
if(expires_at IS NULL, NULL, expires_at::Int64) AS expires_at,
coalesce(cache_version, 0) AS cache_version,
customer_id,
feature_id,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', assumeNotNull(id)) AS __idempotency_key
FROM raw
TYPE COPY
TARGET_DATASOURCE customer_entitlements

31
server/tinybird/copies/customer_prices_backfill.pipe Normal file
View File

@@ -0,0 +1,31 @@
DESCRIPTION >
Chunked backfill of customer_prices from Postgres into customer_prices.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
assumeNotNull(id) AS id,
created_at::Int64 AS created_at,
price_id,
ifNull(toString(options), 'null') AS options,
internal_customer_id,
customer_product_id,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', assumeNotNull(id)) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'customer_prices',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
TYPE COPY
TARGET_DATASOURCE customer_prices

51
server/tinybird/copies/customer_products_backfill.pipe Normal file
View File

@@ -0,0 +1,51 @@
DESCRIPTION >
Chunked backfill of customer_products from Postgres into customer_products.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
assumeNotNull(id) AS id,
coalesce(internal_customer_id, '') AS internal_customer_id,
coalesce(internal_product_id, '') AS internal_product_id,
internal_entity_id,
created_at::Int64 AS created_at,
status,
ifNull(toString(processor), 'null') AS processor,
toUInt8(coalesce(canceled, false)) AS canceled,
if(canceled_at IS NULL, NULL, canceled_at::Int64) AS canceled_at,
if(ended_at IS NULL, NULL, ended_at::Int64) AS ended_at,
if(starts_at IS NULL, NULL, starts_at::Int64) AS starts_at,
if(options IS NULL, '[]', toJSONString(arrayMap(x -> assumeNotNull(x), options))) AS options,
product_id,
free_trial_id,
if(trial_ends_at IS NULL, NULL, trial_ends_at::Int64) AS trial_ends_at,
coalesce(collection_method, 'charge_automatically') AS collection_method,
arrayMap(x -> assumeNotNull(x), coalesce(subscription_ids, [])) AS subscription_ids,
arrayMap(x -> assumeNotNull(x), coalesce(scheduled_ids, [])) AS scheduled_ids,
coalesce(quantity, 1)::Float64 AS quantity,
toUInt8(if(is_custom, 1, 0)) AS is_custom,
customer_id,
entity_id,
billing_version,
if(api_version IS NULL, NULL, api_version::Int64) AS api_version,
api_semver,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', assumeNotNull(id)) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'customer_products',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
AND (quantity IS NULL OR quantity < 1e15)
TYPE COPY
TARGET_DATASOURCE customer_products

37
server/tinybird/copies/customers_backfill.pipe Normal file
View File

@@ -0,0 +1,37 @@
DESCRIPTION >
Chunked backfill of customers from Postgres into customers.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
coalesce(internal_id, '') AS internal_id,
coalesce(org_id, '') AS org_id,
created_at::Int64 AS created_at,
name,
id,
email,
fingerprint,
coalesce(toString(metadata), '{}') AS metadata,
coalesce(env, '') AS env,
coalesce(toString(processor), 'null') AS processor,
coalesce(toString(processors), '{}') AS processors,
if(coalesce(send_email_receipts, false) = true, 1, 0) AS send_email_receipts,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', coalesce(internal_id, '')) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'customers',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
TYPE COPY
TARGET_DATASOURCE customers

35
server/tinybird/copies/entities_backfill.pipe Normal file
View File

@@ -0,0 +1,35 @@
DESCRIPTION >
Chunked backfill of entities from Postgres into entities.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
id,
org_id,
created_at::Int64 AS created_at,
coalesce(internal_id, '') AS internal_id,
coalesce(internal_customer_id, '') AS internal_customer_id,
env,
name,
toUInt8(if(deleted, 1, 0)) AS deleted,
internal_feature_id,
feature_id,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', coalesce(internal_id, '')) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'entities',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
TYPE COPY
TARGET_DATASOURCE entities

42
server/tinybird/copies/entitlements_backfill.pipe Normal file
View File

@@ -0,0 +1,42 @@
DESCRIPTION >
Chunked backfill of entitlements from Postgres into entitlements.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
assumeNotNull(id) AS id,
created_at::Int64 AS created_at,
coalesce(internal_feature_id, '') AS internal_feature_id,
internal_product_id,
if(coalesce(is_custom, false) = true, 1, 0) AS is_custom,
allowance_type,
if(allowance IS NULL, NULL, allowance::Float64) AS allowance,
interval,
coalesce(interval_count, 1)::Float64 AS interval_count,
if(coalesce(carry_from_previous, false) = true, 1, 0) AS carry_from_previous,
entity_feature_id,
org_id,
feature_id,
if(usage_limit IS NULL, NULL, usage_limit::Float64) AS usage_limit,
ifNull(toString(rollover), 'null') AS rollover,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', assumeNotNull(id)) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'entitlements',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
AND (allowance IS NULL OR allowance < 1e15)
AND (usage_limit IS NULL OR usage_limit < 1e15)
TYPE COPY
TARGET_DATASOURCE entitlements

36
server/tinybird/copies/features_backfill.pipe Normal file
View File

@@ -0,0 +1,36 @@
DESCRIPTION >
Chunked backfill of features from Postgres into features.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
coalesce(internal_id, '') AS internal_id,
coalesce(org_id, '') AS org_id,
if(created_at IS NULL, NULL, created_at::Int64) AS created_at,
env,
coalesce(id, '') AS id,
name,
coalesce(type, '') AS type,
ifNull(toString(config), 'null') AS config,
ifNull(toString(display), 'null') AS display,
toUInt8(if(archived = true, 1, 0)) AS archived,
arrayMap(x -> assumeNotNull(x), coalesce(event_names, [])) AS event_names,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', coalesce(internal_id, '')) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'features',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
TYPE COPY
TARGET_DATASOURCE features

33
server/tinybird/copies/free_trials_backfill.pipe Normal file
View File

@@ -0,0 +1,33 @@
DESCRIPTION >
Chunked backfill of free_trials from Postgres into free_trials.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
assumeNotNull(id) AS id,
created_at::Int64 AS created_at,
internal_product_id,
coalesce(duration, 'day') AS duration,
if(length IS NULL, NULL, length::Float64) AS length,
toUInt8(if(coalesce(unique_fingerprint, false), 1, 0)) AS unique_fingerprint,
toUInt8(if(coalesce(is_custom, false), 1, 0)) AS is_custom,
toUInt8(if(coalesce(card_required, false), 1, 0)) AS card_required,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', assumeNotNull(id)) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'free_trials',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
TYPE COPY
TARGET_DATASOURCE free_trials

39
server/tinybird/copies/invoices_backfill.pipe Normal file
View File

@@ -0,0 +1,39 @@
DESCRIPTION >
Chunked backfill of invoices from Postgres into invoices.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
assumeNotNull(id) AS id,
created_at::Int64 AS created_at,
arrayMap(x -> assumeNotNull(x), product_ids) AS product_ids,
arrayMap(x -> assumeNotNull(x), internal_product_ids) AS internal_product_ids,
assumeNotNull(internal_customer_id) AS internal_customer_id,
internal_entity_id,
assumeNotNull(stripe_id) AS stripe_id,
assumeNotNull(status) AS status,
hosted_invoice_url,
total::Float64 AS total,
assumeNotNull(currency) AS currency,
coalesce(discounts::String, '[]') AS discounts,
coalesce(items::String, '[]') AS items,
'read' AS __action,
now() AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', id) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'invoices',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
AND (total IS NULL OR total < 1e15)
TYPE COPY
TARGET_DATASOURCE invoices

36
server/tinybird/copies/organizations_backfill.pipe Normal file
View File

@@ -0,0 +1,36 @@
DESCRIPTION >
Chunked backfill of organizations from Postgres into organizations.
Chunks by created_at (epoch ms). Cast numeric/timestamp columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
assumeNotNull(id) AS id,
assumeNotNull(slug) AS slug,
assumeNotNull(name) AS name,
logo,
createdAt::String AS created_at_ts,
metadata,
coalesce(default_currency, 'usd') AS default_currency,
if(stripe_connected = true, 1, 0) AS stripe_connected,
created_at::Int64 AS created_at,
if(onboarded = true, 1, 0) AS onboarded,
if(deployed = true, 1, 0) AS deployed,
'read' AS __action,
now() AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', id) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'organizations',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
TYPE COPY
TARGET_DATASOURCE organizations

35
server/tinybird/copies/prices_backfill.pipe Normal file
View File

@@ -0,0 +1,35 @@
DESCRIPTION >
Chunked backfill of prices from Postgres into prices.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
assumeNotNull(id) AS id,
coalesce(org_id, '') AS org_id,
coalesce(internal_product_id, '') AS internal_product_id,
ifNull(toString(config), 'null') AS config,
created_at::Int64 AS created_at,
billing_type,
tier_behavior,
toUInt8(if(coalesce(is_custom, false) = true, 1, 0)) AS is_custom,
entitlement_id,
ifNull(toString(proration_config), 'null') AS proration_config,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', assumeNotNull(id)) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'prices',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
TYPE COPY
TARGET_DATASOURCE prices

39
server/tinybird/copies/products_backfill.pipe Normal file
View File

@@ -0,0 +1,39 @@
DESCRIPTION >
Chunked backfill of products from Postgres into products.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
coalesce(internal_id, '') AS internal_id,
coalesce(id, '') AS id,
name,
description,
coalesce(org_id, '') AS org_id,
created_at::Int64 AS created_at,
coalesce(env, '') AS env,
toUInt8(if(is_add_on = true, 1, 0)) AS is_add_on,
toUInt8(if(is_default = true, 1, 0)) AS is_default,
group,
coalesce(version, 1)::Float64 AS version,
ifNull(toString(processor), 'null') AS processor,
base_variant_id,
toUInt8(if(archived = true, 1, 0)) AS archived,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', coalesce(internal_id, '')) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'products',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
TYPE COPY
TARGET_DATASOURCE products

30
server/tinybird/copies/replaceables_backfill.pipe Normal file
View File

@@ -0,0 +1,30 @@
DESCRIPTION >
Chunked backfill of replaceables from Postgres into replaceables.
Chunks by created_at (epoch ms / bigint). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
assumeNotNull(id) AS id,
coalesce(cus_ent_id, '') AS cus_ent_id,
created_at::Int64 AS created_at,
from_entity_id,
if(coalesce(delete_next_cycle, false) = true, 1, 0) AS delete_next_cycle,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', assumeNotNull(id)) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'replaceables',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
TYPE COPY
TARGET_DATASOURCE replaceables

35
server/tinybird/copies/rollovers_backfill.pipe Normal file
View File

@@ -0,0 +1,35 @@
DESCRIPTION >
Offset-based backfill of rollovers from Postgres into rollovers.
Pages through the full table using LIMIT + OFFSET ordered by id.
Stop when a page returns fewer rows than page_size.
NODE migrate
SQL >
%
SELECT
assumeNotNull(id) AS id,
coalesce(cus_ent_id, '') AS cus_ent_id,
coalesce(balance, 0)::Float64 AS balance,
if(expires_at IS NULL, NULL, expires_at::Int64) AS expires_at,
coalesce(usage, 0)::Float64 AS usage,
ifNull(toString(entities), '{}') AS entities,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', assumeNotNull(id)) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'rollovers',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE balance < 1e15
AND usage < 1e15
ORDER BY id ASC
LIMIT {{Int32(page_size, 5000)}}
OFFSET {{Int32(offset, 0)}}
TYPE COPY
TARGET_DATASOURCE rollovers

35
server/tinybird/copies/subscriptions_backfill.pipe Normal file
View File

@@ -0,0 +1,35 @@
DESCRIPTION >
Chunked backfill of subscriptions from Postgres into subscriptions.
Chunks by created_at (epoch ms). Cast numeric columns to match datasource schema.
NODE migrate
SQL >
%
SELECT
assumeNotNull(id) AS id,
coalesce(org_id, '') AS org_id,
stripe_id,
stripe_schedule_id,
created_at::Int64 AS created_at,
ifNull(toString(metadata), '{}') AS metadata,
arrayMap(x -> assumeNotNull(x), coalesce(usage_features, [])) AS usage_features,
env,
if(current_period_start IS NULL, NULL, current_period_start::Int64) AS current_period_start,
if(current_period_end IS NULL, NULL, current_period_end::Int64) AS current_period_end,
'read' AS __action,
toDateTime64(now(), 6) AS __commit_timestamp,
0 AS __commit_lsn,
concat('backfill-', assumeNotNull(id)) AS __idempotency_key
FROM postgresql(
'us-west-3.pg.psdb.cloud:5432',
'postgres',
'subscriptions',
{{tb_secret('PG_USERNAME')}},
{{tb_secret('PG_PASSWORD')}},
'public'
)
WHERE created_at > {{Int64(start_epoch_ms, 0)}}
AND created_at <= {{Int64(end_epoch_ms, 9999999999999)}}
TYPE COPY
TARGET_DATASOURCE subscriptions

34
server/tinybird/datasources/customer_entitlements.datasource Normal file
View File

@@ -0,0 +1,34 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
customer_entitlements CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`id` String `json:$.id`,
`customer_product_id` Nullable(String) `json:$.customer_product_id`,
`entitlement_id` String `json:$.entitlement_id`,
`internal_customer_id` String `json:$.internal_customer_id`,
`internal_entity_id` Nullable(String) `json:$.internal_entity_id`,
`internal_feature_id` String `json:$.internal_feature_id`,
`unlimited` UInt8 `json:$.unlimited`,
`balance` Float64 `json:$.balance`,
`created_at` Int64 `json:$.created_at`,
`next_reset_at` Nullable(Int64) `json:$.next_reset_at`,
`usage_allowed` UInt8 `json:$.usage_allowed`,
`adjustment` Nullable(Float64) `json:$.adjustment`,
`additional_balance` Float64 `json:$.additional_balance`,
`entities` String `json:$.entities`,
`expires_at` Nullable(Int64) `json:$.expires_at`,
`cache_version` Int32 `json:$.cache_version`,
`customer_id` Nullable(String) `json:$.customer_id`,
`feature_id` Nullable(String) `json:$.feature_id`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

22
server/tinybird/datasources/customer_prices.datasource Normal file
View File

@@ -0,0 +1,22 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
customer_prices CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`id` String `json:$.id`,
`created_at` Int64 `json:$.created_at`,
`price_id` Nullable(String) `json:$.price_id`,
`options` String `json:$.options`,
`internal_customer_id` Nullable(String) `json:$.internal_customer_id`,
`customer_product_id` Nullable(String) `json:$.customer_product_id`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

41
server/tinybird/datasources/customer_products.datasource Normal file
View File

@@ -0,0 +1,41 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
customer_products CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`id` String `json:$.id`,
`internal_customer_id` String `json:$.internal_customer_id`,
`internal_product_id` String `json:$.internal_product_id`,
`internal_entity_id` Nullable(String) `json:$.internal_entity_id`,
`created_at` Nullable(Int64) `json:$.created_at`,
`status` Nullable(String) `json:$.status`,
`processor` String `json:$.processor`,
`canceled` UInt8 `json:$.canceled`,
`canceled_at` Nullable(Int64) `json:$.canceled_at`,
`ended_at` Nullable(Int64) `json:$.ended_at`,
`starts_at` Nullable(Int64) `json:$.starts_at`,
`options` String `json:$.options`,
`product_id` Nullable(String) `json:$.product_id`,
`free_trial_id` Nullable(String) `json:$.free_trial_id`,
`trial_ends_at` Nullable(Int64) `json:$.trial_ends_at`,
`collection_method` String `json:$.collection_method`,
`subscription_ids` Array(String) `json:$.subscription_ids[:]`,
`scheduled_ids` Array(String) `json:$.scheduled_ids[:]`,
`quantity` Float64 `json:$.quantity`,
`is_custom` UInt8 `json:$.is_custom`,
`customer_id` Nullable(String) `json:$.customer_id`,
`entity_id` Nullable(String) `json:$.entity_id`,
`billing_version` Nullable(String) `json:$.billing_version`,
`api_version` Nullable(Int64) `json:$.api_version`,
`api_semver` Nullable(String) `json:$.api_semver`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

26
server/tinybird/datasources/customers.datasource Normal file
View File

@@ -0,0 +1,26 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
Customer CDC events from Sequin (PlanetScale Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`internal_id` String `json:$.internal_id`,
`org_id` String `json:$.org_id`,
`created_at` Nullable(Int64) `json:$.created_at`,
`name` Nullable(String) `json:$.name`,
`id` Nullable(String) `json:$.id`,
`email` Nullable(String) `json:$.email`,
`fingerprint` Nullable(String) `json:$.fingerprint`,
`metadata` String `json:$.metadata`,
`env` String `json:$.env`,
`processor` String `json:$.processor`,
`processors` String `json:$.processors`,
`send_email_receipts` UInt8 `json:$.send_email_receipts`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "internal_id"
ENGINE_PRIMARY_KEY "internal_id"

27
server/tinybird/datasources/entities.datasource Normal file
View File

@@ -0,0 +1,27 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
entities CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
Primary key is internal_id (the stable surrogate PK), not id (the user-facing id).
SCHEMA >
`id` Nullable(String) `json:$.id`,
`org_id` Nullable(String) `json:$.org_id`,
`created_at` Int64 `json:$.created_at`,
`internal_id` String `json:$.internal_id`,
`internal_customer_id` String `json:$.internal_customer_id`,
`env` Nullable(String) `json:$.env`,
`name` Nullable(String) `json:$.name`,
`deleted` UInt8 `json:$.deleted`,
`internal_feature_id` Nullable(String) `json:$.internal_feature_id`,
`feature_id` Nullable(String) `json:$.feature_id`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "internal_id"
ENGINE_PRIMARY_KEY "internal_id"

31
server/tinybird/datasources/entitlements.datasource Normal file
View File

@@ -0,0 +1,31 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
entitlements CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`id` String `json:$.id`,
`created_at` Int64 `json:$.created_at`,
`internal_feature_id` String `json:$.internal_feature_id`,
`internal_product_id` Nullable(String) `json:$.internal_product_id`,
`is_custom` UInt8 `json:$.is_custom`,
`allowance_type` Nullable(String) `json:$.allowance_type`,
`allowance` Nullable(Float64) `json:$.allowance`,
`interval` Nullable(String) `json:$.interval`,
`interval_count` Float64 `json:$.interval_count`,
`carry_from_previous` UInt8 `json:$.carry_from_previous`,
`entity_feature_id` Nullable(String) `json:$.entity_feature_id`,
`org_id` Nullable(String) `json:$.org_id`,
`feature_id` Nullable(String) `json:$.feature_id`,
`usage_limit` Nullable(Float64) `json:$.usage_limit`,
`rollover` String `json:$.rollover`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

27
server/tinybird/datasources/features.datasource Normal file
View File

@@ -0,0 +1,27 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
features CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`internal_id` String `json:$.internal_id`,
`org_id` String `json:$.org_id`,
`created_at` Nullable(Int64) `json:$.created_at`,
`env` Nullable(String) `json:$.env`,
`id` String `json:$.id`,
`name` Nullable(String) `json:$.name`,
`type` String `json:$.type`,
`config` String `json:$.config`,
`display` String `json:$.display`,
`archived` UInt8 `json:$.archived`,
`event_names` Array(String) `json:$.event_names[:]`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "internal_id"
ENGINE_PRIMARY_KEY "internal_id"

24
server/tinybird/datasources/free_trials.datasource Normal file
View File

@@ -0,0 +1,24 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
free_trials CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`id` String `json:$.id`,
`created_at` Int64 `json:$.created_at`,
`internal_product_id` Nullable(String) `json:$.internal_product_id`,
`duration` String `json:$.duration`,
`length` Nullable(Float64) `json:$.length`,
`unique_fingerprint` UInt8 `json:$.unique_fingerprint`,
`is_custom` UInt8 `json:$.is_custom`,
`card_required` UInt8 `json:$.card_required`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

27
server/tinybird/datasources/invoices.datasource Normal file
View File

@@ -0,0 +1,27 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
Invoice CDC events from Sequin (PlanetScale Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`id` String `json:$.id`,
`created_at` Nullable(Int64) `json:$.created_at`,
`product_ids` Array(String) `json:$.product_ids[:]`,
`internal_product_ids` Array(String) `json:$.internal_product_ids[:]`,
`internal_customer_id` String `json:$.internal_customer_id`,
`internal_entity_id` Nullable(String) `json:$.internal_entity_id`,
`stripe_id` String `json:$.stripe_id`,
`status` String `json:$.status`,
`hosted_invoice_url` Nullable(String) `json:$.hosted_invoice_url`,
`total` Nullable(Float64) `json:$.total`,
`currency` String `json:$.currency`,
`discounts` String `json:$.discounts`,
`items` String `json:$.items`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

25
server/tinybird/datasources/organizations.datasource Normal file
View File

@@ -0,0 +1,25 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
Organization CDC events from Sequin (PlanetScale Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`id` String `json:$.id`,
`slug` String `json:$.slug`,
`name` String `json:$.name`,
`logo` Nullable(String) `json:$.logo`,
`created_at_ts` Nullable(String) `json:$.created_at_ts`,
`metadata` Nullable(String) `json:$.metadata`,
`default_currency` String `json:$.default_currency`,
`stripe_connected` UInt8 `json:$.stripe_connected`,
`created_at` Nullable(Int64) `json:$.created_at`,
`onboarded` UInt8 `json:$.onboarded`,
`deployed` UInt8 `json:$.deployed`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

26
server/tinybird/datasources/prices.datasource Normal file
View File

@@ -0,0 +1,26 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
prices CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`id` String `json:$.id`,
`org_id` String `json:$.org_id`,
`internal_product_id` String `json:$.internal_product_id`,
`config` String `json:$.config`,
`created_at` Int64 `json:$.created_at`,
`billing_type` Nullable(String) `json:$.billing_type`,
`tier_behavior` Nullable(String) `json:$.tier_behavior`,
`is_custom` UInt8 `json:$.is_custom`,
`entitlement_id` Nullable(String) `json:$.entitlement_id`,
`proration_config` String `json:$.proration_config`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

30
server/tinybird/datasources/products.datasource Normal file
View File

@@ -0,0 +1,30 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
products CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`internal_id` String `json:$.internal_id`,
`id` String `json:$.id`,
`name` Nullable(String) `json:$.name`,
`description` Nullable(String) `json:$.description`,
`org_id` String `json:$.org_id`,
`created_at` Int64 `json:$.created_at`,
`env` String `json:$.env`,
`is_add_on` UInt8 `json:$.is_add_on`,
`is_default` UInt8 `json:$.is_default`,
`group` Nullable(String) `json:$.group`,
`version` Float64 `json:$.version`,
`processor` String `json:$.processor`,
`base_variant_id` Nullable(String) `json:$.base_variant_id`,
`archived` UInt8 `json:$.archived`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "internal_id"
ENGINE_PRIMARY_KEY "internal_id"

21
server/tinybird/datasources/replaceables.datasource Normal file
View File

@@ -0,0 +1,21 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
replaceables CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`id` String `json:$.id`,
`cus_ent_id` String `json:$.cus_ent_id`,
`created_at` Int64 `json:$.created_at`,
`from_entity_id` Nullable(String) `json:$.from_entity_id`,
`delete_next_cycle` UInt8 `json:$.delete_next_cycle`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

23
server/tinybird/datasources/rollovers.datasource Normal file
View File

@@ -0,0 +1,23 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
rollovers CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
No created_at — deduplication is by id + __commit_lsn.
SCHEMA >
`id` String `json:$.id`,
`cus_ent_id` String `json:$.cus_ent_id`,
`balance` Float64 `json:$.balance`,
`expires_at` Nullable(Int64) `json:$.expires_at`,
`usage` Float64 `json:$.usage`,
`entities` String `json:$.entities`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

26
server/tinybird/datasources/subscriptions.datasource Normal file
View File

@@ -0,0 +1,26 @@
TOKEN sequin_cdc APPEND
DESCRIPTION >
subscriptions CDC events from Sequin (Postgres → Tinybird via WAL).
action = insert | update | delete | read (read = initial backfill row).
SCHEMA >
`id` String `json:$.id`,
`org_id` String `json:$.org_id`,
`stripe_id` Nullable(String) `json:$.stripe_id`,
`stripe_schedule_id` Nullable(String) `json:$.stripe_schedule_id`,
`created_at` Int64 `json:$.created_at`,
`metadata` String `json:$.metadata`,
`usage_features` Array(String) `json:$.usage_features[:]`,
`env` Nullable(String) `json:$.env`,
`current_period_start` Nullable(Int64) `json:$.current_period_start`,
`current_period_end` Nullable(Int64) `json:$.current_period_end`,
`__action` String `json:$.__action`,
`__commit_timestamp` DateTime64(6) `json:$.__commit_timestamp`,
`__commit_lsn` Int64 `json:$.__commit_lsn`,
`__idempotency_key` String `json:$.__idempotency_key`
ENGINE "ReplacingMergeTree"
ENGINE_VER "__commit_lsn"
ENGINE_SORTING_KEY "id"
ENGINE_PRIMARY_KEY "id"

284
server/tinybird/scripts/backfill_base.ts Normal file
View File

@@ -0,0 +1,284 @@
#!/usr/bin/env bun
/**
* Shared utilities and config for all epoch-ms-chunked Tinybird backfill scripts.
*/
import { execSync } from "child_process";
// ============================================================================
// CONSTANTS
// ============================================================================
export const MAX_RETRIES = 3;
export const RETRY_DELAY_MS = 5000;
export const DELAY_BETWEEN_CHUNKS_MS = 2000;
export const DEFAULT_CHUNK_DAYS = 30;
// ============================================================================
// TYPES
// ============================================================================
export interface Chunk {
startEpochMs: number;
endEpochMs: number;
}
export interface TableConfig {
/** Human-readable label, e.g. "Customers" */
label: string;
/** Tinybird copy pipe name, e.g. "customers_backfill" */
copyPipeName: string;
/** Tinybird datasource table name, e.g. "customers" */
datasource: string;
/** Epoch ms of the earliest known row in Postgres */
startEpochMs: number;
}
// ============================================================================
// UTILITIES
// ============================================================================
export function execCapture({ cmd }: { cmd: string }): string {
try {
return execSync(cmd, { encoding: "utf-8", stdio: ["pipe", "pipe", "pipe"] }).trim();
} catch (error: any) {
return error.stdout?.trim() ?? "";
}
}
const TB_API_URL = process.env.TINYBIRD_API_URL ?? "https://api.us-west-2.aws.tinybird.co";
const TB_TOKEN = process.env.TINYBIRD_TOKEN;
if (!TB_TOKEN) {
console.error("TINYBIRD_TOKEN env var is not set. Export it before running backfill scripts.");
process.exit(1);
}
/** Query Tinybird SQL API directly — returns parsed rows as an array of objects. */
export async function tbSql({ query }: { query: string }): Promise<Record<string, unknown>[]> {
const queryWithFormat = `${query.trimEnd()} FORMAT JSONEachRow`;
const url = `${TB_API_URL}/v0/sql?q=${encodeURIComponent(queryWithFormat)}`;
const res = await fetch(url, {
headers: { Authorization: `Bearer ${TB_TOKEN}` },
});
if (!res.ok) {
const body = await res.text();
throw new Error(`Tinybird SQL API error ${res.status}: ${body}`);
}
const text = await res.text();
// JSONEachRow returns one JSON object per line
return text
.split("\n")
.map((l) => l.trim())
.filter((l) => l.length > 0)
.map((l) => JSON.parse(l) as Record<string, unknown>);
}
export function exec({ cmd }: { cmd: string }): void {
execSync(cmd, { encoding: "utf-8", stdio: "inherit" });
}
export async function sleep({ ms }: { ms: number }): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
/** Extract a single numeric value from the first row/column of a tbSql result. */
export function extractNumber({ rows, col }: { rows: Record<string, unknown>[]; col: string }): number | null {
const val = rows[0]?.[col];
if (val === null || val === undefined) return null;
const num = Number(val);
return isNaN(num) ? null : num;
}
export function formatEpochMs({ epochMs }: { epochMs: number }): string {
return new Date(epochMs).toISOString();
}
// ============================================================================
// TINYBIRD HELPERS
// ============================================================================
/** Returns the MAX created_at among backfill ('read') rows, or null if none. */
export async function getMaxCreatedAt({ datasource }: { datasource: string }): Promise<number | null> {
const rows = await tbSql({
query: `SELECT max(created_at) AS val FROM ${datasource} WHERE __action = 'read'`,
});
const num = extractNumber({ rows, col: "val" });
return num && num > 0 ? num : null;
}
/** Returns count of non-deleted rows in the datasource. */
export async function getRowCount({ datasource }: { datasource: string }): Promise<number> {
const rows = await tbSql({
query: `SELECT count() AS val FROM ${datasource} FINAL WHERE __action != 'delete'`,
});
return extractNumber({ rows, col: "val" }) ?? 0;
}
// ============================================================================
// CHUNK GENERATION (FORWARD — oldest → newest)
// ============================================================================
export function generateChunksForward({
startEpochMs,
endEpochMs,
chunkDays,
}: {
startEpochMs: number;
endEpochMs: number;
chunkDays: number;
}): Chunk[] {
const chunks: Chunk[] = [];
const chunkMs = chunkDays * 24 * 60 * 60 * 1000;
let current = startEpochMs;
while (current < endEpochMs) {
const next = Math.min(current + chunkMs, endEpochMs);
chunks.push({ startEpochMs: current, endEpochMs: next });
current = next;
}
return chunks;
}
// ============================================================================
// COPY JOB EXECUTION
// ============================================================================
export async function waitForCopyJobs(): Promise<void> {
const maxAttempts = 60;
for (let attempt = 0; attempt < maxAttempts; attempt++) {
const waiting = execCapture({
cmd: "tb --cloud job ls --status waiting --kind copy 2>/dev/null | grep -c '^id:' || echo 0",
});
const working = execCapture({
cmd: "tb --cloud job ls --status working --kind copy 2>/dev/null | grep -c '^id:' || echo 0",
});
const total = parseInt(waiting, 10) + parseInt(working, 10);
if (total === 0) return;
if (attempt === 0) console.log(` Waiting for ${total} existing copy job(s) to complete...`);
await sleep({ ms: 5000 });
}
throw new Error("Timed out waiting for existing copy jobs to complete");
}
export async function runCopyJob({
copyPipeName,
startEpochMs,
endEpochMs,
retries = MAX_RETRIES,
}: {
copyPipeName: string;
startEpochMs: number;
endEpochMs: number;
retries?: number;
}): Promise<boolean> {
for (let attempt = 1; attempt <= retries; attempt++) {
try {
await waitForCopyJobs();
exec({
cmd: `tb --cloud copy run ${copyPipeName} --param start_epoch_ms="${startEpochMs}" --param end_epoch_ms="${endEpochMs}" --wait`,
});
return true;
} catch (error: any) {
const msg = error.message || error.toString();
if (attempt < retries) {
console.log(` Attempt ${attempt}/${retries} failed. Retrying in ${RETRY_DELAY_MS / 1000}s...`);
console.log(` Error: ${msg.substring(0, 200)}`);
await sleep({ ms: RETRY_DELAY_MS });
} else {
console.error(` All ${retries} attempts failed for chunk ${startEpochMs} -> ${endEpochMs}`);
console.error(` Error: ${msg}`);
return false;
}
}
}
return false;
}
// ============================================================================
// CORE BACKFILL RUNNER
// ============================================================================
/**
* Runs a complete forward (oldest → newest) backfill for a single table config.
* Idempotent: resumes from MAX(created_at) of existing 'read' rows.
*/
export async function runTableBackfill({
config,
chunkDays,
dryRun,
}: {
config: TableConfig;
chunkDays: number;
dryRun: boolean;
}): Promise<void> {
console.log(`\n=== ${config.label} Backfill ===`);
const tinybirdMax = await getMaxCreatedAt({ datasource: config.datasource });
if (tinybirdMax) {
console.log(` Resume point: ${tinybirdMax} (${formatEpochMs({ epochMs: tinybirdMax })})`);
} else {
console.log(" No backfilled rows yet — starting from beginning");
}
const resumePoint = tinybirdMax ? tinybirdMax + 1 : config.startEpochMs;
const endPoint = Date.now();
console.log(` Start: ${formatEpochMs({ epochMs: resumePoint })}`);
console.log(` End: ${formatEpochMs({ epochMs: endPoint })}`);
if (resumePoint >= endPoint) {
console.log(` Already complete. Rows in Tinybird: ${await getRowCount({ datasource: config.datasource })}`);
return;
}
const chunks = generateChunksForward({ startEpochMs: resumePoint, endEpochMs: endPoint, chunkDays });
console.log(` Chunks: ${chunks.length} (${chunkDays} days each)\n`);
if (dryRun) {
chunks.forEach((chunk, i) => {
console.log(
` ${i + 1}. ${formatEpochMs({ epochMs: chunk.startEpochMs })} -> ${formatEpochMs({ epochMs: chunk.endEpochMs })}`,
);
});
return;
}
for (let i = 0; i < chunks.length; i++) {
const chunk = chunks[i];
const chunkNum = i + 1;
console.log(
`Chunk ${chunkNum}/${chunks.length}: ${formatEpochMs({ epochMs: chunk.startEpochMs })} -> ${formatEpochMs({ epochMs: chunk.endEpochMs })}`,
);
const startTime = Date.now();
const success = await runCopyJob({
copyPipeName: config.copyPipeName,
startEpochMs: chunk.startEpochMs,
endEpochMs: chunk.endEpochMs,
});
const duration = Math.round((Date.now() - startTime) / 1000);
if (success) {
const count = await getRowCount({ datasource: config.datasource });
console.log(` ✓ Chunk ${chunkNum} done in ${duration}s. Rows: ${count}\n`);
} else {
console.log(` ✗ Chunk ${chunkNum} FAILED after ${duration}s`);
console.log(" Re-run this script to resume.\n");
process.exit(1);
}
if (i < chunks.length - 1) {
await sleep({ ms: DELAY_BETWEEN_CHUNKS_MS });
}
}
console.log(`=== ${config.label} Backfill Complete. Rows: ${await getRowCount({ datasource: config.datasource })} ===\n`);
}

303
server/tinybird/scripts/backfill_cli.ts Normal file
View File

@@ -0,0 +1,303 @@
#!/usr/bin/env bun
/**
* Interactive CLI for backfilling Postgres tables into Tinybird.
*
* Multi-select which tables to backfill, then runs them sequentially.
* All jobs are idempotent — safe to re-run, automatically resumes from progress.
*
* Usage:
* bun scripts/backfill_cli.ts
* bun scripts/backfill_cli.ts --chunk-days 14
* bun scripts/backfill_cli.ts --dry-run
* bun scripts/backfill_cli.ts --all
*/
import * as readline from "readline";
import {
DEFAULT_CHUNK_DAYS,
type TableConfig,
runTableBackfill,
} from "./backfill_base.js";
// ============================================================================
// TABLE REGISTRY
// NOTE: rollovers is NOT listed here — it uses cursor-based pagination.
// Run: bun scripts/backfill_rollovers.ts
// ============================================================================
const TABLE_CONFIGS: TableConfig[] = [
{
label: "customers",
copyPipeName: "customers_backfill",
datasource: "customers",
startEpochMs: 1706987055000,
},
{
label: "invoices",
copyPipeName: "invoices_backfill",
datasource: "invoices",
startEpochMs: 1706987056000,
},
{
label: "organizations",
copyPipeName: "organizations_backfill",
datasource: "organizations",
startEpochMs: 1737543151220,
},
{
label: "customer_products",
copyPipeName: "customer_products_backfill",
datasource: "customer_products",
startEpochMs: 1677713742000,
},
{
label: "customer_entitlements",
copyPipeName: "customer_entitlements_backfill",
datasource: "customer_entitlements",
startEpochMs: 1738268254191,
},
{
label: "customer_prices",
copyPipeName: "customer_prices_backfill",
datasource: "customer_prices",
startEpochMs: 1738341655109,
},
{
label: "replaceables",
copyPipeName: "replaceables_backfill",
datasource: "replaceables",
startEpochMs: 1751360953240,
},
{
label: "entitlements",
copyPipeName: "entitlements_backfill",
datasource: "entitlements",
startEpochMs: 1737570713388,
},
{
label: "free_trials",
copyPipeName: "free_trials_backfill",
datasource: "free_trials",
startEpochMs: 1738168893927,
},
{
label: "entities",
copyPipeName: "entities_backfill",
datasource: "entities",
startEpochMs: 1743685142432,
},
{
label: "subscriptions",
copyPipeName: "subscriptions_backfill",
datasource: "subscriptions",
startEpochMs: 1744118448491,
},
{
label: "features",
copyPipeName: "features_backfill",
datasource: "features",
startEpochMs: 1737570301197,
},
{
label: "prices",
copyPipeName: "prices_backfill",
datasource: "prices",
startEpochMs: 1737570713388,
},
{
label: "products",
copyPipeName: "products_backfill",
datasource: "products",
startEpochMs: 1737570326143,
},
// rollovers intentionally excluded — use backfill_rollovers.ts instead
];
// ============================================================================
// ARGUMENT PARSING
// ============================================================================
interface Args {
chunkDays: number;
dryRun: boolean;
all: boolean;
}
function parseArgs(): Args {
const args: Args = {
chunkDays: DEFAULT_CHUNK_DAYS,
dryRun: false,
all: false,
};
for (let i = 2; i < process.argv.length; i++) {
const arg = process.argv[i];
if (arg === "--chunk-days" && process.argv[i + 1]) {
args.chunkDays = parseInt(process.argv[++i], 10);
} else if (arg === "--dry-run") {
args.dryRun = true;
} else if (arg === "--all") {
args.all = true;
} else if (arg === "--help" || arg === "-h") {
console.log(`
Interactive Tinybird backfill CLI
Usage: bun scripts/backfill_cli.ts [options]
Options:
--chunk-days <n> Days per chunk (default: ${DEFAULT_CHUNK_DAYS})
--dry-run Show chunks without executing copy jobs
--all Skip prompt, backfill all tables
--help, -h Show this help message
Available tables:
${TABLE_CONFIGS.map((t, i) => ` ${i + 1}. ${t.label}`).join("\n")}
Note: rollovers uses cursor-based pagination — run separately:
bun scripts/backfill_rollovers.ts
`);
process.exit(0);
}
}
return args;
}
// ============================================================================
// MULTI-SELECT PROMPT
// ============================================================================
async function multiSelect({
items,
prompt,
}: {
items: string[];
prompt: string;
}): Promise<number[]> {
const selected = new Set<number>();
// Pre-select all
items.forEach((_, i) => selected.add(i));
const rl = readline.createInterface({
input: process.stdin,
output: process.stdout,
});
// Keys: 1-9 for indices 0-8, then a-z for indices 9-34
// (a/A are reserved for select-all, n/N for select-none — skip those)
const KEYS = "123456789bcdefghijklmopqrstuvwxyz";
const keyForIndex = (i: number) => KEYS[i] ?? "?";
const renderMenu = () => {
console.clear();
console.log(prompt);
console.log("(key to toggle, A to select all, N to select none, Enter to confirm)\n");
items.forEach((item, i) => {
const checked = selected.has(i) ? "[x]" : "[ ]";
console.log(` ${checked} ${keyForIndex(i)}. ${item}`);
});
console.log("");
};
return new Promise((resolve) => {
renderMenu();
// Use raw mode for keypress detection
if (process.stdin.isTTY) {
process.stdin.setRawMode(true);
}
const onKeypress = (chunk: Buffer) => {
const key = chunk.toString().toLowerCase();
if (key === "\r" || key === "\n") {
// Enter — confirm
process.stdin.removeListener("data", onKeypress);
if (process.stdin.isTTY) {
process.stdin.setRawMode(false);
}
rl.close();
console.clear();
resolve([...selected].sort((a, b) => a - b));
return;
}
if (key === "a") {
items.forEach((_, i) => selected.add(i));
renderMenu();
return;
}
if (key === "n") {
selected.clear();
renderMenu();
return;
}
if (key === "\x03") {
// Ctrl+C
process.exit(0);
}
// Map key to index
const idx = KEYS.indexOf(key);
if (idx !== -1 && idx < items.length) {
if (selected.has(idx)) {
selected.delete(idx);
} else {
selected.add(idx);
}
renderMenu();
return;
}
};
process.stdin.on("data", onKeypress);
});
}
// ============================================================================
// MAIN
// ============================================================================
async function main() {
const args = parseArgs();
let selectedConfigs: TableConfig[];
if (args.all) {
selectedConfigs = TABLE_CONFIGS;
console.log("--all flag set: backfilling all tables.\n");
} else {
const selectedIndices = await multiSelect({
items: TABLE_CONFIGS.map((t) => t.label),
prompt: "Select tables to backfill:",
});
if (selectedIndices.length === 0) {
console.log("No tables selected. Exiting.");
process.exit(0);
}
selectedConfigs = selectedIndices.map((i) => TABLE_CONFIGS[i]);
}
console.log(`\nBackfilling ${selectedConfigs.length} table(s): ${selectedConfigs.map((t) => t.label).join(", ")}`);
console.log(`Chunk size: ${args.chunkDays} days`);
if (args.dryRun) console.log("DRY RUN — no copy jobs will be executed.\n");
for (const config of selectedConfigs) {
await runTableBackfill({ config, chunkDays: args.chunkDays, dryRun: args.dryRun });
}
console.log("\n==========================================");
console.log("All selected backfills complete.");
}
main().catch((error) => {
console.error("Fatal error:", error);
process.exit(1);
});

2
server/tinybird/scripts/backfill_events.ts
View File

@@ -124,7 +124,7 @@ function exec(cmd: string, silent = false): string {
function execCapture(cmd: string): string {
try {
return execSync(cmd, { encoding: "utf-8", stderr: "pipe" }).trim();
return execSync(cmd, { encoding: "utf-8", stdio: ["pipe", "pipe", "pipe"] }).trim();
} catch (error: any) {
return error.stdout?.trim() ?? "";
}

170
server/tinybird/scripts/backfill_rollovers.ts Normal file
View File

@@ -0,0 +1,170 @@
#!/usr/bin/env bun
/**
* Offset-based idempotent backfill for rollovers from Postgres to Tinybird.
*
* Rollovers have no created_at so we can't chunk by date.
* Pages through the full table with LIMIT + OFFSET ordered by id.
* Stops when a page returns fewer rows than page_size.
*
* Usage:
* bun scripts/backfill_rollovers.ts
* bun scripts/backfill_rollovers.ts --page-size 1000
* bun scripts/backfill_rollovers.ts --offset 10000
* bun scripts/backfill_rollovers.ts --dry-run
*/
import {
MAX_RETRIES,
RETRY_DELAY_MS,
DELAY_BETWEEN_CHUNKS_MS,
exec,
tbSql,
sleep,
extractNumber,
waitForCopyJobs,
} from "./backfill_base.js";
// ============================================================================
// CONFIGURATION
// ============================================================================
const DATASOURCE = "rollovers";
const COPY_PIPE = "rollovers_backfill";
const DEFAULT_PAGE_SIZE = 5000;
// ============================================================================
// ARGUMENT PARSING
// ============================================================================
function parseArgs(): { pageSize: number; offsetOverride?: number; dryRun: boolean } {
const args = { pageSize: DEFAULT_PAGE_SIZE, offsetOverride: undefined as number | undefined, dryRun: false };
for (let i = 2; i < process.argv.length; i++) {
const arg = process.argv[i];
if (arg === "--page-size" && process.argv[i + 1]) {
args.pageSize = parseInt(process.argv[++i], 10);
} else if (arg === "--offset" && process.argv[i + 1]) {
args.offsetOverride = parseInt(process.argv[++i], 10);
} else if (arg === "--dry-run") {
args.dryRun = true;
} else if (arg === "--help" || arg === "-h") {
console.log(`
Offset-based backfill for rollovers
Usage: bun scripts/backfill_rollovers.ts [options]
Options:
--page-size <n> Rows per page (default: ${DEFAULT_PAGE_SIZE})
--offset <n> Start from this offset (default: 0)
--dry-run Print plan without executing
--help, -h Show this help
`);
process.exit(0);
}
}
return args;
}
// ============================================================================
// HELPERS
// ============================================================================
async function getRowCount(): Promise<number> {
const rows = await tbSql({
query: `SELECT count() AS val FROM ${DATASOURCE} FINAL WHERE __action != 'delete'`,
});
return extractNumber({ rows, col: "val" }) ?? 0;
}
async function runPage({
offset,
pageSize,
retries = MAX_RETRIES,
}: {
offset: number;
pageSize: number;
retries?: number;
}): Promise<boolean> {
for (let attempt = 1; attempt <= retries; attempt++) {
try {
await waitForCopyJobs();
exec({
cmd: `TB_VERSION_WARNING=0 tb --cloud copy run ${COPY_PIPE} --param offset="${offset}" --param page_size="${pageSize}" --wait`,
});
return true;
} catch (error: any) {
const msg = error.message || error.toString();
if (attempt < retries) {
console.log(` Attempt ${attempt}/${retries} failed. Retrying in ${RETRY_DELAY_MS / 1000}s...`);
console.log(` Error: ${msg.substring(0, 200)}`);
await sleep({ ms: RETRY_DELAY_MS });
} else {
console.error(` All ${retries} attempts failed at offset ${offset}`);
console.error(` Error: ${msg}`);
return false;
}
}
}
return false;
}
// ============================================================================
// MAIN
// ============================================================================
async function main() {
console.log("=== Rollovers Backfill (Offset-based) ===\n");
const args = parseArgs();
let offset = args.offsetOverride ?? 0;
console.log(`Page size: ${args.pageSize}`);
console.log(`Starting offset: ${offset}\n`);
if (args.dryRun) {
console.log("DRY RUN — would run pages starting at offset", offset);
process.exit(0);
}
let pageNum = 0;
while (true) {
pageNum++;
console.log(`=== Page ${pageNum}: offset=${offset} ===`);
const startTime = Date.now();
const success = await runPage({ offset, pageSize: args.pageSize });
const duration = Math.round((Date.now() - startTime) / 1000);
if (!success) {
console.log(`✗ Page ${pageNum} FAILED after ${duration}s`);
console.log("To resume from this point, re-run with:");
console.log(` bun scripts/backfill_rollovers.ts --offset ${offset}\n`);
process.exit(1);
}
const rowCount = await getRowCount();
console.log(`✓ Page ${pageNum} done in ${duration}s. Rows in Tinybird: ${rowCount}\n`);
// A page returning fewer rows than page_size means we've hit the end
// We detect this by comparing expected next offset vs actual row count
offset += args.pageSize;
if (rowCount < offset) {
console.log("Last page was partial — backfill complete.");
break;
}
await sleep({ ms: DELAY_BETWEEN_CHUNKS_MS });
}
console.log("==========================================");
console.log("=== Rollovers Backfill Complete ===");
console.log(`Total rollovers in Tinybird: ${await getRowCount()}`);
}
main().catch((error) => {
console.error("Fatal error:", error);
process.exit(1);
});