Import unmanaged resources into Terraform state and reconcile

id: terraform-unmanaged-resource-import-reconcile
namespace: company.team

description: |
  Someone created the resource by hand and Terraform does not know it exists.
  The configuration still declares it, so terraform plan reports a create and
  the next apply dies on a name collision or builds a duplicate. This flow
  reads the provider inventory over its own API, matches every planned create
  against the live object, imports only the addresses inside an allowed prefix
  list, removes a state entry whose object is gone first, undoes an import that
  would make the next apply destroy the object, then re-reads plan and state and
  fails the run when the reconcile did not land.

labels:
  owner: platform-team
  system: terraform

inputs:
  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Report the import plan without touching the state. Safe mode, on by
      default.
    defaults: true

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Allow state writes. Still needs dry_run false plus an approval on
      the paused execution.
    defaults: false

  - id: docker_host
    type: STRING
    displayName: Provider endpoint
    description: >-
      The daemon this configuration manages. Point it at the remote daemon that
      owns the resources, never at the daemon running Kestra. The flow reads the
      same endpoint over the Engine API, independently of the provider.
    defaults: tcp://127.0.0.1:2375

  - id: allowed_address_prefixes
    type: ARRAY
    itemType: STRING
    displayName: Allowed address prefixes
    description: >-
      An import is only ever attempted for an address starting with one of these
      prefixes. Every other collision is reported and left alone.
    defaults: [ "docker_container.edge_", "docker_network.edge_" ]

  - id: protected_addresses
    type: ARRAY
    itemType: STRING
    displayName: Protected addresses
    description: >-
      Exact addresses this flow never imports, even when they sit inside an
      allowed prefix and collide with a live object.
    defaults: []

  - id: importable_types
    type: ARRAY
    itemType: STRING
    displayName: Importable resource types
    description: Resource types whose import id this flow knows how to read from the
      provider API.
    defaults: [ "docker_container", "docker_network", "docker_volume" ]

  - id: max_imports
    type: INT
    displayName: Maximum imports per run
    description: Ceiling on state writes in one run. Anything above it is reported
      as held back.
    defaults: 5

  - id: tolerated_attributes
    type: ARRAY
    itemType: STRING
    displayName: Attributes the provider cannot read back
    description: >-
      Attributes that exist only in the configuration, so a freshly imported
      resource always shows them as an in place update. Any other attribute
      difference after an import fails the run.
    defaults:
      [
        "must_run",
        "attach",
        "logs",
        "start",
        "rm",
        "wait",
        "wait_timeout",
        "remove_volumes",
        "container_read_refresh_timeout_milliseconds"
      ]

  - id: task_network
    type: STRING
    displayName: Docker network for the task containers
    description: >-
      Network the Terraform containers join so they can reach the state backend
      and the provider endpoint by name. Use bridge when both are reachable by
      address.
    defaults: bridge

  - id: state_pg_endpoint
    type: STRING
    displayName: State backend host:port
    description: Postgres holding the Terraform state. The pg backend ships with
      Terraform, no extra provider.
    defaults: postgres:5432

  - id: state_pg_database
    type: STRING
    displayName: State backend database
    description: Database inside that Postgres holding the state table.
    defaults: terraform_state

  - id: node_image
    type: STRING
    displayName: Image the managed containers run
    description: Passed to the configuration below as a Terraform variable.
    defaults: busybox:1.37

tasks:
  - id: detect
    type: io.kestra.plugin.terraform.cli.TerraformCLI
    description: >-
      Read the truth twice in one container. First the provider inventory
      straight off the Engine API, then terraform plan with a refresh. A planned
      create whose object is already live is an unmanaged resource, not new
      work.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: "{{ inputs.task_network }}"
    containerImage: hashicorp/terraform:1.10.5
    env:
      PG_CONN_STR: "postgres://{{ secret('POSTGRES_USERNAME') }}:{{
        secret('POSTGRES_PASSWORD') }}@{{ inputs.state_pg_endpoint }}/{{
        inputs.state_pg_database }}?sslmode=disable"
      TF_IN_AUTOMATION: "true"
      TF_VAR_docker_host: "{{ inputs.docker_host }}"
      TF_VAR_node_image: "{{ inputs.node_image }}"
      ALLOWED: "{{ inputs.allowed_address_prefixes | toJson }}"
      PROTECTED: "{{ inputs.protected_addresses | toJson }}"
      TYPES: "{{ inputs.importable_types | toJson }}"
      MAX_IMPORTS: "{{ inputs.max_imports }}"
    inputFiles:
      main.tf: |
        # Replace this with your own configuration. Keep the backend remote so
        # every task in the flow reads and writes the same state.
        terraform {
          required_version = ">= 1.6"
          backend "pg" {}
          required_providers {
            docker = {
              source  = "kreuzwerker/docker"
              version = "~> 3.0"
            }
          }
        }

        variable "docker_host" { type = string }

        variable "node_image" {
          type    = string
          default = "busybox:1.37"
        }

        provider "docker" {
          host = var.docker_host
        }

        resource "docker_image" "node" {
          name         = var.node_image
          keep_locally = true
        }

        resource "docker_network" "edge_net" {
          name   = "edge-net"
          driver = "bridge"
        }

        resource "docker_container" "edge_api" {
          name     = "edge-api"
          image    = docker_image.node.name
          must_run = true
          restart  = "no"
          command  = ["sleep", "86400"]

          lifecycle {
            # An imported container reports the env the image sets. Because env
            # forces replacement, every import without this would look like a
            # resource the next apply has to rebuild.
            ignore_changes = [env]
          }
        }

        resource "docker_container" "edge_cache" {
          name     = "edge-cache"
          image    = docker_image.node.name
          must_run = true
          restart  = "no"
          command  = ["sleep", "86400"]

          lifecycle {
            # An imported container reports the env the image sets. Because env
            # forces replacement, every import without this would look like a
            # resource the next apply has to rebuild.
            ignore_changes = [env]
          }
        }

        resource "docker_container" "edge_worker" {
          name     = "edge-worker"
          image    = docker_image.node.name
          must_run = true
          restart  = "no"
          command  = ["sleep", "86400"]

          lifecycle {
            # An imported container reports the env the image sets. Because env
            # forces replacement, every import without this would look like a
            # resource the next apply has to rebuild.
            ignore_changes = [env]
          }
        }

        resource "docker_container" "edge_cron" {
          name     = "edge-cron"
          image    = docker_image.node.name
          must_run = true
          restart  = "no"
          command  = ["sleep", "86400"]

          lifecycle {
            # An imported container reports the env the image sets. Because env
            # forces replacement, every import without this would look like a
            # resource the next apply has to rebuild.
            ignore_changes = [env]
          }
        }

        resource "docker_container" "legacy_sidecar" {
          name     = "legacy-sidecar"
          image    = docker_image.node.name
          must_run = true
          restart  = "no"
          command  = ["sleep", "86400"]

          lifecycle {
            # An imported container reports the env the image sets. Because env
            # forces replacement, every import without this would look like a
            # resource the next apply has to rebuild.
            ignore_changes = [env]
          }
        }
      inventory.sh: |
        #!/bin/sh
        # Read the provider inventory from the Engine API, not from Terraform.
        # Writes live.json: [{type, name, live_id}]
        set -e
        API=$(echo "$TF_VAR_docker_host" | sed -e 's|^tcp://|http://|' -e 's|^unix://|http://localhost|')
        wget -qO- "$API/containers/json?all=true" \
          | jq -c '[.[] | {type: "docker_container", name: (.Names[0] | ltrimstr("/")), live_id: .Id}]' > live_containers.json
        wget -qO- "$API/networks" \
          | jq -c '[.[] | {type: "docker_network", name: .Name, live_id: .Id}]' > live_networks.json
        wget -qO- "$API/volumes" \
          | jq -c '[.Volumes[]? | {type: "docker_volume", name: .Name, live_id: .Name}]' > live_volumes.json
        jq -s -c add live_containers.json live_networks.json live_volumes.json > live.json
      reconcile.jq: |
        # Select import targets from a Terraform JSON plan.
        # stdin: terraform show -json <planfile>
        # $live: [{type,name,live_id}] read from the provider API
        # $st: terraform show -json, the persisted state
        # $allowed: address prefixes that may be touched
        # $protected: addresses never touched
        # $types: resource types whose import id this flow can resolve
        # $max: per run cap on state writes
        ($live[0] // []) as $live_objects
        | ($st[0].values.root_module.resources // []) as $state_resources
        | ([$state_resources[] | .address] | sort) as $state_addresses
        | [$live_objects[] | .live_id] as $live_ids
        | [.resource_changes[]? | select(.change.actions == ["create"])] as $creates
        | [.resource_changes[]? | select(.change.actions != ["no-op"]) | .address] as $pending
        | [$pending[] | select(. as $a | ([$allowed[] | select(. as $p | ($a | startswith($p)))] | length) > 0)] as $pending_in_scope
        | [$creates[]
           | . as $c
           | ($c.change.after.name // null) as $name
           | ($live_objects | map(select(.type == $c.type and .name == $name)) | .[0].live_id // null) as $live_id
           | ($state_resources | map(select(.address == $c.address)) | .[0].values.id // null) as $recorded_id
           | (($state_addresses | index($c.address)) != null) as $in_state
           | {address: $c.address,
              type: $c.type,
              name: $name,
              live_id: $live_id,
              recorded_id: $recorded_id,
              in_state: $in_state,
              verdict:
                (if ($protected | index($c.address)) != null then "protected"
                 elif ([$allowed[] | select(. as $p | ($c.address | startswith($p)))] | length) == 0 then "outside_allowed_prefix"
                 elif ($types | index($c.type)) == null then "type_not_importable"
                 elif $name == null then "no_identity_in_plan"
                 elif $live_id == null then "create_not_a_collision"
                 elif $in_state and ($recorded_id != null) and (($live_ids | index($recorded_id)) != null) then "recorded_object_still_live"
                 elif $in_state then "state_rm_then_import"
                 else "import" end)}]
          as $candidates
        | ([$candidates[] | select(.verdict == "import" or .verdict == "state_rm_then_import")] | sort_by(.address)) as $eligible
        | {counts: {candidates: ($candidates | length),
                    collisions: ($eligible | length),
                    selected: ($eligible[0:$max] | length),
                    held_back: ($eligible[$max:] | length),
                    spared: ([$candidates[] | select(.verdict != "import" and .verdict != "state_rm_then_import")] | length),
                    state_entries: ($state_addresses | length),
                    live_objects: ($live_objects | length),
                    pending_changes: ($pending | length),
                    pending_in_scope: ($pending_in_scope | length)},
           state_addresses: $state_addresses,
           pending_in_scope: ($pending_in_scope | sort),
           selected: $eligible[0:$max],
           held_back: $eligible[$max:],
           spared: [$candidates[] | select(.verdict != "import" and .verdict != "state_rm_then_import")],
           candidates: $candidates}
    outputFiles:
      - main.tf
      - inventory.sh
      - reconcile.jq
      - live.json
      - plan.txt
      - targets.json
      - state_backup.json
    beforeCommands:
      - apk add --no-cache jq > /dev/null
      - terraform init -input=false -no-color
    commands:
      - |
        set -e
        sh inventory.sh
        terraform state pull > state_backup.json
        set +e
        terraform plan -refresh=true -detailed-exitcode -no-color -input=false -lock-timeout=60s -out=tfplan > plan.txt 2>&1
        PLAN_EXIT=$?
        set -e
        if [ "$PLAN_EXIT" = "1" ]; then
          cat plan.txt
          echo "terraform plan failed, nothing was read"
          exit 1
        fi
        terraform show -json tfplan > plan.json
        terraform show -json > state.json
        jq -f reconcile.jq \
          --slurpfile live live.json --slurpfile st state.json \
          --argjson allowed "$ALLOWED" --argjson protected "$PROTECTED" \
          --argjson types "$TYPES" --argjson max "$MAX_IMPORTS" \
          plan.json > targets.json
        echo "counts $(jq -c .counts targets.json)"
        jq -r '.candidates[] | "candidate \(.address) name=\(.name) live=\(.live_id // "none") verdict=\(.verdict)"' targets.json
        printf '::{"outputs":{"plan_exit":%s,"collisions":%s,"selected":%s,"selected_count":%s,"held_back":%s,"spared":%s,"state_entries":%s,"live_objects":%s,"pending_in_scope":%s,"state_addresses":%s}}::\n' \
          "$PLAN_EXIT" \
          "$(jq -r .counts.collisions targets.json)" \
          "$(jq -c .selected targets.json)" \
          "$(jq -r .counts.selected targets.json)" \
          "$(jq -r .counts.held_back targets.json)" \
          "$(jq -r .counts.spared targets.json)" \
          "$(jq -r .counts.state_entries targets.json)" \
          "$(jq -r .counts.live_objects targets.json)" \
          "$(jq -r .counts.pending_in_scope targets.json)" \
          "$(jq -c .state_addresses targets.json)"

  - id: report
    type: io.kestra.plugin.core.log.Log
    description: Say what was found and whether this run is allowed to write anything.
    message: |
      terraform plan detailed exit code {{ outputs.detect.vars.plan_exit }} on {{ inputs.docker_host }}
      live objects {{ outputs.detect.vars.live_objects }}, state entries {{ outputs.detect.vars.state_entries }}
      pending changes inside the allowed prefixes {{ outputs.detect.vars.pending_in_scope }}
      unmanaged resources that collide with a live object {{ outputs.detect.vars.collisions }}
      selected for import {{ outputs.detect.vars.selected_count }}, held back by max_imports {{ outputs.detect.vars.held_back }}, spared {{ outputs.detect.vars.spared }}
      selected {{ outputs.detect.vars.selected | toJson }}
      dry_run {{ inputs.dry_run }}, auto_remediate {{ inputs.auto_remediate }}
      {{ (outputs.detect.vars.collisions > 0 and inputs.auto_remediate and inputs.dry_run == false) ? 'the run will pause for approval before any state write' : 'nothing will be written on this run' }}

  - id: reconcile_gate
    type: io.kestra.plugin.core.flow.If
    description: The write path. Needs a real collision, auto_remediate true and
      dry_run false.
    condition: "{{ outputs.detect.vars.collisions > 0 and inputs.auto_remediate and
      inputs.dry_run == false }}"
    then:
      - id: approve_import
        type: io.kestra.plugin.core.flow.Pause
        description: >-
          An import rewrites state for a live resource, so a human answers
          first. Resume with approved true to write, false to walk away.
        pauseDuration: PT1H
        onResume:
          - id: approved
            type: BOOL
            displayName: Import the selected addresses
            description: true writes state for the addresses listed above. Anything else
              leaves the state alone.
            defaults: false

      - id: when_approved
        type: io.kestra.plugin.core.flow.If
        description: An unanswered or declined pause must write nothing.
        condition: "{{ outputs.approve_import.onResume.approved ?? false }}"
        then:
          - id: recheck
            type: io.kestra.plugin.terraform.cli.TerraformCLI
            description: >-
              The approval is a snapshot of the past. Read the inventory and the
              plan again, then keep only the addresses whose live object still
              carries the id the human approved.
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
              networkMode: "{{ inputs.task_network }}"
            containerImage: hashicorp/terraform:1.10.5
            env:
              PG_CONN_STR: "postgres://{{ secret('POSTGRES_USERNAME') }}:{{
                secret('POSTGRES_PASSWORD') }}@{{ inputs.state_pg_endpoint }}/{{
                inputs.state_pg_database }}?sslmode=disable"
              TF_IN_AUTOMATION: "true"
              TF_VAR_docker_host: "{{ inputs.docker_host }}"
              TF_VAR_node_image: "{{ inputs.node_image }}"
              ALLOWED: "{{ inputs.allowed_address_prefixes | toJson }}"
              PROTECTED: "{{ inputs.protected_addresses | toJson }}"
              TYPES: "{{ inputs.importable_types | toJson }}"
              MAX_IMPORTS: "{{ inputs.max_imports }}"
              APPROVED: "{{ outputs.detect.vars.selected | toJson }}"
            inputFiles:
              main.tf: "{{ outputs.detect.outputFiles['main.tf'] }}"
              inventory.sh: "{{ outputs.detect.outputFiles['inventory.sh'] }}"
              reconcile.jq: "{{ outputs.detect.outputFiles['reconcile.jq'] }}"
            outputFiles:
              - main.tf
              - inventory.sh
              - confirmed.json
              - state_backup.json
            beforeCommands:
              - apk add --no-cache jq > /dev/null
              - terraform init -input=false -no-color
            commands:
              - |
                set -e
                sh inventory.sh
                terraform state pull > state_backup.json
                set +e
                terraform plan -refresh=true -detailed-exitcode -no-color -input=false -lock-timeout=60s -out=tfplan > replan.txt 2>&1
                PLAN_EXIT=$?
                set -e
                if [ "$PLAN_EXIT" = "1" ]; then
                  cat replan.txt
                  echo "terraform plan failed after the approval, nothing was written"
                  exit 1
                fi
                terraform show -json tfplan > plan.json
                terraform show -json > state.json
                jq -f reconcile.jq \
                  --slurpfile live live.json --slurpfile st state.json \
                  --argjson allowed "$ALLOWED" --argjson protected "$PROTECTED" \
                  --argjson types "$TYPES" --argjson max "$MAX_IMPORTS" \
                  plan.json > targets_now.json
                echo "$APPROVED" > approved.json
                jq -c -n --slurpfile a approved.json --slurpfile t targets_now.json '
                  ($a[0] // []) as $approved
                  | ($t[0].selected // []) as $now
                  | {confirmed: [$approved[] | . as $x | select($now | map(select(.address == $x.address and .live_id == $x.live_id)) | length > 0)],
                     stale: [$approved[] | . as $x | select(($now | map(select(.address == $x.address and .live_id == $x.live_id)) | length) == 0) | .address]}
                ' > diff.json
                jq -c .confirmed diff.json > confirmed.json
                echo "approved $(jq -r 'length' approved.json), still eligible $(jq -r '.confirmed | length' diff.json), stale $(jq -c .stale diff.json)"
                jq -r '.selected[] | "eligible now \(.address) live=\(.live_id)"' targets_now.json
                printf '::{"outputs":{"confirmed":%s,"confirmed_count":%s,"stale":%s,"stale_count":%s,"collisions_now":%s,"state_before":%s,"pending_in_scope_before":%s}}::\n' \
                  "$(jq -c .confirmed diff.json)" \
                  "$(jq -r '.confirmed | length' diff.json)" \
                  "$(jq -c .stale diff.json)" \
                  "$(jq -r '.stale | length' diff.json)" \
                  "$(jq -r .counts.collisions targets_now.json)" \
                  "$(jq -c .state_addresses targets_now.json)" \
                  "$(jq -r .counts.pending_in_scope targets_now.json)"

          - id: fail_when_plan_stale
            type: io.kestra.plugin.core.execution.Fail
            description: >-
              Something moved between the approval and now. Stop before the
              first write instead of importing an id that no longer exists.
            condition: "{{ outputs.recheck.vars.stale_count > 0 }}"
            errorMessage: >-
              The approved plan is stale. {{ outputs.recheck.vars.stale | toJson
              }} no longer match the live objects the approval was given for. No
              state was written. Run the flow again to approve the current plan.

          - id: import_targets
            type: io.kestra.plugin.core.flow.Loop
            description: >-
              One state write at a time. The pg backend takes a lock per write,
              and parallel imports would queue on it, so the loop is serial.
            values: "{{ outputs.recheck.vars.confirmed | toJson }}"
            concurrencyLimit: 1
            tasks:
              - id: import_one
                type: io.kestra.plugin.terraform.cli.TerraformCLI
                description: >-
                  Read this one object from the Engine API again, drop a state
                  entry whose object is gone, import the live id, then check the
                  resulting plan. An import that would destroy the object on the
                  next apply is removed from state again and fails the run.
                taskRunner:
                  type: io.kestra.plugin.scripts.runner.docker.Docker
                  networkMode: "{{ inputs.task_network }}"
                containerImage: hashicorp/terraform:1.10.5
                env:
                  PG_CONN_STR: "postgres://{{ secret('POSTGRES_USERNAME') }}:{{
                    secret('POSTGRES_PASSWORD') }}@{{ inputs.state_pg_endpoint
                    }}/{{ inputs.state_pg_database }}?sslmode=disable"
                  TF_IN_AUTOMATION: "true"
                  TF_VAR_docker_host: "{{ inputs.docker_host }}"
                  TF_VAR_node_image: "{{ inputs.node_image }}"
                  ADDR: "{{ fromJson(item.value).address }}"
                  OBJ_TYPE: "{{ fromJson(item.value).type }}"
                  OBJ_NAME: "{{ fromJson(item.value).name }}"
                  LIVE_ID: "{{ fromJson(item.value).live_id }}"
                inputFiles:
                  main.tf: "{{ outputs.recheck.outputFiles['main.tf'] }}"
                  inventory.sh: "{{ outputs.recheck.outputFiles['inventory.sh'] }}"
                beforeCommands:
                  - apk add --no-cache jq > /dev/null
                  - terraform init -input=false -no-color
                commands:
                  - |
                    set -e
                    sh inventory.sh
                    FRESH=$(jq -r --arg t "$OBJ_TYPE" --arg n "$OBJ_NAME" \
                      '[.[] | select(.type == $t and .name == $n)] | .[0].live_id // ""' live.json)
                    if [ "$FRESH" != "$LIVE_ID" ]; then
                      echo "refusing $ADDR: the approved object was $LIVE_ID, the live object is now ${FRESH:-gone}"
                      exit 1
                    fi
                    if terraform state list | grep -Fxq "$ADDR"; then
                      RECORDED=$(terraform show -json | jq -r --arg a "$ADDR" \
                        '[.values.root_module.resources[]? | select(.address == $a)] | .[0].values.id // ""')
                      if [ "$RECORDED" = "$LIVE_ID" ]; then
                        echo "$ADDR already records $LIVE_ID, this run writes nothing for it"
                        exit 0
                      fi
                      echo "state entry for $ADDR records ${RECORDED:-nothing}, which is not the live object, removing it first"
                      terraform state rm -lock-timeout=120s "$ADDR"
                    fi
                    terraform import -input=false -no-color -lock-timeout=120s "$ADDR" "$LIVE_ID"
                    terraform plan -refresh=false -no-color -input=false -lock-timeout=60s -out=after.tfplan > after_plan.txt 2>&1 || {
                      cat after_plan.txt
                      echo "cannot plan after importing $ADDR"
                      exit 1
                    }
                    DESTROY=$(terraform show -json after.tfplan | jq -r --arg a "$ADDR" \
                      '[.resource_changes[]? | select(.address == $a) | select((.change.actions | index("delete")) != null)] | length')
                    if [ "$DESTROY" != "0" ]; then
                      echo "the imported object does not match the configuration for $ADDR, the next apply would destroy it"
                      terraform state rm -lock-timeout=120s "$ADDR"
                      echo "the import was undone, $ADDR is unmanaged again and the object is untouched"
                      exit 1
                    fi
                    echo "imported $ADDR as $LIVE_ID"

          - id: verify
            type: io.kestra.plugin.terraform.cli.TerraformCLI
            description: >-
              Read the system a third time. Every approved address must now hold
              the live id, the live object must still carry the id it had before
              the run and the plan must no longer want to create or destroy it.
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
              networkMode: "{{ inputs.task_network }}"
            containerImage: hashicorp/terraform:1.10.5
            env:
              PG_CONN_STR: "postgres://{{ secret('POSTGRES_USERNAME') }}:{{
                secret('POSTGRES_PASSWORD') }}@{{ inputs.state_pg_endpoint }}/{{
                inputs.state_pg_database }}?sslmode=disable"
              TF_IN_AUTOMATION: "true"
              TF_VAR_docker_host: "{{ inputs.docker_host }}"
              TF_VAR_node_image: "{{ inputs.node_image }}"
              ALLOWED: "{{ inputs.allowed_address_prefixes | toJson }}"
              APPROVED: "{{ outputs.recheck.vars.confirmed | toJson }}"
              STATE_BEFORE: "{{ outputs.recheck.vars.state_before | toJson }}"
              RESIDUAL: "{{ inputs.tolerated_attributes | toJson }}"
            inputFiles:
              main.tf: "{{ outputs.recheck.outputFiles['main.tf'] }}"
              inventory.sh: "{{ outputs.recheck.outputFiles['inventory.sh'] }}"
              verify.jq: |
                # Verify a reconcile against the system itself.
                # stdin: terraform show -json <planfile>, taken after the imports
                # $live: provider inventory read again
                # $st: terraform show -json, the state now
                # $approved: the targets the loop was given
                # $before: state addresses recorded before the imports
                # $residual: attributes the provider cannot read back on import
                ($live[0] // []) as $live_objects
                | ($st[0].values.root_module.resources // []) as $state_resources
                | ([$state_resources[] | .address] | sort) as $state_now
                | ($approved[0] // []) as $targets
                | ($before[0] // []) as $state_before
                | [.resource_changes[]?
                   | select(.change.actions != ["no-op"])
                   | {address: .address,
                      actions: .change.actions,
                      attrs: (.change as $c
                              | (($c.after // {}) as $a | ($c.before // {}) as $b
                                 | (($a | keys) + ($b | keys) | unique
                                    | map(select((($a[.]) // null) != (($b[.]) // null))))))}]
                  as $changes
                | [$targets[]
                   | . as $t
                   | ($live_objects | map(select(.type == $t.type and .name == $t.name)) | .[0].live_id // null) as $live_now
                   | ($state_resources | map(select(.address == $t.address)) | .[0].values.id // null) as $state_id
                   | ($changes | map(select(.address == $t.address)) | .[0]) as $chg
                   | {address: $t.address,
                      name: $t.name,
                      approved_live_id: $t.live_id,
                      live_id_now: $live_now,
                      state_id_now: $state_id,
                      plan_actions: ($chg.actions // ["no-op"]),
                      changed_attrs: ($chg.attrs // []),
                      unexpected_attrs: (($chg.attrs // []) - $residual),
                      in_state: (($state_now | index($t.address)) != null),
                      object_survived: ($live_now != null and $live_now == $t.live_id),
                      state_matches_object: ($state_id != null and $state_id == $t.live_id),
                      plan_is_quiet: ((($chg.actions // ["no-op"]) | map(select(. != "no-op" and . != "update")) | length) == 0),
                      residual_is_allowed: (((($chg.attrs // []) - $residual) | length) == 0)}]
                  as $results
                | [$targets[] | .address] as $target_addresses
                | ([$targets[] | select(.in_state == false) | .address] | sort) as $expected_added
                | ([$targets[] | select(.in_state) | .address] | sort) as $reimported
                | (($state_now - $state_before) | sort) as $state_added
                | [$changes[] | .address] as $pending
                | [$pending[] | select(. as $a | ([$allowed[] | select(. as $p | ($a | startswith($p)))] | length) > 0)] as $pending_in_scope
                | {ok: (([$results[] | select(.object_survived and .state_matches_object and .in_state and .plan_is_quiet and .residual_is_allowed)] | length) == ($targets | length)
                        and ($state_added == $expected_added)),
                   counts: {approved: ($targets | length),
                            reconciled: ([$results[] | select(.object_survived and .state_matches_object and .in_state and .plan_is_quiet and .residual_is_allowed)] | length),
                            objects_replaced: ([$results[] | select(.object_survived == false)] | length),
                            pending_changes: ($changes | length),
                            pending_in_scope: ($pending_in_scope | length),
                            unexpected_attr_targets: ([$results[] | select(.residual_is_allowed == false)] | length),
                            reimported_in_place: ($reimported | length)},
                   state_added: $state_added,
                   state_added_expected: $expected_added,
                   reimported_in_place: $reimported,
                   state_now: $state_now,
                   pending_in_scope: ($pending_in_scope | sort),
                   targets: $results}
            outputFiles:
              - verified.json
              - replan.txt
            beforeCommands:
              - apk add --no-cache jq > /dev/null
              - terraform init -input=false -no-color
            commands:
              - |
                set -e
                sh inventory.sh
                echo "$APPROVED" > approved.json
                echo "$STATE_BEFORE" > state_before.json
                set +e
                terraform plan -refresh=true -detailed-exitcode -no-color -input=false -lock-timeout=60s -out=tfplan > replan.txt 2>&1
                PLAN_EXIT=$?
                set -e
                if [ "$PLAN_EXIT" = "1" ]; then
                  cat replan.txt
                  echo "terraform plan failed after the imports"
                  exit 1
                fi
                terraform show -json tfplan > plan.json
                terraform show -json > state.json
                jq -f verify.jq \
                  --slurpfile live live.json --slurpfile st state.json \
                  --slurpfile approved approved.json --slurpfile before state_before.json \
                  --argjson residual "$RESIDUAL" --argjson allowed "$ALLOWED" \
                  plan.json > verified.json
                echo "counts $(jq -c .counts verified.json)"
                jq -r '.targets[] | "verified \(.address) state=\(.state_id_now // "none") live=\(.live_id_now // "gone") actions=\(.plan_actions | join(",")) unexpected=\(.unexpected_attrs | join(","))"' verified.json
                echo "state entries added $(jq -c .state_added verified.json), expected $(jq -c .state_added_expected verified.json)"
                printf '::{"outputs":{"ok":%s,"reconciled":%s,"approved":%s,"objects_replaced":%s,"pending_in_scope":%s,"unexpected_attr_targets":%s,"reimported_in_place":%s,"state_added":%s,"state_added_expected":%s}}::\n' \
                  "$(jq -r .ok verified.json)" \
                  "$(jq -r .counts.reconciled verified.json)" \
                  "$(jq -r .counts.approved verified.json)" \
                  "$(jq -r .counts.objects_replaced verified.json)" \
                  "$(jq -r .counts.pending_in_scope verified.json)" \
                  "$(jq -r .counts.unexpected_attr_targets verified.json)" \
                  "$(jq -r .counts.reimported_in_place verified.json)" \
                  "$(jq -c .state_added verified.json)" \
                  "$(jq -c .state_added_expected verified.json)"

          - id: fail_when_not_reconciled
            type: io.kestra.plugin.core.execution.Fail
            description: The imports going green proves nothing. Only the re-read counts.
            condition: "{{ outputs.verify.vars.ok == false }}"
            errorMessage: >-
              Reconcile did not land. {{ outputs.verify.vars.reconciled }} of {{
              outputs.verify.vars.approved }} approved addresses hold their live
              id, {{ outputs.verify.vars.objects_replaced }} live objects no
              longer carry the id they had before the run, {{
              outputs.verify.vars.unexpected_attr_targets }} show an attribute
              difference outside tolerated_attributes and state gained {{
              outputs.verify.vars.state_added | toJson }} where {{
              outputs.verify.vars.state_added_expected | toJson }} was expected.
              The state backup saved before the writes is on the recheck task as
              state_backup.json.

          - id: reconcile_confirmed
            type: io.kestra.plugin.core.log.Log
            description: Before and after, both read back from the system itself.
            message: |
              imported {{ outputs.verify.vars.reconciled }} of {{ outputs.verify.vars.approved }} approved addresses
              state entries {{ outputs.detect.vars.state_entries }} before, {{ outputs.verify.vars.state_added | toJson }} added
              addresses whose stale entry was replaced in place {{ outputs.verify.vars.reimported_in_place }}
              pending changes inside the allowed prefixes {{ outputs.detect.vars.pending_in_scope }} before, {{ outputs.verify.vars.pending_in_scope }} after
              live objects whose id changed during the run {{ outputs.verify.vars.objects_replaced }}
              every imported object kept the id it was created with by hand, so nothing was rebuilt

        else:
          - id: import_declined
            type: io.kestra.plugin.core.log.Log
            description: Nothing was written. Say so.
            message: |
              the import of {{ outputs.detect.vars.selected | toJson }} was declined or the pause went unanswered
              state entries are still {{ outputs.detect.vars.state_entries }}
              nothing was imported, nothing was removed from state

    else:
      - id: no_write
        type: io.kestra.plugin.core.log.Log
        description: Either there is nothing to reconcile or this run is read only.
        message: |
          unmanaged resources that collide with a live object {{ outputs.detect.vars.collisions }}
          {{ (outputs.detect.vars.collisions == 0) ? 'every planned create is new work, not an unmanaged resource' : 'collisions found but this run is read only' }}
          dry_run {{ inputs.dry_run }}, auto_remediate {{ inputs.auto_remediate }}
          terraform plan detailed exit code {{ outputs.detect.vars.plan_exit }}, pending changes inside the allowed prefixes {{ outputs.detect.vars.pending_in_scope }}

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Tell the on-call channel which stage failed and on which endpoint.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": ":rotating_light: terraform-unmanaged-resource-import-reconcile failed on {{ inputs.docker_host }}. Execution {{ execution.id }}, flow {{ flow.namespace }}.{{ flow.id }}."
      }

triggers:
  - id: hourly_drift_scan
    type: io.kestra.plugin.core.trigger.Schedule
    description: Hourly read only scan. Enable it once the endpoint, the
      configuration and the allowed prefixes are yours.
    cron: "23 * * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false