FTPS Partner File Ingestion Stall & Stale Quarantine Sentinel

id: ftps-partner-file-ingestion-stall-and-quarantine-sentinel
namespace: company.data

inputs:
  - id: ftps_host
    type: STRING
    defaults: "localhost"
    description: Hostname or IP address of the target FTPS server.

  - id: ftps_port
    type: STRING
    defaults: "990"
    description: Port number for the FTPS service (typically 990 for implicit or 21
      for explicit TLS).

  - id: incoming_directory
    type: STRING
    defaults: "/incoming"
    description: Remote directory path where external partners upload incoming batch files.

  - id: quarantine_directory
    type: STRING
    defaults: "/quarantine"
    description: Remote directory path where unprocessable or stale files are isolated.

  - id: file_pattern
    type: STRING
    defaults: ".*\\.(edi|csv|dat|xml)"
    description: Regular expression matching partner file names to monitor and audit.

  - id: backlog_critical_threshold
    type: INT
    defaults: 50
    description: Maximum acceptable count of unread files before declaring an
      ingestion queue stall.

  - id: auto_quarantine
    type: BOOL
    defaults: false
    description: When true, automatically moves stale or stalled files to the
      quarantine directory.

  - id: dry_run
    type: BOOL
    defaults: false
    description: When true, simulates quarantine actions without moving files on the
      FTPS server.

  - id: slack_channel
    type: STRING
    defaults: "#ops-alerts"
    description: Slack channel destination for operational incident notifications.

triggers:
  - id: periodic_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/30 * * * *"
    disabled: true
    inputs:
      ftps_host: "localhost"
      ftps_port: "990"
      incoming_directory: "/incoming"
      quarantine_directory: "/quarantine"
      file_pattern: ".*\\.(edi|csv|dat|xml)"
      backlog_critical_threshold: 50
      auto_quarantine: false
      dry_run: false
      slack_channel: "#ops-alerts"

tasks:
  - id: probe_incoming_files
    type: io.kestra.plugin.fs.ftps.List
    host: "{{ inputs.ftps_host }}"
    port: "{{ inputs.ftps_port }}"
    username: "{{ secret('FTPS_USERNAME') }}"
    password: "{{ secret('FTPS_PASSWORD') }}"
    from: "{{ inputs.incoming_directory }}"
    regExp: "{{ inputs.file_pattern }}"
    sort: LAST_MODIFIED_ASC
    maxFiles: 200
    insecureTrustAllCertificates: true
    allowFailure: true
    description: Probes the FTPS incoming partner directory, retrieving metadata and
      timestamps for queued files.

  - id: classify_ingestion_health
    type: io.kestra.plugin.core.debug.Return
    format: >-
      {{
        (outputs.probe_incoming_files is not defined or outputs.probe_incoming_files.files is not defined) ? 'PROBE_FAILED' :
        (outputs.probe_incoming_files.files | length >= inputs.backlog_critical_threshold) ? 'CRITICAL_BACKLOG' :
        (outputs.probe_incoming_files.files | length > 0) ? 'ACTIVE_QUEUE' :
        'HEALTHY_EMPTY'
      }}
    description: Evaluates queue depth and file counts against operational SLA thresholds.

  - id: route_by_health_status
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.classify_ingestion_health.value }}"
    cases:
      CRITICAL_BACKLOG:
        - id: alert_critical_backlog_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[CRITICAL] FTPS Partner Ingestion Stall Detected\n• Endpoint: {{ inputs.ftps_host }}:{{ inputs.ftps_port }}\n• Directory: {{ inputs.incoming_directory }}\n• Queued Files: {{ outputs.probe_incoming_files.files | length }} (Threshold: {{ inputs.backlog_critical_threshold }})\n• Execution ID: {{ execution.id }}\nAction Required: Downstream processing consumer is stalled. Inspect consumer logs before partner files are overwritten."
            }

        - id: check_remediation
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.auto_quarantine == true and inputs.dry_run == false }}"
          then:
            - id: quarantine_stalled_files
              type: io.kestra.plugin.core.flow.Loop
              values: "{{ outputs.probe_incoming_files.files }}"
              tasks:
                - id: move_to_quarantine
                  type: io.kestra.plugin.fs.ftps.Move
                  host: "{{ inputs.ftps_host }}"
                  port: "{{ inputs.ftps_port }}"
                  username: "{{ secret('FTPS_USERNAME') }}"
                  password: "{{ secret('FTPS_PASSWORD') }}"
                  from: "{{ item.value.path }}"
                  to: "{{ inputs.quarantine_directory }}/"
                  insecureTrustAllCertificates: true

            - id: notify_quarantine_action_slack
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "channel": "{{ inputs.slack_channel }}",
                  "text": "[REMEDIATION] Stalled partner files have been moved to {{ inputs.quarantine_directory }} to unblock the ingestion pipeline."
                }

      ACTIVE_QUEUE:
        - id: log_active_queue
          type: io.kestra.plugin.core.log.Log
          message: "FTPS ingestion queue nominal. Currently processing {{
            outputs.probe_incoming_files.files | length }} file(s) in {{
            inputs.incoming_directory }}."

      HEALTHY_EMPTY:
        - id: log_healthy_empty
          type: io.kestra.plugin.core.log.Log
          message: "FTPS partner dropzone is clear on {{ inputs.ftps_host }}. Zero pending
            files in {{ inputs.incoming_directory }}."

      PROBE_FAILED:
        - id: alert_probe_failed_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[ALERT] FTPS Connection Failure\n• Target Host: {{ inputs.ftps_host }}:{{ inputs.ftps_port }}\n• Execution ID: {{ execution.id }}\nUnable to connect to FTPS server or TLS negotiation failed. Verify network routing and credentials."
            }
    description: Dispatches high-priority Slack notifications or triggers safe
      quarantine remediation based on queue status.

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    message: "FTPS sentinel execution {{ execution.id }} completed. Verdict: {{
      outputs.classify_ingestion_health.value }}, Files evaluated: {{
      outputs.probe_incoming_files.files | length ?? 0 }}, Remediation
      triggered: {{ inputs.auto_quarantine and not inputs.dry_run }}."
    description: Records an immutable audit log entry documenting execution metrics
      and remediation decisions.

errors:
  - id: alert_flow_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": "[ERROR] FTPS Sentinel Workflow Failure\nFlow execution {{ execution.id }} encountered an unhandled exception against {{ inputs.ftps_host }}. Review Kestra task logs."
      }