Send CRM lead stages to Meta Conversions API (Conversion Leads)

id: meta-capi-crm-conversion-leads
namespace: company.team

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: dataset_id
    type: STRING
    displayName: Meta dataset (pixel) ID
    description: ID of the Meta dataset that receives the CRM events.
    defaults: "123456789012345"

  - id: lead_event_source
    type: STRING
    displayName: CRM name
    description: Sent as `custom_data.lead_event_source`, the name of the tool the
      leads come from. Required by Meta.
    defaults: In-house CRM

  - id: test_event_code
    type: STRING
    displayName: Test event code
    description: >
      Optional code from Events Manager > Test events. With it, the batch goes
      to the Test events tab and the rows stay pending, so they are still sent
      for real on the next scheduled run.
    required: false

  - id: graph_api_url
    type: STRING
    displayName: Graph API base URL
    defaults: https://graph.facebook.com/v26.0

tasks:
  - id: check_inputs
    type: io.kestra.plugin.core.execution.Fail
    condition: "{{ inputs.lead_event_source | trim is empty }}"
    errorMessage: "`lead_event_source` is required by Meta on every Conversion Leads event."

  - id: prepare_batch
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    description: >
      One statement normalizes the pending rows following Meta's rules, takes
      out the ones that would make Meta reject the whole batch, and builds up to
      1,000 events (the API limit). Email and phone are hashed here, so raw PII
      never reaches Kestra outputs or logs.
    fetchType: FETCH_ONE
    sql: |
      WITH normalized AS (
        SELECT
          id,
          event_time,
          nullif(btrim(regexp_replace(event_name, '[[:space:][:cntrl:]\u00a0]+', ' ', 'g')), '') AS event_name,
          -- Meta: lead IDs are 15 to 17 digits
          CASE WHEN lead_id BETWEEN 100000000000000 AND 99999999999999999 THEN lead_id END AS lead_id,
          -- Meta: trim and lowercase; any whitespace, including invisible characters, would change the hash
          CASE WHEN e ~ '^[^@]+@[^@]+$' THEN e END AS email,
          -- Meta: digits only, no leading zeros, country code included
          CASE WHEN length(p) BETWEEN 8 AND 15 THEN p END AS phone,
          CASE WHEN value BETWEEN 0 AND 1e12 AND upper(btrim(currency)) ~ '^[A-Z]{3}$' THEN value END AS value,
          CASE WHEN value BETWEEN 0 AND 1e12 AND upper(btrim(currency)) ~ '^[A-Z]{3}$' THEN upper(btrim(currency)) END AS currency
        FROM meta_capi_outbox
        CROSS JOIN LATERAL (
          SELECT
            lower(regexp_replace(coalesce(email, ''), '[[:space:]\u00a0\u200b\ufeff]', '', 'g')) AS e,
            ltrim(regexp_replace(coalesce(phone, ''), '\D', '', 'g'), '0') AS p
        ) AS cleaned
        WHERE status = 'pending'
      ),
      classified AS (
        SELECT *,
          CASE
            -- Meta rejects the batch if one event_time is more than 7 days old; keep a 1-hour margin
            WHEN event_time < now() - interval '7 days' + interval '1 hour' THEN 'expired'
            WHEN event_name IS NULL OR (lead_id IS NULL AND email IS NULL AND phone IS NULL) THEN 'invalid'
            ELSE 'sendable'
          END AS verdict
        FROM normalized
      ),
      discarded AS (
        UPDATE meta_capi_outbox AS o
        SET status = c.verdict, updated_at = now()
        FROM classified AS c
        WHERE o.id = c.id AND c.verdict <> 'sendable'
        RETURNING o.status
      ),
      batch AS (
        SELECT *
        FROM classified
        WHERE verdict = 'sendable'
        ORDER BY event_time
        LIMIT 1000
      )
      SELECT
        (SELECT count(*) FROM batch) AS event_count,
        (SELECT count(*) FROM discarded WHERE status = 'expired') AS expired_count,
        (SELECT count(*) FROM discarded WHERE status = 'invalid') AS invalid_count,
        (SELECT coalesce(string_agg(id::text, ','), '') FROM batch) AS ids,
        (SELECT coalesce(json_agg(jsonb_strip_nulls(jsonb_build_object(
          'event_name', event_name,
          'event_time', extract(epoch FROM least(event_time, now()))::bigint,
          'event_id', 'crm-' || id,
          'action_source', 'system_generated',
          'user_data', jsonb_build_object(
            'lead_id', lead_id,
            'em', CASE WHEN email IS NOT NULL THEN jsonb_build_array(encode(sha256(convert_to(email, 'UTF8')), 'hex')) END,
            'ph', CASE WHEN phone IS NOT NULL THEN jsonb_build_array(encode(sha256(convert_to(phone, 'UTF8')), 'hex')) END
          ),
          'custom_data', jsonb_build_object(
            'event_source', 'crm',
            'lead_event_source', '{{ inputs.lead_event_source | trim | replace({"'": "''"}) }}',
            'value', value,
            'currency', currency
          )
        )) ORDER BY event_time), '[]')::text FROM batch) AS payload;

  - id: has_events
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.prepare_batch.row.event_count > 0 }}"
    then:
      - id: send_events
        type: io.kestra.plugin.core.http.Request
        description: One request per run with the whole batch. The token goes in the
          body, not in the URL.
        uri: "{{ inputs.graph_api_url }}/{{ inputs.dataset_id }}/events"
        method: POST
        contentType: application/json
        body: |
          {
            "data": {{ outputs.prepare_batch.row.payload }},
            {% if inputs.test_event_code is not empty %}"test_event_code": {{ inputs.test_event_code | toJson }},{% endif %}
            "access_token": "{{ secret('META_ACCESS_TOKEN') }}"
          }

      - id: mark_sent
        type: io.kestra.plugin.jdbc.postgresql.Query
        url: "{{ secret('POSTGRES_URL') }}"
        username: "{{ secret('POSTGRES_USER') }}"
        password: "{{ secret('POSTGRES_PASSWORD') }}"
        description: >
          Only the ids of this batch are updated, so rows inserted while the
          request was in flight stay pending. Test runs skip this step: Test
          events do not count as production events.
        runIf: "{{ inputs.test_event_code is empty }}"
        fetchType: NONE
        sql: |
          UPDATE meta_capi_outbox
          SET status = 'sent',
              sent_at = now(),
              updated_at = now(),
              fbtrace_id = '{{ outputs.send_events.body | jq('.fbtrace_id // ""') | first }}'
          WHERE id = ANY(string_to_array('{{ outputs.prepare_batch.row.ids }}', ',')::bigint[]);

      - id: log_result
        type: io.kestra.plugin.core.log.Log
        message: >
          {% if inputs.test_event_code is not empty %}Test run, rows left
          pending. {% endif %}Meta received {{ outputs.send_events.body |
          jq('.events_received') | first }} of {{
          outputs.prepare_batch.row.event_count }} events (fbtrace_id {{
          outputs.send_events.body | jq('.fbtrace_id // ""') | first }}); {{
          outputs.prepare_batch.row.expired_count }} expired and {{
          outputs.prepare_batch.row.invalid_count }} invalid row(s) were taken
          out.
    else:
      - id: nothing_to_send
        type: io.kestra.plugin.core.log.Log
        message: >
          No CRM events to send; {{ outputs.prepare_batch.row.expired_count }}
          expired and {{ outputs.prepare_batch.row.invalid_count }} invalid
          row(s) were taken out.

errors:
  - id: log_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: >
      The run failed. Rows not marked as sent stay pending and go out again on
      the next run. If send_events failed, its logs contain Meta's error: one
      invalid event makes Meta reject the whole batch.

triggers:
  - id: every_15_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"