# Sourced by managed-container; uses the shared OCF and parameter environment.

node_to_addr()
{
    local target="${1:-}"

    [ -n "${target}" ] || return 1

    corosync-cmapctl 2>/dev/null |
    awk -v target="${target}" '
        /nodelist\.node\.[0-9]+\.name / {
            key=$1
            sub(/\.name$/, "", key)
            names[key]=$NF
        }

        /nodelist\.node\.[0-9]+\.ring0_addr / {
            key=$1
            sub(/\.ring0_addr$/, "", key)
            addrs[key]=$NF
        }

        END {
            for (key in names) {
                if (names[key] == target && addrs[key] != "") {
                    print addrs[key]
                    exit 0
                }
            }
            exit 1
        }
    '
}

# Read the final checkpoint only after pg_ctl has confirmed clean primary shutdown.
mc_pg_shutdown_checkpoint()
{
    local control summary
    control="$(mc_runtime_exec_bounded 3s --user "${PG_OS_USER}" --env LC_ALL=C \
        "${CONTAINER_NAME}" sh -c \
        'exec "$(pg_config --bindir)/pg_controldata" "$PGDATA"')" || return 1
    summary="$(printf '%s\n' "${control}" | awk -F ': *' '
        /^Database cluster state:/ { state=$2; ns++ }
        /^Database system identifier:/ { sid=$2; ni++ }
        /^Latest checkpoint location:/ { checkpoint=$2; nc++ }
        /^Latest checkpoint.s TimeLineID:/ { timeline=$2; nt++ }
        END {
            if (ns != 1 || ni != 1 || nc != 1 || nt != 1 || state != "shut down") exit 1;
            print sid "|" timeline "|" checkpoint;
        }')" || return 1
    [[ "${summary}" =~ ^[0-9]+\|[1-9][0-9]*\|[0-9A-F]+/[0-9A-F]+$ ]] || return 1
    printf '%s\n' "${summary}"
}

# A checkpoint LSN is its record START. Replay must be strictly beyond it to
# establish that the complete shutdown checkpoint has been replayed.
mc_pg_wait_shutdown_replay()
{
    local host="$1" port="$2" observation="$3" deadline="$4"
    local system timeline checkpoint response ready identity rest got_system got_timeline
    local remaining duration
    IFS='|' read -r system timeline checkpoint <<< "${observation}"
    [[ "${observation}" =~ ^[0-9]+\|[1-9][0-9]*\|[0-9A-F]+/[0-9A-F]+$ ]] || return 1
    while (( SECONDS < deadline )); do
        remaining=$((deadline - SECONDS))
        duration=3
        (( remaining < duration )) && duration=${remaining}
        response="$(mc_runtime_exec_bounded "${duration}s" --user "${PG_OS_USER}" \
            --env PGCONNECT_TIMEOUT=2 --env 'PGOPTIONS=-c statement_timeout=1000' \
            "${CONTAINER_NAME}" sh -c '
                export PGPASSWORD="${POSTGRES_REPLICATION_PASSWORD:?}"
                psql -X -w -v ON_ERROR_STOP=1 -h "$1" -p "$2" -U "$3" -d postgres -Atqc \
                    "SELECT CASE WHEN pg_is_in_recovery() THEN
                        NOT pg_is_wal_replay_paused() AND
                        pg_last_wal_replay_lsn() > '\''$4'\''::pg_lsn
                     ELSE false END;" || exit 1
                exec psql -X -w -v ON_ERROR_STOP=1 -h "$1" -p "$2" -U "$3" \
                    -d replication=true -Atqc IDENTIFY_SYSTEM
            ' sh "${host}" "${port}" "${REPLICATION_USER}" "${checkpoint}" 2>/dev/null)" || response=''
        ready="${response%%$'\n'*}"
        identity="${response#*$'\n'}"
        IFS='|' read -r got_system got_timeline rest <<< "${identity}"
        if [ "${ready}" = t ] && [ "${got_system}" = "${system}" ] &&
           [ "${got_timeline}" = "${timeline}" ]; then
            return 0
        fi
        (( SECONDS < deadline )) && sleep 1
    done
    return 1
}

managed_demote()
{
    local monitor_rc
    local target_node target_addr
    local local_node slot_name target_port
    local role receiver_status i checkpoint handoff_deadline
    local action_started=${SECONDS} action_budget=${OCF_RESKEY_CRM_meta_timeout:-60000}
    [[ "${action_budget}" =~ ^[0-9]{1,9}$ ]] || return "${OCF_ERR_CONFIGURED}"
    action_budget=$((10#${action_budget} / 1000))
    if (( action_budget < 15 )); then
        ocf_exit_reason "Demotion timeout is too short for a verified shutdown-WAL handoff"
        return "${OCF_ERR_CONFIGURED}"
    fi

    managed_monitor
    monitor_rc=$?

    case "${monitor_rc}" in
        "${OCF_RUNNING_PROMOTED}")
            ;;
        "${OCF_SUCCESS}")
            ocf_log info \
                "PostgreSQL is already unpromoted"
            return "${OCF_SUCCESS}"
            ;;
        "${OCF_NOT_RUNNING}")
            ocf_exit_reason \
                "Cannot complete demotion because PostgreSQL is stopped"
            return "${OCF_ERR_GENERIC}"
            ;;
        *)
            ocf_exit_reason \
                "Cannot demote unhealthy PostgreSQL instance (monitor rc=${monitor_rc})"
            return "${monitor_rc}"
            ;;
    esac

    target_node="$(demote_target)" || {
        ocf_exit_reason \
            "No fresh valid Pacemaker pre-demote target is available"
        return "${OCF_ERR_GENERIC}"
    }

    target_addr="$(node_to_addr "${target_node}")" || {
        ocf_exit_reason \
            "Unable to resolve Pacemaker promotion target ${target_node} through Corosync"
        return "${OCF_ERR_GENERIC}"
    }

    target_port="${REJOIN_SOURCE_PORT:-5432}"
    local_node="${OCF_RESKEY_CRM_meta_on_node:-$(hostname)}"
    slot_name="ha_$(printf '%s' "${local_node}" |
        tr '[:upper:]-' '[:lower:]_')"

    ocf_log info \
        "Preparing controlled demotion of ${local_node}; successor=${target_node} (${target_addr})"

    if ! candidate_is_standby "${target_addr}"; then
        ocf_exit_reason \
            "Promotion target ${target_node} is not a reachable PostgreSQL standby"
        return "${OCF_ERR_GENERIC}"
    fi

    if ! candidate_caught_up "${target_node}"; then
        ocf_exit_reason \
            "Promotion target ${target_node} is not streaming and fully replayed to the current flushed WAL position"
        return "${OCF_ERR_GENERIC}"
    fi

    if ! ensure_remote_slot "${target_addr}" "${slot_name}"; then
        ocf_exit_reason \
            "Unable to ensure physical replication slot ${slot_name} on ${target_node}"
        return "${OCF_ERR_GENERIC}"
    fi

    ocf_log info \
        "Demotion safety checks passed; stopping PostgreSQL primary"

    if ! pg_command '
        BINDIR="$(pg_config --bindir)" || exit 1
        exec "$BINDIR/pg_ctl" \
            -D "$PGDATA" \
            stop -m fast -w
    '
    then
        ocf_exit_reason \
            "pg_ctl stop failed during controlled demotion"
        return "${OCF_ERR_GENERIC}"
    fi

    if pg_running; then
        ocf_exit_reason \
            "PostgreSQL remained running after pg_ctl stop"
        return "${OCF_ERR_GENERIC}"
    fi

    if ! service_running || ! container_running; then
        ocf_exit_reason \
            "Maintenance service/container stopped unexpectedly during demotion"
        return "${OCF_ERR_GENERIC}"
    fi

    # Keep at most 20 seconds for the handoff, reserving 10 seconds of the
    # action budget for subsequent work. Pacemaker remains the overall timeout.
    checkpoint="$(mc_pg_shutdown_checkpoint)" || {
        ocf_exit_reason "Unable to read a valid clean-shutdown checkpoint; PostgreSQL remains stopped"
        return "${OCF_ERR_GENERIC}"
    }
    handoff_deadline=$((action_started + action_budget - 10))
    (( handoff_deadline > SECONDS + 20 )) && handoff_deadline=$((SECONDS + 20))
    if ! mc_pg_wait_shutdown_replay "${target_addr}" "${target_port}" "${checkpoint}" "${handoff_deadline}"; then
        ocf_exit_reason "Successor has not verified final shutdown-checkpoint replay; PostgreSQL remains stopped"
        return "${OCF_ERR_GENERIC}"
    fi

    ocf_log info \
        "Configuring ${local_node} as cascading standby of ${target_node}"

    if ! mc_runtime_exec \
        --user "${PG_OS_USER}" \
        --env "RA_PRIMARY_HOST=${target_addr}" \
        --env "RA_PRIMARY_PORT=${target_port}" \
        --env "RA_REPLICATION_USER=${REPLICATION_USER}" \
        --env "RA_REPLICATION_SLOT=${slot_name}" \
        --env "RA_APPLICATION_NAME=${local_node}" \
        "${CONTAINER_NAME}" \
        sh -c '
            set -e

            sed -i \
                -e "/^[[:space:]]*primary_conninfo[[:space:]]*=/d" \
                -e "/^[[:space:]]*primary_slot_name[[:space:]]*=/d" \
                "$PGDATA/postgresql.auto.conf"

            printf "%s
" \
                "primary_conninfo = '\''host=${RA_PRIMARY_HOST} port=${RA_PRIMARY_PORT} user=${RA_REPLICATION_USER} application_name=${RA_APPLICATION_NAME}'\''" \
                "primary_slot_name = '\''${RA_REPLICATION_SLOT}'\''" \
                >> "$PGDATA/postgresql.auto.conf"

            touch "$PGDATA/standby.signal"

            chmod 600 \
                "$PGDATA/postgresql.auto.conf" \
                "$PGDATA/standby.signal"
        '
    then
        ocf_exit_reason \
            "Failed to configure PostgreSQL as standby of ${target_node}"
        return "${OCF_ERR_GENERIC}"
    fi

    ocf_log info \
        "Starting PostgreSQL as cascading standby of ${target_node}"

    if ! pg_command '
        BINDIR="$(pg_config --bindir)" || exit 1
        exec "$BINDIR/pg_ctl" \
            -D "$PGDATA" \
            -l "$PGDATA/postgresql.log" \
            start -w
    '
    then
        ocf_exit_reason \
            "PostgreSQL failed to start as cascading standby"
        return "${OCF_ERR_GENERIC}"
    fi

    for i in $(seq 1 30); do
        role="$(pg_role 2>/dev/null || true)"
        receiver_status="$(pg_wal_receiver_status 2>/dev/null || true)"

        if [ "${role}" = "t" ] &&
           [ "${receiver_status}" = "streaming" ]
        then
            ocf_log info \
                "Controlled demotion complete: ${local_node} is streaming from ${target_node}"
            return "${OCF_SUCCESS}"
        fi

        sleep 1
    done

    ocf_exit_reason \
        "PostgreSQL started after demotion but did not become a streaming standby"
    return "${OCF_ERR_GENERIC}"
}