Archive old Postgres rows to Parquet, verify the archive, then purge after approval

id: postgres-verified-cold-archive
namespace: company.team

description: |
  Archive old rows of a hot Postgres table to Parquet in MinIO/S3, prove the archive by reading it back,
  then delete the rows only after a human types the table name to confirm.
  Nothing is deleted unless: archived rows == exported rows == rows still in Postgres at purge time.

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: table
    type: STRING
    defaults: events
    validator: ^[a-z_][a-z0-9_]{0,62}$
    description: Unquoted lowercase table name (no schema prefix). Validated because
      it is interpolated into SQL.

  - id: date_column
    type: STRING
    defaults: created_at
    validator: ^[a-z_][a-z0-9_]{0,62}$
    description: Timestamp or date column used for the retention cutoff.

  - id: retention_days
    type: INT
    defaults: 365
    min: 30
    description: Rows older than this many days are archived, then purged.

  - id: bucket
    type: STRING
    defaults: cold-archive

  - id: notify_slack
    type: BOOL
    defaults: false
    description: Post the approval request to Slack (needs the SLACK_WEBHOOK_URL secret).

variables:
  cutoff: "{{ execution.startDate | dateAdd(inputs.retention_days * -1, 'DAYS') |
    date('yyyy-MM-dd') }}"
  object_key: "archive/{{ inputs.table }}/cutoff={{ execution.startDate |
    dateAdd(inputs.retention_days * -1, 'DAYS') | date('yyyy-MM-dd') }}/{{
    execution.id }}.parquet"

tasks:
  - id: count_candidates
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: "SELECT count(*) AS n FROM {{ inputs.table }} WHERE {{ inputs.date_column
      }} < CAST(:cutoff AS date)"
    parameters:
      cutoff: "{{ render(vars.cutoff) }}"

  - id: has_rows
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.count_candidates.row.n > 0 }}"
    then:
      - id: export_rows
        type: io.kestra.plugin.jdbc.postgresql.Query
        url: "{{ secret('POSTGRES_URL') }}"
        username: "{{ secret('POSTGRES_USERNAME') }}"
        password: "{{ secret('POSTGRES_PASSWORD') }}"
        fetchType: STORE
        sql: "SELECT * FROM {{ inputs.table }} WHERE {{ inputs.date_column }} <
          CAST(:cutoff AS date)"
        parameters:
          cutoff: "{{ render(vars.cutoff) }}"

      - id: to_parquet
        type: io.kestra.plugin.serdes.parquet.IonToParquet
        from: "{{ outputs.export_rows.uri }}"
        inferAllFields: true

      - id: upload_archive
        type: io.kestra.plugin.minio.Upload
        endpoint: "{{ secret('MINIO_ENDPOINT') }}"
        accessKeyId: "{{ secret('MINIO_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('MINIO_SECRET_KEY_ID') }}"
        bucket: "{{ inputs.bucket }}"
        from: "{{ outputs.to_parquet.uri }}"
        key: "{{ render(vars.object_key) }}"

      - id: download_back
        type: io.kestra.plugin.minio.Download
        endpoint: "{{ secret('MINIO_ENDPOINT') }}"
        accessKeyId: "{{ secret('MINIO_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('MINIO_SECRET_KEY_ID') }}"
        bucket: "{{ inputs.bucket }}"
        key: "{{ render(vars.object_key) }}"

      - id: count_archive
        type: io.kestra.plugin.jdbc.duckdb.Query
        inputFiles:
          archive.parquet: "{{ outputs.download_back.uri }}"
        fetchType: FETCH_ONE
        sql: "SELECT count(*) AS n FROM read_parquet('archive.parquet')"

      - id: verify_archive
        type: io.kestra.plugin.core.execution.Assert
        conditions:
          - "{{ outputs.count_archive.row.n == outputs.to_parquet.size }}"
          - "{{ outputs.to_parquet.size == outputs.export_rows.size }}"
        errorMessage: Archive read back from the bucket does not match the exported
          rows. Nothing was deleted.

      - id: ask_slack
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: notify_approval
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: "Archive verified: {{ outputs.count_archive.row.n }} rows of {{
              inputs.table }} older than {{ render(vars.cutoff) }} are in
              s3://{{ inputs.bucket }}/{{ render(vars.object_key) }}. Resume
              execution {{ execution.id }} and type the table name to purge
              them."

      - id: approve_purge
        type: io.kestra.plugin.core.flow.Pause
        pauseDuration: P2D
        behavior: CANCEL
        onResume:
          - id: confirm_table
            type: STRING
            description: Type the table name to confirm deletion of the archived rows.

      - id: confirm_name
        type: io.kestra.plugin.core.execution.Assert
        conditions:
          - "{{ outputs.approve_purge.onResume.confirm_table == inputs.table }}"
        errorMessage: Confirmation did not match the table name. Nothing was deleted.

      - id: recount
        type: io.kestra.plugin.jdbc.postgresql.Query
        url: "{{ secret('POSTGRES_URL') }}"
        username: "{{ secret('POSTGRES_USERNAME') }}"
        password: "{{ secret('POSTGRES_PASSWORD') }}"
        fetchType: FETCH_ONE
        sql: "SELECT count(*) AS n FROM {{ inputs.table }} WHERE {{ inputs.date_column
          }} < CAST(:cutoff AS date)"
        parameters:
          cutoff: "{{ render(vars.cutoff) }}"

      - id: still_matches
        type: io.kestra.plugin.core.execution.Assert
        conditions:
          - "{{ outputs.recount.row.n == outputs.count_archive.row.n }}"
        errorMessage: Rows older than the cutoff changed while waiting for approval.
          Nothing was deleted; re-run to archive again.

      - id: purge
        type: io.kestra.plugin.jdbc.postgresql.Query
        url: "{{ secret('POSTGRES_URL') }}"
        username: "{{ secret('POSTGRES_USERNAME') }}"
        password: "{{ secret('POSTGRES_PASSWORD') }}"
        fetchType: FETCH_ONE
        sql: "WITH d AS (DELETE FROM {{ inputs.table }} WHERE {{ inputs.date_column }} <
          CAST(:cutoff AS date) RETURNING 1) SELECT count(*) AS deleted FROM d"
        parameters:
          cutoff: "{{ render(vars.cutoff) }}"

      - id: verify_purge
        type: io.kestra.plugin.core.execution.Assert
        conditions:
          - "{{ outputs.purge.row.deleted == outputs.count_archive.row.n }}"
        errorMessage: Deleted row count differs from the archived row count. Restore
          from the archive object and investigate.

      - id: log_done
        type: io.kestra.plugin.core.log.Log
        message: "Archived and purged {{ outputs.purge.row.deleted }} rows from {{
          inputs.table }} (older than {{ render(vars.cutoff) }}). Archive:
          s3://{{ inputs.bucket }}/{{ render(vars.object_key) }}"
    else:
      - id: log_nothing
        type: io.kestra.plugin.core.log.Log
        message: "No rows in {{ inputs.table }} older than {{ render(vars.cutoff) }}.
          Nothing to archive."

errors:
  - id: log_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: "Cold archive of {{ inputs.table }} stopped before completion. Rows are
      only deleted by the purge task, after verification and approval; check its
      state in execution {{ execution.id }}."

triggers:
  - id: archive_request
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('ARCHIVE_WEBHOOK_KEY') }}"
    inputs:
      table: "{{ trigger.body.table ?? 'events' }}"
      date_column: "{{ trigger.body.date_column ?? 'created_at' }}"
      retention_days: "{{ trigger.body.retention_days ?? 365 }}"