Apache CouchDB Database Compaction and Conflict Resolution Gate

id: couchdb-compaction-and-conflict-resolution-gate
namespace: company.database

description: |
  Audit Apache CouchDB database fragmentation and multi-master revision conflicts via REST APIs,
  track disk bloat and conflict trends against historical KV snapshots, pause for operator approval,
  prune conflicting document revisions, trigger database compaction and view cleanup, and verify
  conflict resolution post-remediation.

inputs:
  - id: couchdb_url
    type: STRING
    displayName: CouchDB Base URL
    description: HTTP endpoint of the Apache CouchDB server or cluster load balancer.
    defaults: "http://localhost:5984"

  - id: database_name
    type: STRING
    displayName: Target Database Name
    description: CouchDB database name to audit for disk bloat and document revision
      conflicts.
    defaults: "ecommerce_orders"

  - id: couchdb_username
    type: STRING
    displayName: CouchDB Admin Username
    description: Administrator or maintenance user account with compaction and
      delete permissions.
    defaults: "admin"

  - id: max_fragmentation_percent
    type: INT
    displayName: Max Fragmentation Percentage Threshold
    description: Wasted disk space percentage ((file_size - active_size) / file_size
      * 100) triggering compaction.
    defaults: 30

  - id: min_wasted_mb_for_compact
    type: INT
    displayName: Minimum Wasted Disk Space (MB)
    description: Minimum uncompacted dead storage in megabytes required before
      triggering compaction.
    defaults: 20

  - id: max_scan_limit
    type: INT
    displayName: Maximum Conflict Scan Limit
    description: Maximum document rows to inspect for conflicting revisions in a
      single execution.
    defaults: 100

  - id: dry_run
    type: BOOL
    displayName: Dry Run Mode
    description: When true, analyzes fragmentation and plans conflict resolutions
      without modifying revisions or invoking compaction.
    defaults: true

triggers:
  - id: periodic_maintenance_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run automated CouchDB fragmentation audit and conflict triage every
      6 hours.
    cron: "0 */6 * * *"
    disabled: true

  - id: on_demand_maintenance_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP webhook endpoint triggered by storage alarms,
      Prometheus Alertmanager, or DBA runbooks.
    key: "{{ secret('COUCHDB_MAINTENANCE_WEBHOOK_KEY') }}"

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.execution.Assert
    description: Assert configuration parameters satisfy operational bounds.
    conditions:
      - "{{ inputs.couchdb_url | length > 0 }}"
      - "{{ inputs.database_name | length > 0 }}"
      - "{{ inputs.couchdb_username | length > 0 }}"
      - "{{ inputs.max_fragmentation_percent >= 5 and
        inputs.max_fragmentation_percent <= 95 }}"
      - "{{ inputs.min_wasted_mb_for_compact >= 1 }}"
      - "{{ inputs.max_scan_limit >= 1 }}"
    errorMessage: "Invalid inputs. Provide valid endpoints, database names, and
      positive thresholds."

  - id: get_database_stats
    type: io.kestra.plugin.core.http.Request
    description: Query CouchDB database statistics endpoint to obtain disk file size
      and active data size.
    uri: "{{ inputs.couchdb_url }}/{{ inputs.database_name }}"
    method: GET
    headers:
      Accept: application/json
    options:
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT15S
      auth:
        type: BASIC
        username: "{{ inputs.couchdb_username }}"
        password: "{{ secret('COUCHDB_PASSWORD') }}"

  - id: query_conflicted_docs
    type: io.kestra.plugin.core.http.Request
    description: Scan target database documents with conflict flags to identify
      unresolved multi-master revision forks.
    uri: "{{ inputs.couchdb_url }}/{{ inputs.database_name
      }}/_all_docs?include_docs=true&conflicts=true&limit={{
      inputs.max_scan_limit }}"
    method: GET
    headers:
      Accept: application/json
    options:
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT30S
      auth:
        type: BASIC
        username: "{{ inputs.couchdb_username }}"
        password: "{{ secret('COUCHDB_PASSWORD') }}"

  - id: inspect_and_classify_maintenance
    type: io.kestra.plugin.scripts.python.Script
    description: Compute database fragmentation metrics, classify document conflicts
      against historical KV state, and formulate remediation plan.
    containerImage: python:3.11-slim
    env:
      DATABASE_NAME: "{{ inputs.database_name }}"
      MAX_FRAGMENTATION_PERCENT: "{{ inputs.max_fragmentation_percent }}"
      MIN_WASTED_MB_FOR_COMPACT: "{{ inputs.min_wasted_mb_for_compact }}"
      DRY_RUN: "{{ inputs.dry_run }}"
      DB_STATS_JSON: "{{ outputs.get_database_stats.body }}"
      ALL_DOCS_JSON: "{{ outputs.query_conflicted_docs.body }}"
      PREVIOUS_STATE_JSON: "{{ kv('couchdb_maintenance_tracker_' ~
        (inputs.database_name | slugify), errorOnMissing=false) ?? '{}' }}"
      EXECUTION_ID: "{{ execution.id }}"
    script: |
      import os
      import sys
      import json
      from datetime import datetime, timezone

      now_utc = datetime.now(timezone.utc)
      now_iso = now_utc.isoformat()

      db_name = os.environ.get("DATABASE_NAME", "ecommerce_orders").strip()
      max_frag_thresh = float(os.environ.get("MAX_FRAGMENTATION_PERCENT", "30"))
      min_wasted_mb_thresh = float(os.environ.get("MIN_WASTED_MB_FOR_COMPACT", "20"))
      dry_run = os.environ.get("DRY_RUN", "true").lower() == "true"
      execution_id = os.environ.get("EXECUTION_ID", "local-test")

      # Parse database stats
      db_stats_raw = os.environ.get("DB_STATS_JSON", "{}")
      try:
          db_stats = json.loads(db_stats_raw) if db_stats_raw.strip() else {}
      except Exception:
          db_stats = {}

      sizes = db_stats.get("sizes", {})
      file_bytes = int(sizes.get("file", 0))
      active_bytes = int(sizes.get("active", 0))
      doc_count = int(db_stats.get("doc_count", 0))
      doc_del_count = int(db_stats.get("doc_del_count", 0))
      compact_running = bool(db_stats.get("compact_running", False))

      wasted_bytes = max(0, file_bytes - active_bytes)
      wasted_mb = round(wasted_bytes / (1024 * 1024), 2)
      file_mb = round(file_bytes / (1024 * 1024), 2)
      active_mb = round(active_bytes / (1024 * 1024), 2)
      frag_percent = round((wasted_bytes / file_bytes) * 100, 2) if file_bytes > 0 else 0.0

      # Determine if database compaction is required
      requires_compaction = (frag_percent >= max_frag_thresh) and (wasted_mb >= min_wasted_mb_thresh) and not compact_running

      # Parse documents to identify revision conflicts
      all_docs_raw = os.environ.get("ALL_DOCS_JSON", "{}")
      try:
          all_docs = json.loads(all_docs_raw) if all_docs_raw.strip() else {}
      except Exception:
          all_docs = {}

      conflict_docs = []
      conflict_deletions = []

      for row in all_docs.get("rows", []):
          doc = row.get("doc", {})
          conflicts = doc.get("_conflicts", [])
          if conflicts:
              doc_id = doc.get("_id")
              winning_rev = doc.get("_rev")
              conflict_docs.append({
                  "doc_id": doc_id,
                  "winning_rev": winning_rev,
                  "conflict_count": len(conflicts),
                  "conflict_revs": conflicts
              })
              for c_rev in conflicts:
                  conflict_deletions.append({
                      "doc_id": doc_id,
                      "rev": c_rev,
                      "path": f"{doc_id}?rev={c_rev}"
                  })

      has_conflicts = len(conflict_docs) > 0

      # Parse historical KV tracker
      prev_state_raw = os.environ.get("PREVIOUS_STATE_JSON", "{}")
      try:
          prev_state = json.loads(prev_state_raw) if prev_state_raw.strip() else {}
      except Exception:
          prev_state = {}

      prev_frag = float(prev_state.get("fragmentation_percent", 0.0))
      prev_wasted = float(prev_state.get("wasted_mb", 0.0))
      frag_delta = round(frag_percent - prev_frag, 2)

      status = "CRITICAL" if (has_conflicts or (frag_percent >= 50.0 and wasted_mb >= 50.0)) else ("WARNING" if (requires_compaction or frag_percent >= max_frag_thresh) else "HEALTHY")

      summary = (
          f"CouchDB Maintenance: {status} | Database '{db_name}' - File: {file_mb} MB, Active: {active_mb} MB, "
          f"Wasted: {wasted_mb} MB (Fragmentation: {frag_percent}%, delta: {frag_delta:+0.2f}%). "
          f"Compaction needed: {requires_compaction} | Conflicted docs: {len(conflict_docs)} (Losing revs: {len(conflict_deletions)}) | "
          f"Dry run: {dry_run}."
      )
      print(summary)

      new_state_snapshot = {
          "database_name": db_name,
          "file_mb": file_mb,
          "active_mb": active_mb,
          "wasted_mb": wasted_mb,
          "fragmentation_percent": frag_percent,
          "doc_count": doc_count,
          "doc_del_count": doc_del_count,
          "conflicted_docs_count": len(conflict_docs),
          "last_evaluated_at": now_iso,
          "execution_id": execution_id
      }

      print("::" + json.dumps({
          "outputs": {
              "status": status,
              "file_mb": file_mb,
              "active_mb": active_mb,
              "wasted_mb": wasted_mb,
              "fragmentation_percent": frag_percent,
              "frag_delta": frag_delta,
              "doc_count": doc_count,
              "doc_del_count": doc_del_count,
              "requires_compaction": requires_compaction,
              "has_conflicts": has_conflicts,
              "conflict_docs_count": len(conflict_docs),
              "conflict_docs": conflict_docs,
              "conflict_deletions": conflict_deletions,
              "new_state_snapshot": new_state_snapshot,
              "summary": summary
          }
      }) + "::")

  - id: save_maintenance_tracker_kv
    type: io.kestra.plugin.core.kv.Set
    description: Persist database storage metrics and conflict telemetry in Kestra
      KV for historical drift tracking.
    key: "couchdb_maintenance_tracker_{{ inputs.database_name | slugify }}"
    value: "{{ outputs.inspect_and_classify_maintenance.vars.new_state_snapshot |
      toJson }}"
    kvType: JSON

  - id: evaluate_maintenance_findings
    type: io.kestra.plugin.core.flow.If
    description: Branch execution when database compaction is required or document
      conflicts exist.
    condition: "{{ outputs.inspect_and_classify_maintenance.vars.requires_compaction
      or outputs.inspect_and_classify_maintenance.vars.has_conflicts }}"
    then:
      - id: notify_dba_triage
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatch incident notification with database fragmentation and
          conflict metrics.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#db-ops",
            "attachments": [
              {
                "color": "{{ outputs.inspect_and_classify_maintenance.vars.status == 'CRITICAL' ? 'danger' : 'warning' }}",
                "title": ":warning: CouchDB Compaction and Conflict Triage Triggered",
                "text": "*Database:* `{{ inputs.database_name }}`\n*Fragmentation:* `{{ outputs.inspect_and_classify_maintenance.vars.fragmentation_percent }}%` (Delta: `{{ outputs.inspect_and_classify_maintenance.vars.frag_delta }}%`)\n*Wasted Storage:* `{{ outputs.inspect_and_classify_maintenance.vars.wasted_mb }} MB` (File: `{{ outputs.inspect_and_classify_maintenance.vars.file_mb }} MB`, Active: `{{ outputs.inspect_and_classify_maintenance.vars.active_mb }} MB`)\n*Conflicted Documents:* `{{ outputs.inspect_and_classify_maintenance.vars.conflict_docs_count }}`\n*Compaction Required:* `{{ outputs.inspect_and_classify_maintenance.vars.requires_compaction }}`\n*Dry Run:* `{{ inputs.dry_run }}`",
                "footer": "Kestra CouchDB Maintenance Gate • Execution {{ execution.id }}"
              }
            ]
          }

      - id: execute_remediation_gate
        type: io.kestra.plugin.core.flow.If
        description: Execute conflict pruning and database compaction when dry_run is
          disabled.
        condition: "{{ not inputs.dry_run }}"
        then:
          - id: pause_for_remediation_approval
            type: io.kestra.plugin.core.flow.Pause
            description: Pause workflow execution for DBA review before deleting conflicting
              revisions and compacting the database. Resume to apply or kill to
              abort.

          - id: resolve_conflicted_revisions
            type: io.kestra.plugin.core.flow.Loop
            description: Prune losing conflicting revisions via CouchDB HTTP DELETE
              endpoint.
            values: "{{ outputs.inspect_and_classify_maintenance.vars.conflict_deletions }}"
            tasks:
              - id: delete_conflicting_revision
                type: io.kestra.plugin.core.http.Request
                description: Delete losing conflicting revision to resolve multi-master conflict
                  fork.
                uri: "{{ inputs.couchdb_url }}/{{ inputs.database_name }}/{{
                  fromJson(item.value).path }}"
                method: DELETE
                headers:
                  Accept: application/json
                options:
                  timeout:
                    connectTimeout: PT10S
                    readIdleTimeout: PT15S
                  auth:
                    type: BASIC
                    username: "{{ inputs.couchdb_username }}"
                    password: "{{ secret('COUCHDB_PASSWORD') }}"

          - id: trigger_compaction_gate
            type: io.kestra.plugin.core.flow.If
            description: Trigger CouchDB database and view compaction when fragmentation
              threshold is crossed.
            condition: "{{ outputs.inspect_and_classify_maintenance.vars.requires_compaction
              }}"
            then:
              - id: request_database_compaction
                type: io.kestra.plugin.core.http.Request
                description: Initiate CouchDB database compaction to reclaim uncompacted B-tree
                  space.
                uri: "{{ inputs.couchdb_url }}/{{ inputs.database_name }}/_compact"
                method: POST
                headers:
                  Accept: application/json
                  Content-Type: application/json
                contentType: application/json
                body: "{}"
                options:
                  timeout:
                    connectTimeout: PT10S
                    readIdleTimeout: PT15S
                  auth:
                    type: BASIC
                    username: "{{ inputs.couchdb_username }}"
                    password: "{{ secret('COUCHDB_PASSWORD') }}"

              - id: request_view_cleanup
                type: io.kestra.plugin.core.http.Request
                description: Remove stale secondary index view files to reclaim view disk space.
                uri: "{{ inputs.couchdb_url }}/{{ inputs.database_name }}/_view_cleanup"
                method: POST
                headers:
                  Accept: application/json
                  Content-Type: application/json
                contentType: application/json
                body: "{}"
                options:
                  timeout:
                    connectTimeout: PT10S
                    readIdleTimeout: PT15S
                  auth:
                    type: BASIC
                    username: "{{ inputs.couchdb_username }}"
                    password: "{{ secret('COUCHDB_PASSWORD') }}"

          - id: verify_post_maintenance_conflicts
            type: io.kestra.plugin.core.http.Request
            description: Post-remediation verification re-querying documents to assert
              conflicting revisions have been eliminated.
            uri: "{{ inputs.couchdb_url }}/{{ inputs.database_name
              }}/_all_docs?include_docs=true&conflicts=true&limit={{
              inputs.max_scan_limit }}"
            method: GET
            headers:
              Accept: application/json
            options:
              timeout:
                connectTimeout: PT10S
                readIdleTimeout: PT30S
              auth:
                type: BASIC
                username: "{{ inputs.couchdb_username }}"
                password: "{{ secret('COUCHDB_PASSWORD') }}"

          - id: count_remaining_conflicts
            type: io.kestra.plugin.scripts.python.Script
            description: Parse the re-queried document rows to count any documents still
              carrying unresolved conflicting revisions.
            containerImage: python:3.11-slim
            env:
              VERIFY_RESPONSE_JSON: "{{ outputs.verify_post_maintenance_conflicts.body }}"
            script: |
              import os
              import json

              try:
                  from kestra import Kestra
              except ImportError:
                  Kestra = None

              raw = os.environ.get("VERIFY_RESPONSE_JSON", "{}")
              try:
                  data = json.loads(raw) if raw.strip() else {}
              except Exception:
                  data = {}

              remaining = 0
              for row in data.get("rows", []):
                  doc = row.get("doc", {}) or {}
                  if doc.get("_conflicts"):
                      remaining += 1

              print(f"Remaining conflicted documents after remediation: {remaining}")

              output_data = {"remaining_conflicted_docs": remaining}
              if Kestra:
                  Kestra.outputs(output_data)
              else:
                  print(f"::{json.dumps({'outputs': output_data})}::")

          - id: assert_conflicts_resolved
            type: io.kestra.plugin.core.execution.Assert
            description: Assert no conflicted documents remain among target scanned rows.
            conditions:
              - "{{
                outputs.count_remaining_conflicts.vars.remaining_conflicted_docs
                == 0 }}"
            errorMessage: "Post-remediation verification failed: {{
              outputs.count_remaining_conflicts.vars.remaining_conflicted_docs
              }} document(s) still show unresolved conflicting revisions."

          - id: log_remediation_success
            type: io.kestra.plugin.core.log.Log
            description: Log verified maintenance and compaction results.
            message: "Successfully resolved {{
              outputs.inspect_and_classify_maintenance.vars.conflict_deletions |
              length }} conflict revision(s) in database '{{
              inputs.database_name }}'. Compaction initiated: {{
              outputs.inspect_and_classify_maintenance.vars.requires_compaction
              }}."

        else:
          - id: log_dry_run_summary
            type: io.kestra.plugin.core.log.Log
            description: Log planned maintenance actions in dry-run mode.
            message: "Dry run mode active. No changes made to database '{{
              inputs.database_name }}'. Planned: {{
              outputs.inspect_and_classify_maintenance.vars.conflict_deletions |
              length }} conflict revisions to prune, Compaction required: {{
              outputs.inspect_and_classify_maintenance.vars.requires_compaction
              }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Dispatch failure notification if CouchDB maintenance workflow fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "#db-ops",
        "attachments": [
          {
            "color": "danger",
            "title": ":rotating_light: CouchDB Maintenance Workflow Failed",
            "text": "Execution `{{ execution.id }}` failed while inspecting database `{{ inputs.database_name }}` on `{{ inputs.couchdb_url }}`. Check CouchDB credentials and endpoint connectivity."
          }
        ]
      }