Clay CRM Lead Enrichment Dispatch and Delivery Sentinel

id: clay-crm-lead-enrichment-dispatch-and-delivery-sentinel
namespace: company.growth

inputs:
  - id: sample_lead_count
    type: INT
    displayName: Sample Leads Count
    description: Number of qualified test leads dispatched when executing in
      standalone verification mode.
    defaults: 3

  - id: chunk_size
    type: INT
    displayName: Chunk Size
    description: Maximum rows sent in each batch request to Clay (1 to 5,000).
      Defaults to 100 per Clay rate limits.
    defaults: 100

  - id: fail_on_partial_error
    type: BOOL
    displayName: Fail on Partial Error
    description: When false, records failed chunks in task output and allows
      downstream sentinel logic to inspect delivery degradation.
    defaults: false

tasks:
  - id: prepare_qualified_leads
    type: io.kestra.plugin.core.output.OutputValues
    description: Formats and prepares qualified inbound prospect leads for Clay
      enrichment waterfalls.
    values:
      qualified_leads:
        - email: "alex.rivera@enterprisecorp.io"
          company: "EnterpriseCorp"
          first_name: "Alex"
          last_name: "Rivera"
          source: "inbound_demo_request"
        - email: "sam.taylor@growthscale.com"
          company: "GrowthScale Systems"
          first_name: "Sam"
          last_name: "Taylor"
          source: "product_signup"
        - email: "morgan.chen@cloudnexus.dev"
          company: "CloudNexus Labs"
          first_name: "Morgan"
          last_name: "Chen"
          source: "webinar_attendee"

  - id: push_leads_to_clay
    type: io.kestra.plugin.clay.PushRows
    description: Sends batch records to the Clay table via its webhook endpoint with
      exponential retry on HTTP 429 and 5xx errors.
    allowFailure: true
    webhookUrl: "{{ secret('CLAY_WEBHOOK_URL') }}"
    authToken: "{{ secret('CLAY_AUTH_TOKEN') }}"
    rows: "{{ outputs.prepare_qualified_leads.values.qualified_leads }}"
    chunkSize: "{{ inputs.chunk_size }}"
    failOnPartialError: "{{ inputs.fail_on_partial_error }}"

  - id: evaluate_delivery_health
    type: io.kestra.plugin.core.output.OutputValues
    description: Compares accepted rows and chunk failure arrays to classify
      enrichment pipeline delivery health.
    values:
      accepted_rows: "{{ outputs.push_leads_to_clay.rowCount ?? 0 }}"
      submitted_chunks: "{{ outputs.push_leads_to_clay.chunkCount ?? 0 }}"
      failed_chunk_count: "{{ outputs.push_leads_to_clay.failedChunks is defined ?
        outputs.push_leads_to_clay.failedChunks | length : 0 }}"
      delivery_status: "{{ outputs.push_leads_to_clay is not defined or
        (outputs.push_leads_to_clay.failedChunks is defined and
        (outputs.push_leads_to_clay.failedChunks | length) > 0) ? 'DEGRADED' :
        ((outputs.push_leads_to_clay.rowCount ?? 0) > 0 ? 'HEALTHY' : 'EMPTY')
        }}"

  - id: route_delivery_findings
    type: io.kestra.plugin.core.flow.If
    description: Routes execution flow based on delivery status to alert RevOps or
      log operational success.
    condition: "{{ outputs.evaluate_delivery_health.values.delivery_status ==
      'DEGRADED' }}"
    then:
      - id: alert_delivery_failure
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alerts RevOps engineers when Clay lead dispatch fails or encounters
          rejected chunks.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "attachments": [
              {
                "color": "#E53935",
                "title": "[ALERT] Clay Lead Delivery Stalled or Partial Chunks Failed",
                "text": "Clay enrichment webhook returned delivery degradation.\n*Accepted Rows:* {{ outputs.evaluate_delivery_health.values.accepted_rows }}\n*Submitted Chunks:* {{ outputs.evaluate_delivery_health.values.submitted_chunks }}\n*Failed Chunks Count:* {{ outputs.evaluate_delivery_health.values.failed_chunk_count }}\n*Status:* {{ outputs.evaluate_delivery_health.values.delivery_status }}\nVerify Clay table capacity, webhook URL token permissions, and column mappings.",
                "footer": "Kestra Sentinel | Clay Lead Enrichment Sentinel"
              }
            ]
          }

      - id: fail_on_breach
        type: io.kestra.plugin.core.execution.Fail
        description: Explicitly marks the execution failed to trigger on-call visibility
          and SLA tracking.

    else:
      - id: log_successful_delivery
        type: io.kestra.plugin.core.log.Log
        description: Emits structured audit log for successful Clay lead enrichment batches.
        message: "[OK] Clay lead enrichment dispatch verified. Accepted rows: {{
          outputs.evaluate_delivery_health.values.accepted_rows }}, Chunks: {{
          outputs.evaluate_delivery_health.values.submitted_chunks }}, Failed
          Chunks: {{ outputs.evaluate_delivery_health.values.failed_chunk_count
          }}."

triggers:
  - id: hourly_enrichment_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"