SFTP Stale File Quarantine Gate

id: sftp-stale-file-quarantine-gate
namespace: company.team
description: |
  List an SFTP dropzone on a schedule, flag any regular file whose last
  modified time is older than a staleness threshold, classify the result
  as CLEAN, STALE, or UNREACHABLE, alert Slack the moment a file has sat
  unpicked, and optionally move stale files into a quarantine directory
  with Move, then re-list the dropzone to verify none remain.

inputs:
  - id: sftp_host
    type: STRING
    displayName: SFTP host
    description: Hostname or container name of the SFTP server, reachable from the
      Kestra Worker's Docker network.
    defaults: "sftp-dropzone"

  - id: sftp_port
    type: STRING
    displayName: SFTP port
    description: SFTP port on the remote server.
    defaults: "22"

  - id: sftp_username
    type: STRING
    displayName: SFTP username
    description: Username used to authenticate against the SFTP server.
    defaults: "kestra"

  - id: sftp_dropzone_path
    type: STRING
    displayName: Dropzone directory
    description: Absolute path on the SFTP server to list for stale files.
    defaults: "/upload/inbound"

  - id: sftp_quarantine_path
    type: STRING
    displayName: Quarantine directory
    description: Absolute path stale files are moved into when auto_quarantine is true.
    defaults: "/upload/quarantine"

  - id: stale_after_minutes
    type: INT
    displayName: Stale threshold in minutes
    description: A regular file whose last modified time is older than this many
      minutes is classified STALE.
    defaults: 60

  - id: auto_quarantine
    type: BOOL
    displayName: Auto quarantine
    description: Allow the Move tasks to run.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_quarantine, for Move to run.
    defaults: true

tasks:
  - id: probe_dropzone
    type: io.kestra.plugin.fs.sftp.List
    description: Read-only. Lists the dropzone directory. A connection, auth, or
      path failure does not stop the flow.
    host: "{{ inputs.sftp_host }}"
    port: "{{ inputs.sftp_port }}"
    username: "{{ inputs.sftp_username }}"
    password: "{{ secret('SFTP_PASSWORD') }}"
    from: "{{ inputs.sftp_dropzone_path }}"
    rootDir: false
    recursive: false
    allowFailure: true

  - id: stale_findings
    type: io.kestra.plugin.core.output.OutputValues
    description: Read-only. Filters the listing to regular files (fileType FILE)
      whose updatedDate is older than stale_after_minutes.
    values:
      items: >-
        {{ {'files': (outputs.probe_dropzone is defined and
        outputs.probe_dropzone.files is defined ? outputs.probe_dropzone.files :
        []), 'threshold': inputs.stale_after_minutes} | jq('.threshold as $t |
        [.files[] | select(.fileType == "FILE") | . + {minutesStale: (((now -
        (.updatedDate | split(".")[0] | if endswith("Z") then . else . + "Z" end
        | fromdateiso8601)) / 60) | floor)}] | map(select(.minutesStale > $t)) |
        map({name: .name, path: .path, minutesStale: .minutesStale})') | first |
        toJson }}

  - id: classify_status
    type: io.kestra.plugin.core.debug.Return
    description: UNREACHABLE when the dropzone listing has no files field, STALE
      when at least one file exceeds stale_after_minutes, otherwise CLEAN.
    format: "{{ outputs.probe_dropzone.files is defined ?
      (fromJson(outputs.stale_findings.values.items) | length > 0 ? 'STALE' :
      'CLEAN') : 'UNREACHABLE' }}"

  - id: route_by_status
    type: io.kestra.plugin.core.flow.Switch
    description: Log a quiet confirmation when clean, page Slack and optionally
      quarantine when stale, or page Slack when the dropzone could not be
      listed.
    value: "{{ outputs.classify_status.value }}"
    cases:
      "CLEAN":
        - id: log_clean
          type: io.kestra.plugin.core.log.Log
          message: "No file in {{ inputs.sftp_dropzone_path }} on {{ inputs.sftp_host }}
            is older than {{ inputs.stale_after_minutes }} minutes."
      "STALE":
        - id: alert_stale
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "SFTP stale file gate STALE: {{ fromJson(outputs.stale_findings.values.items) | length }} file(s) in {{ inputs.sftp_dropzone_path }} on {{ inputs.sftp_host }} have sat past {{ inputs.stale_after_minutes }} minutes ({{ fromJson(outputs.stale_findings.values.items) | jq('map(\"\\(.name) (\\(.minutesStale)m)\") | join(\", \")') | first }}). {{ inputs.auto_quarantine and not inputs.dry_run ? 'Quarantining now.' : 'No file moved (auto_quarantine=' ~ inputs.auto_quarantine ~ ', dry_run=' ~ inputs.dry_run ~ ').' }} Execution {{ execution.id }}."
            }

        - id: quarantine_gate
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.auto_quarantine and not inputs.dry_run }}"
          then:
            - id: quarantine_loop
              type: io.kestra.plugin.core.flow.Loop
              description: One Move per stale file, one SFTP session at a time.
              concurrencyLimit: 1
              values: "{{ outputs.stale_findings.values.items }}"
              tasks:
                - id: move_to_quarantine
                  type: io.kestra.plugin.fs.sftp.Move
                  host: "{{ inputs.sftp_host }}"
                  port: "{{ inputs.sftp_port }}"
                  username: "{{ inputs.sftp_username }}"
                  password: "{{ secret('SFTP_PASSWORD') }}"
                  from: "{{ item.value | jq('.path') | first }}"
                  to: "{{ inputs.sftp_quarantine_path }}/{{ item.value | jq('.name') | first }}"
                  overwrite: false

            - id: reread_dropzone
              type: io.kestra.plugin.fs.sftp.List
              host: "{{ inputs.sftp_host }}"
              port: "{{ inputs.sftp_port }}"
              username: "{{ inputs.sftp_username }}"
              password: "{{ secret('SFTP_PASSWORD') }}"
              from: "{{ inputs.sftp_dropzone_path }}"
              rootDir: false
              recursive: false
              allowFailure: true

            - id: verify_quarantine
              type: io.kestra.plugin.core.flow.If
              description: Fails if any file that was supposed to be quarantined is still
                present in the dropzone after the Move tasks ran.
              condition: "{{ {'moved_names': (fromJson(outputs.stale_findings.values.items) |
                jq('map(.name)') | first), 'remaining_names':
                (outputs.reread_dropzone.files ?? [] | jq('map(.name)') |
                first)} | jq('(.moved_names - (.moved_names - .remaining_names))
                | length > 0') | first }}"
              then:
                - id: quarantine_failed
                  type: io.kestra.plugin.core.execution.Fail
                  errorMessage: "Move ran, but at least one flagged file is still present in the
                    dropzone after quarantine."
              else:
                - id: quarantined
                  type: io.kestra.plugin.core.log.Log
                  message: "Quarantined {{ fromJson(outputs.stale_findings.values.items) | length
                    }} stale file(s) from {{ inputs.sftp_dropzone_path }} to {{
                    inputs.sftp_quarantine_path }} and verified none remain in
                    the dropzone."
          else:
            - id: not_quarantined
              type: io.kestra.plugin.core.log.Log
              message: "Stale files found but not moved (auto_quarantine={{
                inputs.auto_quarantine }}, dry_run={{ inputs.dry_run }})."
      "UNREACHABLE":
        - id: alert_unreachable
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": "SFTP stale file gate UNREACHABLE: listing {{ inputs.sftp_dropzone_path }} on {{ inputs.sftp_host }}:{{ inputs.sftp_port }} returned no files field (connection refused, auth failure, or path does not exist). Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        message: "Completed SFTP stale file audit with status: {{
          outputs.classify_status.value }}"

  - id: log_audit_status
    type: io.kestra.plugin.core.log.Log
    description: Post-audit status line recorded on every run regardless of branch,
      for a continuous dropzone trend in the execution history.
    message: "SFTP stale file audit: host={{ inputs.sftp_host }}, path={{
      inputs.sftp_dropzone_path }}, stale_after_minutes={{
      inputs.stale_after_minutes }}, status={{ outputs.classify_status.value }},
      execution={{ execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the gate flow itself fails, so a broken audit
      never silently stops monitoring the dropzone.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "SFTP stale file gate flow {{ flow.id }} FAILED (execution {{ execution.id }}). Dropzone status is now unknown; check the task logs and the SFTP server manually."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ outputs.classify_status.value }}"

triggers:
  - id: schedule_every_15_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: Audit the dropzone every 15 minutes, always in safe mode.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      sftp_host: "sftp-dropzone"
      sftp_port: "22"
      sftp_username: "kestra"
      sftp_dropzone_path: "/upload/inbound"
      sftp_quarantine_path: "/upload/quarantine"
      stale_after_minutes: 60
      auto_quarantine: false
      dry_run: true