Stale Event Processing Delay Sentinel

id: stale-event-processing-delay-sentinel
namespace: company.team

description: |
  Reusable Kestra Blueprint for detecting stale or delayed events.

  The workflow receives an event through a webhook, validates its timestamp,
  evaluates the processing delay against a configurable freshness threshold,
  logs whether the event is fresh or stale, and generates a structured
  monitoring event.

inputs:
  - id: max_delay_minutes
    type: INT
    defaults: 15
    description: Maximum allowed event processing delay in minutes.

tasks:
  - id: inspect_event
    type: io.kestra.plugin.core.log.Log
    message: |
      STALE EVENT PROCESSING SENTINEL

      Event received successfully.

      Event payload:
      {{ trigger.body }}

  - id: validate_timestamp
    type: io.kestra.plugin.core.flow.If
    condition: "{{ trigger.body.event_timestamp is not null }}"

    then:
      - id: evaluate_delay
        type: io.kestra.plugin.core.flow.If
        condition: >-
          {{
            (
              trigger.body.event_timestamp
              | date(existingFormat="iso_zoned_date_time")
              | timestampMilli
            )
            <
            (
              now()
              | dateAdd(-inputs.max_delay_minutes, "MINUTES")
              | timestampMilli
            )
          }}

        then:
          - id: stale_event_detected
            type: io.kestra.plugin.core.log.Log
            message: |
              STALE EVENT DETECTED

              Event timestamp:
              {{ trigger.body.event_timestamp }}

              Maximum allowed delay:
              {{ inputs.max_delay_minutes }} minutes

              Current time:
              {{ now() }}

              The event is older than the allowed processing delay.

          - id: generate_stale_event
            type: io.kestra.plugin.datagen.core.Generate
            generator:
              type: io.kestra.plugin.datagen.generators.JsonObjectGenerator
              value:
                event_type: "stale_event_detected"
                event_timestamp: "{{ trigger.body.event_timestamp }}"
                max_delay_minutes: "{{ inputs.max_delay_minutes }}"
                execution_id: "{{ execution.id }}"
                check_time: "{{ execution.startDate }}"

        else:
          - id: fresh_event_detected
            type: io.kestra.plugin.core.log.Log
            message: |
              FRESH EVENT DETECTED

              Event timestamp:
              {{ trigger.body.event_timestamp }}

              Maximum allowed delay:
              {{ inputs.max_delay_minutes }} minutes

              Current time:
              {{ now() }}

              The event is within the allowed processing delay.

          - id: generate_fresh_event
            type: io.kestra.plugin.datagen.core.Generate
            generator:
              type: io.kestra.plugin.datagen.generators.JsonObjectGenerator
              value:
                event_type: "fresh_event_detected"
                event_timestamp: "{{ trigger.body.event_timestamp }}"
                max_delay_minutes: "{{ inputs.max_delay_minutes }}"
                execution_id: "{{ execution.id }}"
                check_time: "{{ execution.startDate }}"

    else:
      - id: missing_event_timestamp
        type: io.kestra.plugin.core.log.Log
        message: |
          STALE EVENT SENTINEL WARNING

          The received event does not contain an event_timestamp.
          The event cannot be evaluated for processing delay.

triggers:
  - id: event_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: stale-event-sentinel