diff --git a/others/sequin/tinybird_mega_transform.ex b/others/sequin/tinybird_mega_transform.ex
new file mode 100644
index 000000000..4048ed9b9
--- /dev/null
+++ b/others/sequin/tinybird_mega_transform.ex
@@ -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
diff --git a/others/sequin/tinybird_routing.ex b/others/sequin/tinybird_routing.ex
new file mode 100644
index 000000000..0167ca238
--- /dev/null
+++ b/others/sequin/tinybird_routing.ex
@@ -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
diff --git a/server/tinybird/copies/customer_entitlements_backfill.pipe b/server/tinybird/copies/customer_entitlements_backfill.pipe
new file mode 100644
index 000000000..e1a7eea7e
--- /dev/null
+++ b/server/tinybird/copies/customer_entitlements_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/customer_prices_backfill.pipe b/server/tinybird/copies/customer_prices_backfill.pipe
new file mode 100644
index 000000000..087376bf1
--- /dev/null
+++ b/server/tinybird/copies/customer_prices_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/customer_products_backfill.pipe b/server/tinybird/copies/customer_products_backfill.pipe
new file mode 100644
index 000000000..df2afd1b9
--- /dev/null
+++ b/server/tinybird/copies/customer_products_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/customers_backfill.pipe b/server/tinybird/copies/customers_backfill.pipe
new file mode 100644
index 000000000..a8d351b71
--- /dev/null
+++ b/server/tinybird/copies/customers_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/entities_backfill.pipe b/server/tinybird/copies/entities_backfill.pipe
new file mode 100644
index 000000000..e6b6b11ae
--- /dev/null
+++ b/server/tinybird/copies/entities_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/entitlements_backfill.pipe b/server/tinybird/copies/entitlements_backfill.pipe
new file mode 100644
index 000000000..825bceca4
--- /dev/null
+++ b/server/tinybird/copies/entitlements_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/features_backfill.pipe b/server/tinybird/copies/features_backfill.pipe
new file mode 100644
index 000000000..7ffb2640e
--- /dev/null
+++ b/server/tinybird/copies/features_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/free_trials_backfill.pipe b/server/tinybird/copies/free_trials_backfill.pipe
new file mode 100644
index 000000000..28e76086a
--- /dev/null
+++ b/server/tinybird/copies/free_trials_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/invoices_backfill.pipe b/server/tinybird/copies/invoices_backfill.pipe
new file mode 100644
index 000000000..4153348b5
--- /dev/null
+++ b/server/tinybird/copies/invoices_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/organizations_backfill.pipe b/server/tinybird/copies/organizations_backfill.pipe
new file mode 100644
index 000000000..5140be0ad
--- /dev/null
+++ b/server/tinybird/copies/organizations_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/prices_backfill.pipe b/server/tinybird/copies/prices_backfill.pipe
new file mode 100644
index 000000000..11db0343c
--- /dev/null
+++ b/server/tinybird/copies/prices_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/products_backfill.pipe b/server/tinybird/copies/products_backfill.pipe
new file mode 100644
index 000000000..f95d2a2d3
--- /dev/null
+++ b/server/tinybird/copies/products_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/replaceables_backfill.pipe b/server/tinybird/copies/replaceables_backfill.pipe
new file mode 100644
index 000000000..310c0bcaa
--- /dev/null
+++ b/server/tinybird/copies/replaceables_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/rollovers_backfill.pipe b/server/tinybird/copies/rollovers_backfill.pipe
new file mode 100644
index 000000000..0bcaea069
--- /dev/null
+++ b/server/tinybird/copies/rollovers_backfill.pipe
@@ -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
diff --git a/server/tinybird/copies/subscriptions_backfill.pipe b/server/tinybird/copies/subscriptions_backfill.pipe
new file mode 100644
index 000000000..a45b77829
--- /dev/null
+++ b/server/tinybird/copies/subscriptions_backfill.pipe
@@ -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
diff --git a/server/tinybird/datasources/customer_entitlements.datasource b/server/tinybird/datasources/customer_entitlements.datasource
new file mode 100644
index 000000000..c50097627
--- /dev/null
+++ b/server/tinybird/datasources/customer_entitlements.datasource
@@ -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"
diff --git a/server/tinybird/datasources/customer_prices.datasource b/server/tinybird/datasources/customer_prices.datasource
new file mode 100644
index 000000000..d626f2885
--- /dev/null
+++ b/server/tinybird/datasources/customer_prices.datasource
@@ -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"
diff --git a/server/tinybird/datasources/customer_products.datasource b/server/tinybird/datasources/customer_products.datasource
new file mode 100644
index 000000000..1f9634b64
--- /dev/null
+++ b/server/tinybird/datasources/customer_products.datasource
@@ -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"
diff --git a/server/tinybird/datasources/customers.datasource b/server/tinybird/datasources/customers.datasource
new file mode 100644
index 000000000..062d97338
--- /dev/null
+++ b/server/tinybird/datasources/customers.datasource
@@ -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"
diff --git a/server/tinybird/datasources/entities.datasource b/server/tinybird/datasources/entities.datasource
new file mode 100644
index 000000000..0b5386f61
--- /dev/null
+++ b/server/tinybird/datasources/entities.datasource
@@ -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"
diff --git a/server/tinybird/datasources/entitlements.datasource b/server/tinybird/datasources/entitlements.datasource
new file mode 100644
index 000000000..d5460a802
--- /dev/null
+++ b/server/tinybird/datasources/entitlements.datasource
@@ -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"
diff --git a/server/tinybird/datasources/features.datasource b/server/tinybird/datasources/features.datasource
new file mode 100644
index 000000000..f7a7d4869
--- /dev/null
+++ b/server/tinybird/datasources/features.datasource
@@ -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"
diff --git a/server/tinybird/datasources/free_trials.datasource b/server/tinybird/datasources/free_trials.datasource
new file mode 100644
index 000000000..a4afd0fcc
--- /dev/null
+++ b/server/tinybird/datasources/free_trials.datasource
@@ -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"
diff --git a/server/tinybird/datasources/invoices.datasource b/server/tinybird/datasources/invoices.datasource
new file mode 100644
index 000000000..480e8172b
--- /dev/null
+++ b/server/tinybird/datasources/invoices.datasource
@@ -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"
\ No newline at end of file
diff --git a/server/tinybird/datasources/organizations.datasource b/server/tinybird/datasources/organizations.datasource
new file mode 100644
index 000000000..74340fa9a
--- /dev/null
+++ b/server/tinybird/datasources/organizations.datasource
@@ -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"
\ No newline at end of file
diff --git a/server/tinybird/datasources/prices.datasource b/server/tinybird/datasources/prices.datasource
new file mode 100644
index 000000000..2d632382e
--- /dev/null
+++ b/server/tinybird/datasources/prices.datasource
@@ -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"
diff --git a/server/tinybird/datasources/products.datasource b/server/tinybird/datasources/products.datasource
new file mode 100644
index 000000000..ad45bf153
--- /dev/null
+++ b/server/tinybird/datasources/products.datasource
@@ -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"
diff --git a/server/tinybird/datasources/replaceables.datasource b/server/tinybird/datasources/replaceables.datasource
new file mode 100644
index 000000000..c0bff1a0a
--- /dev/null
+++ b/server/tinybird/datasources/replaceables.datasource
@@ -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"
diff --git a/server/tinybird/datasources/rollovers.datasource b/server/tinybird/datasources/rollovers.datasource
new file mode 100644
index 000000000..cf6099efe
--- /dev/null
+++ b/server/tinybird/datasources/rollovers.datasource
@@ -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"
diff --git a/server/tinybird/datasources/subscriptions.datasource b/server/tinybird/datasources/subscriptions.datasource
new file mode 100644
index 000000000..6e54ce583
--- /dev/null
+++ b/server/tinybird/datasources/subscriptions.datasource
@@ -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"
diff --git a/server/tinybird/scripts/backfill_base.ts b/server/tinybird/scripts/backfill_base.ts
new file mode 100644
index 000000000..b1d582ba2
--- /dev/null
+++ b/server/tinybird/scripts/backfill_base.ts
@@ -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`);
+}
diff --git a/server/tinybird/scripts/backfill_cli.ts b/server/tinybird/scripts/backfill_cli.ts
new file mode 100644
index 000000000..b7df3ab7f
--- /dev/null
+++ b/server/tinybird/scripts/backfill_cli.ts
@@ -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);
+});
diff --git a/server/tinybird/scripts/backfill_events.ts b/server/tinybird/scripts/backfill_events.ts
index 40ba07066..fecb9a310 100644
--- a/server/tinybird/scripts/backfill_events.ts
+++ b/server/tinybird/scripts/backfill_events.ts
@@ -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() ?? "";
 	}
diff --git a/server/tinybird/scripts/backfill_rollovers.ts b/server/tinybird/scripts/backfill_rollovers.ts
new file mode 100644
index 000000000..37193e77e
--- /dev/null
+++ b/server/tinybird/scripts/backfill_rollovers.ts
@@ -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);
+});