Ruslan Myshchyshyn
Analytics & Data Engineering

SELECT
  CAST(d.ID AS INT64) AS ID,
  DATE(PARSE_TIMESTAMP('%Y-%m-%dT%H:%M:%S%Ez', d.DATE_CREATE),
       'America/New_York') AS create_date,

  -- Stage group: semantic stage (S/P/F)
  CASE
    WHEN d.STAGE_SEMANTIC_ID = "S" THEN "Successful"
    WHEN d.STAGE_SEMANTIC_ID = "P" THEN "Progress"
    WHEN d.STAGE_SEMANTIC_ID = "F" THEN "Failure"
  END AS stage_group,

  COALESCE(stage_status.name, 'Unknown / Other') AS stage,
  COALESCE(assignee.full_name, CONCAT('ID: ', d.ASSIGNED_BY_ID))
    AS responsible_person,

  -- Normalize traffic source
  CASE
    WHEN d.UTM_SOURCE IN ('instagram','ig') THEN 'instagram'
    WHEN d.UTM_SOURCE IN ('facebook','fb')  THEN 'facebook'
    WHEN d.UTM_SOURCE = 'telegram'          THEN 'telegram'
    ELSE 'unknown'
  END AS source,

  -- Multi-value language field: [760, 762] → "English, Ukrainian"
  COALESCE(
    (SELECT STRING_AGG(le.option_value, ', ')
     FROM UNNEST(SPLIT(
       REGEXP_REPLACE(d.UF_CRM_1744805668224, r"[\[\]\s']",''),','
     )) AS lang_id
     JOIN userfield_enums le
       ON le.field_id = 'UF_CRM_1744805668224'
       AND TRIM(lang_id) = le.option_id),
    d.UF_CRM_1744805668224
  ) AS language,

  -- First payment date (earliest from payment_schedule)
  DATE(SAFE.PARSE_TIMESTAMP('%Y-%m-%dT%H:%M:%S%Ez',
       ps.first_payment_time), 'America/New_York')
    AS first_payment_date

FROM deals d
LEFT JOIN users assignee     ON d.ASSIGNED_BY_ID = assignee.id
LEFT JOIN contacts           ON CAST(d.CONTACT_ID AS STRING) = contacts.id
LEFT JOIN crm_status         ON d.STAGE_ID = stage_status.status_id
LEFT JOIN userfield_enums    ON ... -- state, location, service_package
LEFT JOIN (SELECT COALESCE(parentId2, xmlId) AS deal_id,
                  MIN(createdTime) AS first_payment_time
           FROM payment_schedule GROUP BY 1) ps
  ON CAST(d.ID AS STRING) = CAST(ps.deal_id AS STRING)

WITH
-- Deduplicate calls from incremental sync
calls_dedup AS (
  SELECT *
  FROM bitrix24.calls
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY COALESCE(
      NULLIF(TRIM(CALL_ID), ''),
      CONCAT(CALL_START_DATE,'|',PORTAL_USER_ID,'|',
             CRM_ENTITY_ID,'|',PHONE_NUMBER))
    ORDER BY CREATED_AT DESC NULLS LAST
  ) = 1
),

-- Map both CONTACT and DEAL entity types to deal attributes
deals_for_calls AS (
  SELECT 'DEAL' AS entity_type, CAST(d.ID AS STRING) AS entity_id,
    dv.location_event, dv.state, dv.stage
  FROM deals d LEFT JOIN deals_view dv ON d.ID = dv.ID

  UNION ALL

  SELECT 'CONTACT', CAST(d.CONTACT_ID AS STRING),
    dv.location_event, dv.state, dv.stage
  FROM deals d LEFT JOIN deals_view dv ON d.ID = dv.ID
  WHERE d.CONTACT_ID IS NOT NULL
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY d.CONTACT_ID ORDER BY d.DATE_MODIFY DESC
  ) = 1
)

SELECT
  -- Mixed timestamp formats → normalized datetime
  DATETIME(
    CASE
      WHEN CALL_START_DATE LIKE '%-%'
        THEN SAFE_CAST(CALL_START_DATE AS TIMESTAMP)
      ELSE SAFE.TIMESTAMP_MICROS(
             DIV(SAFE_CAST(CALL_START_DATE AS INT64), 1000))
    END, 'America/New_York') AS timestamp,

  COALESCE(u.full_name, CONCAT('ID: ', PORTAL_USER_ID)) AS manager,

  -- Call classification
  CASE WHEN call_type = 2 AND duration = 0 THEN 'Missed'
       WHEN duration < 10                  THEN 'No answer'
       ELSE 'Success' END AS call_outcome,

  duration > 120 AS call_longer_2min,
  duration > 300 AS call_longer_5min,
  deals.state, deals.location_event

FROM calls_dedup
LEFT JOIN users_dedup u    ON PORTAL_USER_ID = u.id
LEFT JOIN deals_for_calls  ON CRM_ENTITY_TYPE = entity_type
                          AND CRM_ENTITY_ID   = entity_id

WITH latest_drb AS (
  -- Latest booking state per deal
  SELECT * EXCEPT(rn) FROM (
    SELECT *, ROW_NUMBER() OVER (
      PARTITION BY deal_id ORDER BY ingested_at DESC) AS rn
    FROM deal_resourcebookings
    WHERE booking_time_json IS NOT NULL
      AND booking_time_json != '[]'
  ) WHERE rn = 1
),

-- Parse nested JSON time slots with 3 date format fallbacks
time_slots_raw AS (
  SELECT drb.deal_id, drb.hours_total,
    drb.cleaners_instruction, drb.client_name,
    slot.from_ts, slot.to_ts
  FROM latest_drb drb
  CROSS JOIN UNNEST(ARRAY(
    SELECT AS STRUCT
      COALESCE(
        SAFE.PARSE_TIMESTAMP('%Y-%m-%d %H:%M:%S',
          JSON_VALUE(el, '$.from'), 'America/New_York'),
        SAFE.PARSE_TIMESTAMP('%m/%d/%Y %H:%M:%S',
          JSON_VALUE(el, '$.from'), 'America/New_York'),
        SAFE.PARSE_TIMESTAMP('%d/%m/%Y %H:%M:%S',
          JSON_VALUE(el, '$.from'), 'America/New_York')
      ) AS from_ts,
      -- (same pattern for $.to)
      ...
    FROM UNNEST(JSON_EXTRACT_ARRAY(
      drb.booking_time_json)) AS el
  )) AS slot
),

-- Polymorphic cleaner JSON → unnested IDs
cleaner_unnested AS (
  SELECT pa.*, cu.vid, cu.name_raw
  FROM parsed_address pa
  JOIN time_slots ts USING (deal_id, booking_from)
  CROSS JOIN UNNEST(
    CASE
      WHEN JSON_TYPE(SAFE.PARSE_JSON(
             ts.assigned_cleaner_json)) = 'array'
      THEN JSON_EXTRACT_ARRAY(ts.assigned_cleaner_json)
      WHEN JSON_TYPE(...) = 'object'
      THEN JSON_EXTRACT_ARRAY(
             CONCAT('[', ts.assigned_cleaner_json, ']'))
      ELSE []
    END
  ) AS el
),

-- Resolve IDs → names via hr_users
cleaner_with_names AS (
  SELECT cu.*,
    COALESCE(
      NULLIF(TRIM(CONCAT(
        COALESCE(h.NAME,''),' ',COALESCE(h.LAST_NAME,'')
      )), ''),
      cu.name_raw
    ) AS cleaner_name
  FROM cleaner_unnested cu
  LEFT JOIN hr_users h ON h.ID = cu.vid
)

-- Final output: one row per (deal, time slot)
SELECT deal_id, booking_from,
  -- Smart end time: hours_total or Bitrix raw
  CASE WHEN hours_total IS NOT NULL
    THEN TIMESTAMP_ADD(booking_from,
      INTERVAL CAST(hours_total * 3600 AS INT64) SECOND)
    ELSE booking_to_old
  END AS booking_to,
  ANY_VALUE(all_cleaners) AS all_cleaners,
  ANY_VALUE(all_emails)   AS all_emails,
  ANY_VALUE(customer_name) AS customer_name,
  ANY_VALUE(CASE stage_id
    WHEN 'C1:EXECUTING' THEN 'Cleaning'
    WHEN 'FINAL_INVOICE' THEN 'Updates'
    WHEN 'WON'           THEN 'Close deal'
  END) AS stage_name
FROM joined GROUP BY deal_id, booking_from, booking_to_old