NFS New File Extension Compliance Gate

id: nfs-new-file-extension-compliance-gate
namespace: company.team

description: |
  Fire in real time whenever new files land on a watched NFS directory, classify how many don't
  match an approved extension pattern, quarantine the non-compliant ones only when explicitly
  authorized, alert Slack otherwise, and log a governance audit trail either way.

inputs:
  - id: allowed_extensions_regex
    type: STRING
    displayName: Allowed extensions pattern
    description: Regular expression a filename must fully match to be considered compliant.
    defaults: ".*\\.(csv|json|parquet)$"

  - id: quarantine_directory
    type: STRING
    displayName: Quarantine directory
    description: NFS directory non-compliant files are moved into when
      confirm_quarantine is true.
    defaults: "/mnt/nfs/quarantine"

  - id: confirm_quarantine
    type: BOOL
    displayName: Confirm automatic quarantine
    description: When true, this flow is authorized to move every non-compliant file
      into quarantine_directory itself. When false, a non-compliant arrival only
      alerts Slack so a human reviews it.
    defaults: false

tasks:
  - id: evaluate
    type: io.kestra.plugin.core.output.OutputValues
    description: The trigger already delivers each new file's name and metadata, so
      this classifies the batch directly from trigger.files rather than
      re-probing NFS, counting and listing the files whose name does not match
      allowed_extensions_regex.
    values:
      noncompliant_count: >-
        {{ ({'files': trigger.files, 'pattern': inputs.allowed_extensions_regex}
        | jq('.pattern as $p | [.files[] | select((.name | test($p)) | not)] |
        length')) | first | number }}
      noncompliant_names: >-
        {{ ({'files': trigger.files, 'pattern': inputs.allowed_extensions_regex}
        | jq('.pattern as $p | [.files[] | select((.name | test($p)) | not) |
        .name] | join("; ")')) | first }}
      severity: >-
        {{ ((({'files': trigger.files, 'pattern':
        inputs.allowed_extensions_regex} | jq('.pattern as $p | [.files[] |
        select((.name | test($p)) | not)] | length')) | first | number) == 0) ?
        'ALL_COMPLIANT' : (inputs.confirm_quarantine ?
        'NONCOMPLIANT_AUTO_QUARANTINE' : 'NONCOMPLIANT_NEEDS_HUMAN') }}

  - id: route
    type: io.kestra.plugin.core.flow.Switch
    description: ALL_COMPLIANT just logs. NONCOMPLIANT_AUTO_QUARANTINE performs the
      safe remediation (moving a file within the same NFS mount, nothing is
      deleted) only when confirm_quarantine authorizes it.
      NONCOMPLIANT_NEEDS_HUMAN alerts Slack instead.
    value: "{{ outputs.evaluate.values.severity }}"
    cases:
      "ALL_COMPLIANT":
        - id: log_all_compliant
          type: io.kestra.plugin.core.log.Log
          message: "{{ trigger.files | length }} new file(s) arrived, all matching {{
            inputs.allowed_extensions_regex }}."
      "NONCOMPLIANT_AUTO_QUARANTINE":
        - id: quarantine_noncompliant_files
          type: io.kestra.plugin.core.flow.Loop
          description: Iterates every file this fire delivered and quarantines only the
            ones that do not match allowed_extensions_regex, now that
            confirm_quarantine explicitly authorizes this flow to act.
          values: "{{ trigger.files }}"
          tasks:
            - id: quarantine_if_noncompliant
              type: io.kestra.plugin.core.flow.If
              condition: >-
                {{ ((({'name': fromJson(item.value).name, 'pattern':
                inputs.allowed_extensions_regex} | jq('.pattern as $p | ([.name]
                | map(select(test($p)))) | length')) | first | number)) == 0 }}
              then:
                - id: move_to_quarantine
                  type: io.kestra.plugin.fs.nfs.Move
                  from: "{{ fromJson(item.value).uri }}"
                  to: "{{ inputs.quarantine_directory }}/{{ fromJson(item.value).name }}"
        - id: alert_auto_quarantined
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {'text': 'NONCOMPLIANT_AUTO_QUARANTINE: ' ~ outputs.evaluate.values.noncompliant_count ~ ' of ' ~ (trigger.files | length) ~ ' new file(s) did not match ' ~ inputs.allowed_extensions_regex ~ ': ' ~ outputs.evaluate.values.noncompliant_names ~ '. Moved them into ' ~ inputs.quarantine_directory ~ ' automatically.'} | toJson }}
      "NONCOMPLIANT_NEEDS_HUMAN":
        - id: alert_noncompliant
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {'text': 'NONCOMPLIANT_NEEDS_HUMAN: ' ~ outputs.evaluate.values.noncompliant_count ~ ' of ' ~ (trigger.files | length) ~ ' new file(s) did not match ' ~ inputs.allowed_extensions_regex ~ ': ' ~ outputs.evaluate.values.noncompliant_names ~ '. Set confirm_quarantine to true to let this flow quarantine them automatically, or review them manually now.'} | toJson }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: "NFS new file extension compliance gate complete: {{
      outputs.evaluate.values.noncompliant_count }} of {{ trigger.files | length
      }} new file(s) non-compliant, severity {{ outputs.evaluate.values.severity
      }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'NFS new file extension compliance gate failed (execution ' ~ execution.id ~ '). Check the logs.'} | toJson }}

triggers:
  - id: watch_directory
    type: io.kestra.plugin.fs.nfs.Trigger
    description: Polls the watched directory and fires once per polling cycle that
      finds new or updated files, deduplicated via trigger state. Shipped
      disabled with a placeholder path; point from at a real NFS-mounted
      directory before enabling.
    disabled: true
    from: "/mnt/nfs/shared/incoming"
    interval: PT30S
    on: CREATE
    recursive: false
    maxFiles: 25