projection.ex

defmodule Truely.EventsRepo.Migrations.AddSessionUsageProjection do
  use Ecto.Migration

  def up do
    cluster_name = System.get_env("CLICKHOUSE_CLUSTER") || nil
    cluster_stmt = if cluster_name, do: "ON CLUSTER #{cluster_name}", else: ""

    execute """
    ALTER TABLE core_subscriptions_events_local #{cluster_stmt}
    ADD PROJECTION session_usage_by_esim
    (
      SELECT 
        iccid,
        session_id,
        toStartOfDay(timestamp) AS day,
        min(timestamp) AS session_start,
        max(timestamp) AS session_end,
        sum(uplink_bytes) AS total_uplink_bytes,
        sum(downlink_bytes) AS total_downlink_bytes,
        sum(uplink_bytes + downlink_bytes) AS total_bytes,
        sum(input_packets) AS total_input_packets,
        sum(output_packets) AS total_output_packets,
        count() AS event_count,
        uniq(event_type) AS unique_event_types,
        uniq(core_country_id) AS countries_visited,
        uniq(core_mobile_network_operator_id) AS operators_used,
        max(session_seconds) AS max_session_seconds,
        any(rattype) AS rat_type,
        any(apn) AS apn,
        any(core_subscription_id) AS core_subscription_id,
        any(core_esim_id) AS core_esim_id
      GROUP BY 
        iccid,
        session_id,
        day
      ORDER BY 
        iccid,
        day,
        session_id
    )
    """

    execute """
    ALTER TABLE core_subscriptions_events_local #{cluster_stmt}
    MATERIALIZE PROJECTION session_usage_by_esim
    """
  end

  def down do
    cluster_name = System.get_env("CLICKHOUSE_CLUSTER") || nil
    cluster_stmt = if cluster_name, do: "ON CLUSTER #{cluster_name}", else: ""

    execute """
    ALTER TABLE core_subscriptions_events_local #{cluster_stmt}
    DROP PROJECTION IF EXISTS session_usage_by_esim
    """
  end
end