Oracle Recycle Bin Space Reclaim Guard

id: oracle-recyclebin-space-reclaim-guard
namespace: company.team

description: |
  Find dropped Oracle tables that are still holding tablespace space in the recycle bin,
  purge the ones older than an age threshold with PURGE TABLE behind a dry run and an
  approval pause, then read dba_recyclebin again and fail if any target survived.

inputs:
  - id: jdbc_url
    type: STRING
    displayName: Oracle JDBC URL
    defaults: "jdbc:oracle:thin:@//localhost:1521/FREEPDB1"

  - id: username
    type: STRING
    displayName: Database user
    description: A user that can read dba_recyclebin (SELECT_CATALOG_ROLE or SELECT
      ANY DICTIONARY) and purge the objects it finds. Purging another schema's
      dropped table needs DROP ANY TABLE. A DBA role covers both.
    defaults: SYSTEM

  - id: password
    type: STRING
    displayName: Database password
    defaults: "{{ secret('ORACLE_PASSWORD') }}"

  - id: object_owners
    type: ARRAY
    itemType: STRING
    displayName: Schema owners to check
    description: Application schema owners whose dropped tables are checked. System
      schemas are left out so a SYS or internal recycle bin entry is never
      touched.
    defaults:
      - APP

  - id: min_age_hours
    type: FLOAT
    displayName: Minimum age in hours before a dropped table may be purged
    description: A table dropped more recently than this is reported but never
      purged, so FLASHBACK TABLE ... TO BEFORE DROP still works for it. Purging
      forfeits that recovery permanently. The default holds every drop for seven
      days.
    defaults: 168.0

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Allow PURGE TABLE to run. Off by default.
    defaults: false

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

variables:
  audit_sql: |
    SELECT t.owner AS owner,
           t.object_name AS object_name,
           t.original_name AS original_name,
           t.ts_name AS ts_name,
           t.droptime AS droptime,
           ROUND((SYSDATE - TO_DATE(t.droptime, 'YYYY-MM-DD:HH24:MI:SS')) * 24, 4) AS age_hours,
           (SELECT NVL(SUM(d.space), 0) FROM dba_recyclebin d WHERE d.base_object = t.base_object)
             * ts.block_size AS space_bytes,
           'PURGE TABLE "' || t.owner || '"."' || t.object_name || '"' AS fix_sql
      FROM dba_recyclebin t
      JOIN dba_tablespaces ts ON ts.tablespace_name = t.ts_name
     WHERE t.type = 'TABLE'
       AND t.can_purge = 'YES'
       AND t.owner IN ('{{ inputs.object_owners | join("','") | upper }}')
       AND (SYSDATE - TO_DATE(t.droptime, 'YYYY-MM-DD:HH24:MI:SS')) * 24 >= {{ inputs.min_age_hours }}
     ORDER BY t.droptime, t.owner, t.object_name

  totals_sql: |
    SELECT COUNT(*) AS object_count,
           NVL(SUM(CASE WHEN r.type = 'TABLE' THEN 1 ELSE 0 END), 0) AS dropped_tables,
           NVL(SUM(r.space * ts.block_size), 0) AS space_bytes,
           NVL(MIN(r.droptime), 'none') AS oldest_droptime
      FROM dba_recyclebin r
      JOIN dba_tablespaces ts ON ts.tablespace_name = r.ts_name
     WHERE r.owner IN ('{{ inputs.object_owners | join("','") | upper }}')

tasks:
  - id: totals_before
    type: io.kestra.plugin.jdbc.oracle.Query
    description: Read-only. How much tablespace space the whole recycle bin holds
      for the checked schemas, at any age.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ inputs.username }}"
    password: "{{ inputs.password }}"
    fetchType: FETCH_ONE
    sql: "{{ render(vars.totals_sql) }}"

  - id: audit
    type: io.kestra.plugin.jdbc.oracle.Query
    description: Read-only. Lists the dropped tables old enough to purge, with the
      space their segments plus dependent indexes still hold and the exact PURGE
      TABLE for each.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ inputs.username }}"
    password: "{{ inputs.password }}"
    fetchType: FETCH
    sql: "{{ render(vars.audit_sql) }}"

  - id: targets
    type: io.kestra.plugin.core.output.OutputValues
    description: The PURGE TABLE statements for the aged dropped tables plus the
      space they are expected to release.
    values:
      sql: "{{ outputs.audit.rows | jq('map(.FIX_SQL)') | first | toJson }}"
      reclaimable_bytes: "{{ outputs.audit.rows | jq('[.[].SPACE_BYTES] | add // 0') | first }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Oracle recycle bin on {{ inputs.jdbc_url }} for schemas {{
      inputs.object_owners | jq('map(ascii_upcase) | join(", ")') | first }}: {{
      outputs.totals_before.row.OBJECT_COUNT }} entry(s) holding {{
      outputs.totals_before.row.SPACE_BYTES }} byte(s), oldest drop {{
      outputs.totals_before.row.OLDEST_DROPTIME }}. {{ outputs.audit.size }}
      dropped table(s) are at least {{ inputs.min_age_hours }} hour(s) old and
      would release {{ outputs.targets.values.reclaimable_bytes }} byte(s). {{
      outputs.audit.rows | jq('map(.OWNER + "." + .ORIGINAL_NAME + " as " +
      .OBJECT_NAME + " in " + .TS_NAME + " dropped " + .DROPTIME + " holding " +
      (.SPACE_BYTES | tostring) + " bytes") | join("; ")') | first }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.audit.size > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Oracle ' ~ inputs.jdbc_url ~ ': ' ~ outputs.audit.size ~ ' dropped table(s) older than ' ~ inputs.min_age_hours ~ 'h still hold ' ~ outputs.targets.values.reclaimable_bytes ~ ' bytes in the recycle bin: ' ~ (outputs.audit.rows | jq('map(.OWNER + "." + .ORIGINAL_NAME + " (" + .TS_NAME + ", dropped " + .DROPTIME + ")") | join(", ")') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Purge is armed and waiting for approval. Flashback recovery for these tables ends at purge.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: approval_gate
            type: io.kestra.plugin.core.flow.Pause
            description: Hold the run until a human approves the purge. PURGE TABLE destroys
              the ability to flashback these tables, so the destructive step
              never starts on its own.
            pauseDuration: PT2H
            behavior: FAIL
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: Set true to purge the listed dropped tables and give up flashback
                  recovery for them. Anything else leaves them in the recycle
                  bin.

          - id: apply
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval_gate.onResume.approved ?? false }}"
            then:
              - id: remediate
                type: io.kestra.plugin.core.flow.Loop
                description: One PURGE TABLE per aged object. Sequential on purpose, so two
                  purges do not contend on the same tablespace extent map.
                concurrencyLimit: 1
                values: "{{ outputs.targets.values.sql }}"
                tasks:
                  - id: purge_one
                    type: io.kestra.plugin.jdbc.oracle.Query
                    description: PURGE TABLE for one dropped table, which also purges its dependent
                      indexes. allowFailure, so one object a privilege or lock
                      problem blocks does not stop the pass.
                    allowFailure: true
                    url: "{{ inputs.jdbc_url }}"
                    username: "{{ inputs.username }}"
                    password: "{{ inputs.password }}"
                    fetchType: NONE
                    sql: "{{ item.value }}"

              - id: reaudit
                type: io.kestra.plugin.jdbc.oracle.Query
                description: Read dba_recyclebin again with the same age window.
                url: "{{ inputs.jdbc_url }}"
                username: "{{ inputs.username }}"
                password: "{{ inputs.password }}"
                fetchType: FETCH
                sql: "{{ render(vars.audit_sql) }}"

              - id: totals_after
                type: io.kestra.plugin.jdbc.oracle.Query
                description: Re-read the recycle bin totals so the run reports the space
                  actually released.
                url: "{{ inputs.jdbc_url }}"
                username: "{{ inputs.username }}"
                password: "{{ inputs.password }}"
                fetchType: FETCH_ONE
                sql: "{{ render(vars.totals_sql) }}"

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.reaudit.size > 0 }}"
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      {{ outputs.reaudit.size }} dropped table(s) are still in
                      the recycle bin after the purge pass: {{
                      outputs.reaudit.rows | jq('map(.OWNER + "." +
                      .ORIGINAL_NAME + " as " + .OBJECT_NAME) | join(", ")') |
                      first }}. The recycle bin still holds {{
                      outputs.totals_after.row.SPACE_BYTES }} byte(s) across {{
                      outputs.totals_after.row.OBJECT_COUNT }} entry(s), down
                      from {{ outputs.totals_before.row.SPACE_BYTES }}. PURGE
                      TABLE is refused when the caller lacks DROP ANY TABLE for
                      that schema, so check the privileges of the user in the
                      username input.
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Purged {{ fromJson(outputs.targets.values.sql) | length }}
                      dropped table(s). Recycle bin went from {{
                      outputs.totals_before.row.OBJECT_COUNT }} entry(s) and {{
                      outputs.totals_before.row.SPACE_BYTES }} byte(s) to {{
                      outputs.totals_after.row.OBJECT_COUNT }} entry(s) and {{
                      outputs.totals_after.row.SPACE_BYTES }} byte(s). Nothing
                      younger than {{ inputs.min_age_hours }} hour(s) was
                      touched.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: "Purge not approved. {{ outputs.audit.size }} dropped table(s) stay in
                  the recycle bin and keep their flashback recovery."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: >-
          No dropped table in the checked schemas is older than {{
          inputs.min_age_hours }} hour(s). The recycle bin holds {{
          outputs.totals_before.row.OBJECT_COUNT }} entry(s) and {{
          outputs.totals_before.row.SPACE_BYTES }} byte(s), all of it still
          inside the flashback window.

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Oracle recycle bin space reclaim guard failed on ' ~ inputs.jdbc_url ~ ' (execution ' ~ execution.id ~ '). Dropped tables may still be holding tablespace space. Check the logs.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode (report only).
    cron: "0 5 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true