Report Unused Postgres Indexes

id: postgres-unused-index-report
namespace: company.team
description: |
  Find Postgres indexes with zero scans since last stats reset and post the
  list to Slack, so dead indexes stop taxing every write.

triggers:
  - id: weekly_index_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekly review so index hygiene is a standing item, not a yearly surprise.
    cron: "0 5 * * 5"
    disabled: true

inputs:
  - id: jdbc_url
    type: STRING
    defaults: "jdbc:postgresql://localhost:5432/app"
    description: JDBC URL for the target database.

tasks:
  - id: find_unused
    type: io.kestra.plugin.scripts.shell.Commands
    description: Query pg_stat_user_indexes for idx_scan = 0 indexes, exclude
      primary keys and unique constraints, and emit the count plus names via the
      stdout outputs protocol. A failed query reports worst case.
    containerImage: python:3.12-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    commands:
      - |
        pip install --quiet psycopg2-binary
        cat > unused.py <<'PYEOF'
        import os, json, re, subprocess
        host = os.environ.get("JDBC_URL", "").replace("jdbc:postgresql://", "")
        try:
            import psycopg2
            db = re.sub(r"^[^/]+:", "", host).lstrip("/").split("?")[0]
            conn = psycopg2.connect(host=host.split("/")[0].split(":")[0], dbname=db, user=os.environ.get("PGUSER", "postgres"), password=os.environ.get("PGPASSWORD", ""), port=os.environ.get("PGPORT", "5432"), connect_timeout=10)
            cur = conn.cursor()
            cur.execute("""SELECT indexrelname FROM pg_stat_user_indexes ui JOIN pg_index i ON i.indexrelid = ui.indexrelid WHERE ui.idx_scan = 0 AND NOT i.indisprimary AND NOT i.indisunique AND ui.schemaname NOT IN ('pg_catalog','information_schema')""")
            names = [r[0] for r in cur.fetchall()]
        except Exception:
            names = ["__query_failed__"]
        print(f"{len(names)} unused index(es)")
        print("::" + json.dumps({"outputs": {"unused_count": len(names), "indexes": ",".join(names)}}) + "::")
        PYEOF
        python3 unused.py
    env:
      JDBC_URL: "{{ inputs.jdbc_url }}"
      PGPASSWORD: "{{ secret('PG_PASSWORD') }}"

  - id: unused_found
    type: io.kestra.plugin.core.flow.If
    description: Only alert when something is unused; a tight schema just logs.
    condition: "{{ outputs.find_unused.vars.unused_count > 0 }}"
    then:
      - id: alert_unused
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Name the candidates so the DROP decision starts there.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":mag: Postgres index audit: {{ outputs.find_unused.vars.unused_count }} index(es) with zero scans since stats reset: {{ outputs.find_unused.vars.indexes }}. Consider dropping them before the next write-heavy season. Execution {{ execution.id }}."
          }
    else:
      - id: log_clean
        type: io.kestra.plugin.core.log.Log
        description: Record the clean audit.
        message: "No unused secondary indexes found."

errors:
  - id: alert_audit_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the audit fails - a dead connection must not read as
      "schema is tight".
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Postgres unused-index audit FAILED in flow {{ flow.id }} (execution {{ execution.id }}). Check the JDBC URL and credentials."
      }

outputs:
  - id: unused_indexes
    type: STRING
    description: 'Comma-joined unused index names, e.g. "idx_old_email, idx_legacy_flag".'
    value: "{{ outputs.find_unused.vars.indexes }}"