#!/usr/bin/perl

# hitachiblock-repl - Replication management CLI for Hitachi Block Storage
#
# Manages TrueCopy (sync), Universal Replicator (async), and
# Global-Active Device (GAD) replication pairs.
#
# Usage:
#   hitachiblock-repl <command> --storeid <id> [options]
#
# Commands:
#   list          List all replication pairs
#   create-tc     Create TrueCopy (sync replication) pair
#   create-ur     Create Universal Replicator (async replication) pair
#   create-gad    Create Global-Active Device pair
#   status        Show pair status
#   split         Split (suspend) a replication pair
#   resync        Resync a split pair
#   delete        Delete a replication pair
#   remote-list   List registered remote storage systems
#   remote-add    Register a remote storage system
#   orphans       Report LDEV/registry inconsistencies for a storage
#   diagnostics   Read-only support bundle (config/REST health/drift/devices)
#   lun-paths     Report per-node LU-path/host-group counts + headroom (read-only)
#   reconcile-maps Unmap orphan LU paths on this node (dry-run unless --apply)
#   pr-check      SCSI-3 PR readiness: qemu-pr-helper + multipath reservation_key (read-only, no array login)

use strict;
use warnings;

use Getopt::Long;
use Pod::Usage;
use Time::HiRes ();
use JSON ();

use lib '/usr/share/perl5';

use PVE::Storage::HitachiBlock::RestClient;
use PVE::Storage::HitachiBlock::Config;
use PVE::Storage::HitachiBlock::Multipath;
use PVE::Storage::HitachiBlock::LunPaths;

# Per-FE-port LU-path budget (model-dependent on VSP; the issue cites 2048/4096).
# Used for headroom reporting and the soft over-budget warning; override with
# --lun-path-budget. The per-port host-group cap is a fixed 255.
my $DEFAULT_LUN_PATH_BUDGET = 2048;
my $HOST_GROUPS_PER_PORT_CAP = 255;

my $command = shift @ARGV || '';

my %opts;
GetOptions(\%opts,
    'storeid=s',
    'remote-storage-id=s',
    'remote-ip=s',
    'pvol=i',
    'svol=i',
    'volume=s',
    'copy-group=s',
    'pair-name=s',
    'pair-id=s',
    'copy-pace=i',
    'journal-id=i',
    'quorum-disk-id=i',
    'consistency-group',
    'mgmt-ip=s',
    'storage-id=s',
    'mgmt-port=i',
    'auto-cleanup',
    'lun-path-budget=i',
    'apply',
    'json',
    'help',
) or pod2usage(2);

pod2usage(1) if $opts{help};

# pr-check is purely host-side (qemu-pr-helper + multipathd). It needs no storeid,
# no credentials, and no array connection — dispatch it before any of that setup so
# it works standalone on any node, even without a configured storage.
if ($command eq 'pr-check') {
    cmd_pr_check();
    exit 0;
}

pod2usage("storeid is required") unless $opts{storeid};

my $config = PVE::Storage::HitachiBlock::Config->new(storeid => $opts{storeid});
my ($username, $password) = $config->read_credentials();

# Resolve connection parameters. Precedence:
#   1. explicit CLI flags (--mgmt-ip / --storage-id / --mgmt-port)
#   2. /etc/pve/storage.cfg for this storeid (the same source the plugin uses)
#   3. HITACHI_* environment variables (legacy fallback)
my $scfg = read_storage_cfg($opts{storeid});

my $mgmt_ip = $opts{'mgmt-ip'} || $scfg->{mgmt_ip} || $ENV{HITACHI_MGMT_IP}
    or die "mgmt_ip not found: pass --mgmt-ip, define the storage in /etc/pve/storage.cfg, or set HITACHI_MGMT_IP\n";
my $storage_id = $opts{'storage-id'} || $scfg->{storage_id} || $ENV{HITACHI_STORAGE_ID}
    or die "storage_id not found: pass --storage-id, define the storage in /etc/pve/storage.cfg, or set HITACHI_STORAGE_ID\n";
my $port = $opts{'mgmt-port'} || $scfg->{mgmt_port} || $ENV{HITACHI_MGMT_PORT}
    || platform_port($scfg->{platform});

my $client = PVE::Storage::HitachiBlock::RestClient->new(
    mgmt_ip     => $mgmt_ip,
    storage_id  => $storage_id,
    username    => $username,
    password    => $password,
    port        => $port,
    tls_verify  => $scfg->{tls_verify},
    tls_ca_file => $scfg->{tls_ca_file},
);

# `diagnostics` is a resilient, read-only support bundle: it must produce output
# even when the management endpoint is unreachable, so it does its own guarded
# (and timed) login probe rather than failing here. (`pr-check` is dispatched
# earlier and never reaches this array setup at all.)
$client->login() unless $command eq 'diagnostics';

my %DISPATCH = (
    'list'           => \&cmd_list,
    'create-tc'      => sub { cmd_create('TC') },
    'create-ur'      => sub { cmd_create('UR') },
    'create-gad'     => sub { cmd_create('GAD') },
    'status'         => \&cmd_status,
    'split'          => \&cmd_split,
    'resync'         => \&cmd_resync,
    'delete'         => \&cmd_delete,
    'remote-list'    => \&cmd_remote_list,
    'remote-add'     => \&cmd_remote_add,
    'orphans'        => \&cmd_orphans,
    'diagnostics'    => \&cmd_diagnostics,
    'lun-paths'      => \&cmd_lun_paths,
    'reconcile-maps' => \&cmd_reconcile_maps,
    # NOTE: pr-check is dispatched early (above), before storeid/array setup —
    # it is host-side only and intentionally not in this array-backed table.
);

eval {
    my $handler = $DISPATCH{$command}
        or die "Unknown command: '$command'\nRun with --help for usage.\n";
    $handler->();
};

my $err = $@;
eval { $client->logout() };
die $err if $err;

exit 0;

# ── Command Implementations ──

sub cmd_list {
    my $pairs = $client->list_remote_copy_pairs();

    return emit_json($pairs) if $opts{json};

    printf "%-12s %-20s %-6s %-8s %-8s %-10s\n",
        'PAIR_ID', 'COPY_GROUP', 'TYPE', 'P-VOL', 'S-VOL', 'STATUS';
    print "-" x 70 . "\n";

    for my $pair (@$pairs) {
        printf "%-12s %-20s %-6s %-8s %-8s %-10s\n",
            $pair->{copyPairId} || '-',
            $pair->{copyGroupName} || '-',
            $pair->{replicationType} || '-',
            $pair->{pvolLdevId} // '-',
            $pair->{svolLdevId} // '-',
            $pair->{pairStatus} || '-';
    }
}

sub cmd_create {
    my ($type) = @_;

    # Allow --volume <pve-volname> as a convenience for --pvol <ldev-id>.
    if (!defined $opts{pvol} && $opts{volume}) {
        $opts{pvol} = resolve_volume_ldev($opts{volume});
    }

    die "--pvol (or --volume) is required\n" unless defined $opts{pvol};
    die "--svol is required\n"             unless defined $opts{svol};
    die "--remote-storage-id is required\n" unless $opts{'remote-storage-id'};
    die "--copy-group is required\n"       unless $opts{'copy-group'};

    my %create_opts = (
        replication_type   => $type,
        pvol_ldev_id       => $opts{pvol},
        svol_ldev_id       => $opts{svol},
        remote_storage_id  => $opts{'remote-storage-id'},
        copy_group_name    => $opts{'copy-group'},
        copy_pace          => $opts{'copy-pace'} || 3,
        consistency_group  => $opts{'consistency-group'} || 0,
    );

    $create_opts{copy_pair_name} = $opts{'pair-name'} if $opts{'pair-name'};
    $create_opts{journal_id}     = $opts{'journal-id'} if defined $opts{'journal-id'};
    $create_opts{quorum_disk_id} = $opts{'quorum-disk-id'} if defined $opts{'quorum-disk-id'};

    my $result = $client->create_remote_copy_pair(%create_opts);
    print "Created $type pair: " . ($result->{resourceId} || $result->{jobId} || 'OK') . "\n";
}

sub cmd_status {
    die "--pair-id is required\n" unless $opts{'pair-id'};

    my $pair = $client->get_remote_copy_pair($opts{'pair-id'});

    return emit_json($pair) if $opts{json};

    print "Pair ID:      " . ($pair->{copyPairId} || '-') . "\n";
    print "Copy Group:   " . ($pair->{copyGroupName} || '-') . "\n";
    print "Type:         " . ($pair->{replicationType} || '-') . "\n";
    print "P-VOL LDEV:   " . ($pair->{pvolLdevId} // '-') . "\n";
    print "S-VOL LDEV:   " . ($pair->{svolLdevId} // '-') . "\n";
    print "Status:       " . ($pair->{pairStatus} || '-') . "\n";
    print "Progress:     " . ($pair->{progressRate} // '-') . "%\n" if defined $pair->{progressRate};
}

sub cmd_split {
    die "--pair-id is required\n" unless $opts{'pair-id'};

    $client->split_remote_copy_pair($opts{'pair-id'});
    print "Split pair: $opts{'pair-id'}\n";
}

sub cmd_resync {
    die "--pair-id is required\n" unless $opts{'pair-id'};

    $client->resync_remote_copy_pair($opts{'pair-id'});
    print "Resynced pair: $opts{'pair-id'}\n";
}

sub cmd_delete {
    die "--pair-id is required\n" unless $opts{'pair-id'};

    $client->delete_remote_copy_pair($opts{'pair-id'});
    print "Deleted pair: $opts{'pair-id'}\n";
}

sub cmd_remote_list {
    my $remotes = $client->list_remote_storages();

    return emit_json($remotes) if $opts{json};

    printf "%-20s %-16s %-10s\n", 'REMOTE_ID', 'IP', 'PATH_GROUP';
    print "-" x 50 . "\n";

    for my $r (@$remotes) {
        printf "%-20s %-16s %-10s\n",
            $r->{remoteStorageDeviceId} || '-',
            $r->{remoteIpAddress} || '-',
            $r->{pathGroupId} // '-';
    }
}

sub cmd_remote_add {
    die "--remote-storage-id is required\n" unless $opts{'remote-storage-id'};
    die "--remote-ip is required\n"         unless $opts{'remote-ip'};

    $client->register_remote_storage(
        remote_storage_id => $opts{'remote-storage-id'},
        remote_ip         => $opts{'remote-ip'},
    );
    print "Registered remote storage: $opts{'remote-storage-id'}\n";
}

# Compute registry-vs-array drift. Returns a hashref:
#   { array_orphans => [...], registry_orphans => [...], scan_min, scan_max }
# Shared by `orphans` and `diagnostics` (read-only; no mutation here).
sub scan_registry_vs_array {
    my $registry = $config->list_registered();

    my %registered;
    for my $volname (keys %$registry) {
        my $e = $registry->{$volname};
        next unless ref $e eq 'HASH' && defined $e->{ldev_id};
        $registered{$e->{ldev_id}} = $volname;
    }

    my $prefix = PVE::Storage::HitachiBlock::Config->label_prefix($opts{storeid});

    # Scan domain: the default list_ldevs() only returns LDEV slots 0-99, so a
    # plain scan misses every LDEV in a high ldev_range and would (a) miss array
    # orphans and (b) FALSELY flag live in-range registry entries as orphans
    # (their LDEV id is absent from the 0-99 window) — which --auto-cleanup would
    # then unregister. Page the configured ldev_range; if unset, cover at least
    # 0..max(registered id) so registry-orphan detection stays sound.
    my ($scan_min, $scan_max);
    if (my $r = $scfg->{ldev_range}) {
        ($scan_min, $scan_max) = $r =~ /^\s*(\d+)\s*-\s*(\d+)\s*$/
            or die "invalid ldev_range '$r' in storage.cfg\n";
    } else {
        $scan_min = 0;
        $scan_max = 99;
        for my $id (keys %registered) { $scan_max = $id if $id > $scan_max; }
        warn "ldev_range not set: scanning LDEVs $scan_min-$scan_max only; "
           . "array-orphan detection may be incomplete.\n";
    }
    my $ldevs = $client->list_defined_ldevs_in_range($scan_min, $scan_max);

    my %array_ids;
    my @array_orphans;
    for my $ldev (@$ldevs) {
        my $id = $ldev->{ldevId};
        next unless defined $id;
        $array_ids{$id} = 1;
        my $label = $ldev->{label} || '';
        next unless $label =~ /^\Q$prefix\E/;
        push @array_orphans, { ldev_id => $id, label => $label }
            unless $registered{$id};
    }

    my @registry_orphans;
    for my $volname (sort keys %$registry) {
        my $e = $registry->{$volname};
        next unless ref $e eq 'HASH' && defined $e->{ldev_id};
        push @registry_orphans, { volname => $volname, ldev_id => $e->{ldev_id} }
            unless $array_ids{$e->{ldev_id}};
    }

    return {
        array_orphans    => \@array_orphans,
        registry_orphans => \@registry_orphans,
        scan_min         => $scan_min,
        scan_max         => $scan_max,
    };
}

sub cmd_orphans {
    my $scan = scan_registry_vs_array();
    my @array_orphans    = @{$scan->{array_orphans}};
    my @registry_orphans = @{$scan->{registry_orphans}};

    # Optionally prune registry entries whose backing LDEV no longer exists. This
    # is safe because the scan above pages the whole ldev_range (where the plugin
    # creates every LDEV), so a live in-range volume is never mistaken for an orphan.
    my $cleaned = 0;
    if ($opts{'auto-cleanup'}) {
        for my $o (@registry_orphans) {
            $config->unregister_ldev($o->{volname});
            $cleaned++;
        }
    }

    return emit_json({
        array_orphans    => \@array_orphans,
        registry_orphans => \@registry_orphans,
        cleaned          => $cleaned,
    }) if $opts{json};

    print "Array orphans (labeled LDEVs not in registry):\n";
    print "  none\n" unless @array_orphans;
    printf "  LDEV %-8s %s\n", $_->{ldev_id}, $_->{label} for @array_orphans;

    print "Registry orphans (registry entries with no array LDEV):\n";
    print "  none\n" unless @registry_orphans;
    printf "  %-20s LDEV %s\n", $_->{volname}, $_->{ldev_id} for @registry_orphans;

    print "\nPruned $cleaned stale registry entr" . ($cleaned == 1 ? "y" : "ies") . ".\n"
        if $opts{'auto-cleanup'};
}

# Read-only support bundle (#18): one coherent, secret-free snapshot for bug
# reports. Performs NO mutations. Every array/host probe is guarded so the report
# is produced even when the management endpoint is down (REST health then shows
# the failure). REST response times are surfaced here — the GUM is intrinsically
# slow, so per-call latency is a primary troubleshooting signal.
sub cmd_diagnostics {
    my %r;   # structured report (also used for --json)

    $r{generated}     = scalar(localtime);
    $r{storeid}       = $opts{storeid};
    $r{plugin_version} = _plugin_version();
    chomp(my $hostname = `hostname -s 2>/dev/null` || '');
    $r{node} = $hostname;

    # Effective config, credentials redacted. password never lives in storage.cfg;
    # redact username (and any stray secret-ish keys) defensively.
    my %redacted = %$scfg;
    $redacted{username} = '<redacted>' if exists $redacted{username};
    delete $redacted{password};
    $r{config} = \%redacted;

    # REST health: timed login probe + array identity/microcode + accumulated
    # per-call timing. All guarded — a failure is reported, not fatal.
    my %rest = (endpoint => $mgmt_ip, storage_id => $storage_id);
    my $t0 = Time::HiRes::time();
    my $ok = eval { $client->login(); 1 };
    $rest{login_ms} = int((Time::HiRes::time() - $t0) * 1000);
    if ($ok) {
        $rest{reachable} = 1;
        my $info = eval { $client->get_storage_info() };
        if ($info) {
            $rest{model}      = $info->{model};
            $rest{serial}     = $info->{serialNumber} // $info->{serial};
            $rest{microcode}  = $info->{dkcMicroVersion} // $info->{microVersion};
        } else {
            $rest{info_error} = _trim($@);
        }
    } else {
        $rest{reachable} = 0;
        $rest{error}     = _trim($@);
    }
    $r{rest} = \%rest;

    # Registry-vs-array drift (guarded: needs REST).
    my $scan = eval { scan_registry_vs_array() };
    if ($scan) {
        $r{drift} = {
            array_orphans    => $scan->{array_orphans},
            registry_orphans => $scan->{registry_orphans},
            scan_range       => "$scan->{scan_min}-$scan->{scan_max}",
        };
    } else {
        $r{drift_error} = _trim($@);
    }

    # Host-side device state for every registered volume on THIS node.
    my $registry = eval { $config->list_registered() } || {};
    my $mp = eval { require PVE::Storage::HitachiBlock::Multipath;
                    PVE::Storage::HitachiBlock::Multipath->new() };
    my @devs;
    for my $volname (sort keys %$registry) {
        my $e = $registry->{$volname};
        next unless ref $e eq 'HASH' && defined $e->{ldev_id};
        my $wwid = $e->{wwid}
            || ($mp && eval { $mp->ldev_to_wwid($storage_id, $e->{ldev_id}) });
        my $path = ($mp && $wwid) ? eval { $mp->get_device_path($wwid) } : undef;
        push @devs, {
            volname => $volname,
            ldev_id => $e->{ldev_id},
            wwid    => $wwid,
            path    => $path,
            present => ($path && -e $path) ? 1 : 0,
            is_block => ($path && -b $path) ? 1 : 0,
        };
    }
    $r{devices} = \@devs;

    # LU-path / host-group counts for this node (guarded: needs REST).
    $r{lun_paths} = eval { build_lun_path_report() };

    # Snapshot the accumulated REST timing AFTER all probes above.
    if (my $ts = $client->rest_timing_stats()) {
        $r{rest}{timing} = {
            calls      => $ts->{count},
            avg_ms     => int($ts->{avg} * 1000),
            max_ms     => int($ts->{max} * 1000),
            last_ms    => int($ts->{last} * 1000),
            slowest    => $ts->{slowest},
        };
    }

    return emit_json(\%r) if $opts{json};

    # ── Human-readable report ──
    print "=== Hitachi Block diagnostics ===\n";
    printf "generated:   %s\n", $r{generated};
    printf "node:        %s\n", $r{node};
    printf "storeid:     %s\n", $r{storeid};
    printf "plugin:      %s\n", $r{plugin_version};

    print "\n-- effective config (secrets redacted) --\n";
    printf "  %-18s %s\n", $_, $redacted{$_} for sort keys %redacted;

    print "\n-- REST / management endpoint --\n";
    printf "  endpoint:    %s\n", $rest{endpoint};
    printf "  storage_id:  %s\n", $rest{storage_id};
    printf "  reachable:   %s (login %d ms)\n", ($rest{reachable} ? 'yes' : 'NO'), $rest{login_ms};
    printf "  error:       %s\n", $rest{error} if $rest{error};
    printf "  model:       %s\n", $rest{model} // '-' if $rest{reachable};
    printf "  serial:      %s\n", $rest{serial} // '-' if $rest{reachable};
    printf "  microcode:   %s\n", $rest{microcode} // '-' if $rest{reachable};
    printf "  info_error:  %s\n", $rest{info_error} if $rest{info_error};
    if (my $tm = $r{rest}{timing}) {
        printf "  timing:      %d calls, avg %d ms, max %d ms (slowest: %s)\n",
            $tm->{calls}, $tm->{avg_ms}, $tm->{max_ms}, ($tm->{slowest} // '-');
    }

    print "\n-- registry vs array drift --\n";
    if ($r{drift}) {
        my $d = $r{drift};
        printf "  scan range:  LDEV %s\n", $d->{scan_range};
        printf "  array orphans (labeled LDEVs not in registry): %d\n", scalar @{$d->{array_orphans}};
        printf "    LDEV %-8s %s\n", $_->{ldev_id}, $_->{label} for @{$d->{array_orphans}};
        printf "  registry orphans (no backing LDEV): %d\n", scalar @{$d->{registry_orphans}};
        printf "    %-20s LDEV %s\n", $_->{volname}, $_->{ldev_id} for @{$d->{registry_orphans}};
    } else {
        printf "  (unavailable: %s)\n", $r{drift_error} // 'unknown error';
    }

    print "\n-- host device state (this node) --\n";
    printf "  %d registered volume(s)\n", scalar @devs;
    for my $d (@devs) {
        printf "  %-20s LDEV %-6s %-40s present=%s block=%s\n",
            $d->{volname}, $d->{ldev_id}, ($d->{path} // '-'),
            ($d->{present} ? 'yes' : 'no'), ($d->{is_block} ? 'yes' : 'no');
    }

    print "\n-- LU-path / host-group counts (this node) --\n";
    if (my $lp = $r{lun_paths}) {
        for my $port (sort keys %{ $lp->{ports} }) {
            my $p = $lp->{ports}{$port};
            printf "  %-8s HG %-4s  LU-paths: %d   host-groups-on-port: %s / %d\n",
                $port, (defined $p->{host_group_number} ? $p->{host_group_number} : '-'),
                $p->{lun_path_count},
                (defined $p->{host_groups_on_port} ? $p->{host_groups_on_port} : '?'),
                $HOST_GROUPS_PER_PORT_CAP;
        }
        printf "  total mapped LU paths: %d ; orphan LU paths: %d\n",
            $lp->{total_lun_paths}, scalar @{ $lp->{orphans} };
        printf "  WARNING: %d orphan LU path(s) mapped here with no registry LDEV"
            . " (run 'reconcile-maps')\n", scalar @{ $lp->{orphans} }
            if @{ $lp->{orphans} };
        printf "  WARNING: port %s at %d%% of LU-path budget (%d/%d)\n",
            $_->{port}, $_->{pct}, $_->{count}, $_->{budget} for @{ $lp->{over_budget} };
    } else {
        print "  (unavailable)\n";
    }

    print "\n=== end diagnostics ===\n";
}

# Build a LunPaths scan report for THIS node against $storeid. Read-only.
sub build_lun_path_report {
    my $multipath = PVE::Storage::HitachiBlock::Multipath->new();
    my $wwns = $multipath->get_local_wwns();

    my $registry = $config->list_registered();
    my %registered;
    for my $e (values %$registry) {
        next unless ref $e eq 'HASH' && defined $e->{ldev_id};
        $registered{ $e->{ldev_id} } = 1;
    }

    my @ports = grep { length } map { s/^\s+|\s+$//gr } split(/,/, $scfg->{target_ports} || '');

    return PVE::Storage::HitachiBlock::LunPaths->scan(
        client           => $client,
        wwns             => $wwns,
        target_ports     => \@ports,
        registered_ldevs => \%registered,
        per_port_budget  => lun_path_budget(),
    );
}

# Effective per-port LU-path budget: --lun-path-budget or the default.
sub lun_path_budget { return $opts{'lun-path-budget'} || $DEFAULT_LUN_PATH_BUDGET }

# Print "LDEV <id>  lun <lun_id>  port <port>" for each orphan path entry.
sub print_orphan_lines {
    my ($list) = @_;
    printf "  LDEV %-6s lun %-14s port %s\n", $_->{ldev_id}, $_->{lun_id}, $_->{port}
        for @$list;
}

# Parse the storage's ldev_range (decimal OR hex, e.g. '256-511' or '0x100-0x1FF')
# into ([min,max], 'ok'). Returns (undef,'unset') when no range is configured and
# (undef,'unparseable') when set but malformed. ldev_range is the load-bearing
# safety fence for reconcile (it keeps the orphan unmap off other storeids' LUNs in
# a shared host group), so the status is used to refuse a fence-less --apply. Hex
# support mirrors the plugin's own ldev_range parser.
sub lun_path_ldev_range {
    my $r = $scfg->{ldev_range};
    return (undef, 'unset') unless defined $r && length $r;
    my $range = PVE::Storage::HitachiBlock::LunPaths->parse_ldev_range($r);
    return $range ? ($range, 'ok') : (undef, 'unparseable');
}

sub cmd_lun_paths {
    my $rep = build_lun_path_report();
    my $budget = lun_path_budget();

    return emit_json($rep) if $opts{json};

    chomp(my $hostname = `hostname -s 2>/dev/null` || '');
    print "=== LU-path / host-group accounting (this node) ===\n";
    printf "node: %s   storeid: %s\n", $hostname, $opts{storeid};
    printf "per-port LU-path budget: %d (warn at %d%%)   host-groups/port cap: %d\n\n",
        $budget, int($PVE::Storage::HitachiBlock::LunPaths::BUDGET_THRESHOLD * 100),
        $HOST_GROUPS_PER_PORT_CAP;

    for my $port (sort keys %{ $rep->{ports} }) {
        my $p = $rep->{ports}{$port};
        my $hgop = defined $p->{host_groups_on_port} ? $p->{host_groups_on_port} : '?';
        printf "  %-8s HG %-4s  LU-paths: %d / %d   host-groups-on-port: %s / %d\n",
            $port, (defined $p->{host_group_number} ? $p->{host_group_number} : '-'),
            $p->{lun_path_count}, $budget, $hgop, $HOST_GROUPS_PER_PORT_CAP;
    }
    printf "\ntotal mapped LU paths (this node): %d\n", $rep->{total_lun_paths};

    my @orphans = @{ $rep->{orphans} };
    printf "orphan LU paths (mapped here, LDEV not in registry): %d\n", scalar @orphans;
    print_orphan_lines(\@orphans);
    print "  (run 'reconcile-maps' to clean them up)\n" if @orphans;

    for my $ob (@{ $rep->{over_budget} }) {
        printf "WARNING: port %s at %d%% of LU-path budget (%d / %d)\n",
            $ob->{port}, $ob->{pct}, $ob->{count}, $ob->{budget};
    }
}

sub cmd_reconcile_maps {
    my $rep   = build_lun_path_report();
    my ($range, $range_status) = lun_path_ldev_range();
    my $apply = $opts{apply} ? 1 : 0;

    # The ldev_range fence is what keeps reconcile from unmapping another storeid's
    # live volumes that share this node's host group. Without a valid fence, refuse
    # to --apply (dry-run is still allowed for visibility). 'unset' and 'unparseable'
    # both fail closed.
    if ($apply && !$range) {
        die "reconcile-maps --apply refused: ldev_range is '$range_status' for"
            . " storeid '$opts{storeid}'. A valid ldev_range fence is required to"
            . " safely unmap (it keeps reconcile off other storeids' LUNs sharing"
            . " this host group). Configure ldev_range, or run without --apply for a"
            . " dry-run.\n";
    }

    my $res = PVE::Storage::HitachiBlock::LunPaths->reconcile(
        $client, $rep, ldev_range => $range, apply => $apply,
    );

    return emit_json($res) if $opts{json};

    if ($apply) {
        printf "Reconcile applied. Unmapped %d orphan LU path(s):\n", scalar @{ $res->{unmapped} };
        print_orphan_lines($res->{unmapped});
    } else {
        printf "DRY-RUN (use --apply to unmap). Would unmap %d orphan LU path(s):\n",
            scalar @{ $res->{would_unmap} };
        print_orphan_lines($res->{would_unmap});
        print "  none\n" unless @{ $res->{would_unmap} };
    }
    if (@{ $res->{skipped} }) {
        printf "Skipped %d:\n", scalar @{ $res->{skipped} };
        printf "  LDEV %-6s lun %-14s (%s)\n", $_->{ldev_id}, $_->{lun_id}, $_->{reason}
            for @{ $res->{skipped} };
    }
}

# Read-only host-side check: is this node ready to service SCSI-3 Persistent
# Reservations? Delegates entirely to Multipath->check_pr_ready() — zero
# duplicated detection logic here. Does NOT contact the array REST endpoint.
sub cmd_pr_check {
    chomp(my $hostname = `hostname -s 2>/dev/null` || '');

    my $mp = PVE::Storage::HitachiBlock::Multipath->new();
    my $result = $mp->check_pr_ready();

    if ($opts{json}) {
        return emit_json({
            node   => $hostname,
            ok     => $result->{ok},
            issues => $result->{issues},
        });
    }

    my $pr_helper_ok = !grep { /qemu-pr-helper/ } @{ $result->{issues} };
    my $resv_key_ok  = !grep { /reservation_key/ } @{ $result->{issues} };

    print "=== SCSI-3 PR readiness check ===\n";
    printf "node:                       %s\n", $hostname;
    printf "qemu-pr-helper socket:      %s\n", ($pr_helper_ok  ? 'active'         : 'NOT active');
    printf "multipath reservation_key:  %s\n", ($resv_key_ok   ? 'configured'     : 'not configured');
    printf "\nOverall: %s\n",                   ($result->{ok}  ? 'READY'          : 'NOT READY');

    if (@{ $result->{issues} }) {
        print "\nIssues:\n";
        printf "  - %s\n", $_ for @{ $result->{issues} };
    }
}

# ── Helpers ──

# Emit $data as canonical (sorted-key) JSON and return. Used by every subcommand
# for --json output so key order is deterministic across all commands.
sub emit_json {
    my ($data) = @_;
    print JSON->new->canonical->encode($data) . "\n";
    return;
}

# Best-effort plugin package version (dpkg), or 'unknown'.
sub _plugin_version {
    my $v = `dpkg-query -W -f='\${Version}' pve-storage-hitachiblock 2>/dev/null`;
    chomp $v if defined $v;
    return ($v && length $v) ? $v : 'unknown';
}

# Collapse an error/message to a single trimmed line for the report.
sub _trim {
    my ($s) = @_;
    return 'unknown error' unless defined $s && length $s;
    $s =~ s/\s+/ /g;
    $s =~ s/^\s+|\s+$//g;
    return length($s) > 200 ? substr($s, 0, 200) . '…' : $s;
}

sub resolve_volume_ldev {
    my ($volname) = @_;
    my $ldev_id = $config->lookup_ldev($volname);
    die "Volume '$volname' not found in registry for storeid '$opts{storeid}'\n"
        unless defined $ldev_id;
    return $ldev_id;
}

sub platform_port {
    my ($platform) = @_;
    my $defaults = PVE::Storage::HitachiBlock::Config->platform_defaults($platform);
    return $defaults->{port};
}

# Minimal /etc/pve/storage.cfg parser: returns the property hash for $storeid,
# or an empty hash if the file or section is absent (env/flags then apply).
sub read_storage_cfg {
    my ($storeid) = @_;

    my $file = $ENV{HITACHI_STORAGE_CFG} || '/etc/pve/storage.cfg';
    return {} unless -r $file;

    open(my $fh, '<', $file) or return {};
    my %cfg;
    my $in_section = 0;
    while (my $line = <$fh>) {
        if ($line =~ /^\s*hitachiblock:\s*(\S+)\s*$/) {
            $in_section = ($1 eq $storeid) ? 1 : 0;
            next;
        }
        # A new top-level section (type: id) ends ours.
        if ($line =~ /^\S+:\s*\S+/) {
            $in_section = 0 if $line !~ /^\s/;
        }
        next unless $in_section;
        if ($line =~ /^\s+(\S+)\s+(.+?)\s*$/) {
            $cfg{$1} = $2;
        }
    }
    close($fh);
    return \%cfg;
}

__END__

=head1 NAME

hitachiblock-repl - Hitachi Block Storage Replication CLI

=head1 SYNOPSIS

  hitachiblock-repl <command> --storeid <id> [options]

  Commands:
    list              List all replication pairs
    create-tc         Create TrueCopy pair (synchronous)
    create-ur         Create Universal Replicator pair (asynchronous)
    create-gad        Create Global-Active Device pair
    status            Show pair status
    split             Split (suspend) a pair
    resync            Resync a split pair
    delete            Delete a pair
    remote-list       List remote storage systems
    remote-add        Register a remote storage system
    orphans           Report LDEV/registry inconsistencies
    diagnostics       Read-only support bundle for bug reports (no mutations;
                      config with secrets redacted, REST health + response
                      times, registry/array drift, host device state,
                      LU-path/host-group counts)
    lun-paths         Per-node LU-path + host-group counts with headroom vs the
                      array caps, plus any orphan LU paths (read-only)
    reconcile-maps    Unmap orphan LU paths (LDEV no longer in the registry) in
                      this node's host groups; DRY-RUN unless --apply,
                      ldev_range-fenced
    pr-check          SCSI-3 PR readiness: verify qemu-pr-helper socket is
                      active and multipath reservation_key is configured on
                      this node; read-only, no array login required

  Options:
    --storeid         PVE storage ID (for credentials and config lookup)
    --pvol            Primary volume LDEV ID
    --svol            Secondary volume LDEV ID
    --volume          PVE volume name (e.g. vm-100-disk-1); resolved to --pvol
    --remote-storage-id  Remote storage serial/device ID
    --remote-ip       Remote storage management IP
    --copy-group      Copy group name
    --pair-name       Copy pair name
    --pair-id         Pair ID (for status/split/resync/delete)
    --copy-pace       Copy pace (1-15, default: 3)
    --journal-id      Journal ID (for UR)
    --quorum-disk-id  Quorum disk ID (for GAD)
    --consistency-group  Enable consistency group
    --auto-cleanup    (orphans) prune registry entries with no backing LDEV
    --apply           (reconcile-maps) actually unmap (default is dry-run)
    --lun-path-budget (lun-paths) per-port LU-path budget for headroom (default 2048)
    --mgmt-ip         Override Configuration Manager IP
    --storage-id      Override storage serial number
    --mgmt-port       Override API port
    --json            Output in JSON format
    --help            Show this help

  Connection parameters (mgmt IP, storage ID, port) are resolved in order:
    1. --mgmt-ip / --storage-id / --mgmt-port flags
    2. the hitachiblock:<storeid> section of /etc/pve/storage.cfg
    3. the HITACHI_MGMT_IP / HITACHI_STORAGE_ID / HITACHI_MGMT_PORT env vars

  Environment variables (legacy fallback):
    HITACHI_MGMT_IP       Management IP of Configuration Manager
    HITACHI_STORAGE_ID    Storage system serial number
    HITACHI_MGMT_PORT     API port (default: platform-specific)
    HITACHI_STORAGE_CFG   Override path to storage.cfg (default /etc/pve/storage.cfg)

=head1 EXAMPLES

  # List all replication pairs
  hitachiblock-repl list --storeid myarray

  # Create TrueCopy pair
  hitachiblock-repl create-tc --storeid myarray \
    --pvol 100 --svol 200 \
    --remote-storage-id 836000789012 \
    --copy-group PVE_TC_GROUP

  # Create GAD pair
  hitachiblock-repl create-gad --storeid myarray \
    --pvol 100 --svol 200 \
    --remote-storage-id 836000789012 \
    --copy-group PVE_GAD_GROUP \
    --quorum-disk-id 0

  # Check pair status
  hitachiblock-repl status --storeid myarray --pair-id TC_PAIR_001

  # Split pair for maintenance
  hitachiblock-repl split --storeid myarray --pair-id TC_PAIR_001

  # Resync after maintenance
  hitachiblock-repl resync --storeid myarray --pair-id TC_PAIR_001

  # Capture a read-only support bundle for a bug report (no secrets, no mutations)
  hitachiblock-repl diagnostics --storeid myarray
  hitachiblock-repl diagnostics --storeid myarray --json > bundle.json

  # Check SCSI-3 PR readiness on this node (no array login needed)
  hitachiblock-repl pr-check --storeid myarray
  hitachiblock-repl pr-check --storeid myarray --json

=cut
