sources:
  # One row per change to ecommerce.order_items, plus a `_gs_op` column
  # ('i' = insert/initial copy, 'u' = update, 'd' = delete).
  #
  # emit_update_before_row makes every UPDATE arrive as two rows: a 'd' carrying
  # the old image, then a 'u' carrying the new one. That is what lets the
  # transform below retract the state an order item used to contribute. It
  # requires ALTER TABLE ecommerce.order_items REPLICA IDENTITY FULL.
  ecommerce.order_items:
    type: postgres_cdc_source
    primary_key: order_item_id
    options:
      host: localhost
      port: "5432"
      database: shop
      username: postgres
      password: postgres
      table: ecommerce.order_items
      publication_name: streamling_order_items
      slot_name: streamling_order_items
      emit_update_before_row: "true"
      # Local machines often sit above etl's memory-backpressure threshold,
      # which pauses replication after the initial copy. See the plugin README.
      memory_backpressure_enabled: "false"

transforms:
  # Turns each CDC row into a signed delta. Additions carry sign = +1,
  # retractions of a prior state carry sign = -1, and the two cancel out for
  # any part of an order item that did not change.
  revenue_deltas:
    type: sql
    primary_key: revenue_date,product_id,currency
    sql: |
      SELECT
          CAST(ordered_at AS DATE)                       AS revenue_date,
          product_id,
          currency,

          sign                                           AS orders_delta,
          sign * quantity                                AS units_delta,

          CAST(sign * quantity * unit_price AS DECIMAL(18, 2))
              AS gross_revenue_delta,
          CAST(sign * discount_amount AS DECIMAL(18, 2))
              AS discount_delta,
          CAST(sign * refund_amount AS DECIMAL(18, 2))
              AS refund_delta,
          CAST(sign * (quantity * unit_price - discount_amount - refund_amount)
               AS DECIMAL(18, 2))
              AS net_revenue_delta,

          -- Every delta is appended to the SummingMergeTree table, including the
          -- negative ones. A retraction is a negative row, never a ClickHouse
          -- delete, so the sink must see 'i' on all of them.
          'i'                                            AS _gs_op
      FROM (
          SELECT
              ordered_at,
              product_id,
              currency,
              quantity,
              -- Postgres NUMERIC arrives as a string, so cast it explicitly.
              CAST(unit_price AS DECIMAL(18, 2))      AS unit_price,
              CAST(discount_amount AS DECIMAL(18, 2)) AS discount_amount,
              CAST(refund_amount AS DECIMAL(18, 2))   AS refund_amount,
              CASE WHEN _gs_op = 'd' THEN -1 ELSE 1 END AS sign
          FROM ecommerce.order_items
          -- Applied to each image independently: a pending -> paid update
          -- drops its retraction and keeps its addition, and a paid ->
          -- cancelled update does the reverse.
          WHERE status IN ('paid', 'fulfilled', 'partially_refunded', 'refunded')
      ) AS cdc

sinks:
  # append_only_mode: false keeps the sink from deriving is_deleted/insert_time
  # columns, which a SummingMergeTree table does not have. Every row arrives as
  # a plain INSERT.
  #
  # deduplicate: false is required. By default the sink collapses each batch to
  # the latest row per primary_key, which is right for upserts and wrong here:
  # a retraction and its addition share a key, and so do two order items for the
  # same product and day. Both rows have to be inserted.
  clickhouse.product_revenue_daily:
    type: clickhouse
    from: revenue_deltas
    table: product_revenue_daily
    primary_key: revenue_date,product_id,currency
    append_only_mode: false
    deduplicate: false