laravel-workkit/src/Services/BackupService.php

<?php

declare(strict_types=1);

namespace Blax\Workkit\Services;

use RuntimeException;

/**
 * Streaming backup pipeline. Dumps, compresses and encrypts in a single
 * shell pipe so PHP holds zero bytes of the database content in memory —
 * regardless of dump size. The original implementation ran each stage
 * separately and ran into "Allowed memory size exhausted" on
 * mid-three-digit-MB compressed dumps because Laravel's `Crypt::encryptString`
 * reads the whole file, base64-encodes (+33%) and JSON-envelopes it.
 *
 * Encryption: AES-256-CBC with PBKDF2 (600 000 iterations, random salt)
 * via the system `openssl` binary. The passphrase is derived from
 * APP_KEY (the `base64:` prefix is stripped, the remainder is used
 * verbatim — PBKDF2 stretches it into the key). A backup is restorable
 * only by a deployment that knows the same APP_KEY.
 *
 * The output file format is the standard `openssl enc -salt` format,
 * which means it's also restorable with vanilla openssl on any host:
 *   openssl enc -d -aes-256-cbc -pbkdf2 -iter 600000 -pass env:K \
 *     -in backup.sql.xz.enc | xz -d | mysql ...
 *
 * Required system binaries: `mysqldump`, `mysql`, `xz`, `openssl`,
 * `bash` (for `set -o pipefail`). All are standard on every reasonable
 * Linux server.
 *
 * IMPORTANT — single source of truth for the target server: every path
 * that touches MySQL (dump, import, drop, table/size queries) builds its
 * client invocation from the SAME `$cfg` array via {@see mysqlClient()}.
 * Never mix in Laravel's PDO connection for one operation and the raw
 * `$cfg` CLI for another — a `url` DSN, `unix_socket`, or read/write split
 * can make those two resolve to *different databases*, and a restore that
 * drops one schema while importing into another is exactly the
 * catastrophe these commands exist to prevent.
 */
class BackupService
{
    public const CIPHER = 'aes-256-cbc';

    public const PBKDF2_ITER = 600000;

    /** Canonical extension every backup this package writes carries. */
    public const BACKUP_EXT = '.sql.xz.enc';

    /** Suffix marking a pre-restore safety snapshot (excluded from auto-restore). */
    public const PRE_RESTORE_MARK = '.pre-restore';

    /** Sidecar metadata suffix written next to each backup (best-effort). */
    public const META_EXT = '.meta.json';

    /**
     * mysqldump → xz → openssl enc → $outPath. One pipeline, zero PHP
     * memory pressure. Pipefail propagates a failure in any stage out
     * to the caller as a non-zero exit code.
     *
     * $xzLevel defaults to 3 — empirically a good balance for SQL
     * (about 10× compression with a fraction of xz -9's time cost).
     */
    public static function dumpCompressEncrypt(array $cfg, string $outPath, int $xzLevel = 3): void
    {
        self::requireBinary('mysqldump');
        self::requireBinary('xz');
        self::requireBinary('openssl');
        self::requireBinary('bash');

        $level = max(0, min(9, $xzLevel));

        $mysqldump = 'mysqldump '
            .'--single-transaction --quick --skip-lock-tables '
            .'--user='.escapeshellarg((string) ($cfg['username'] ?? 'root')).' '
            .'--host='.escapeshellarg((string) ($cfg['host'] ?? 'localhost')).' '
            .'--port='.escapeshellarg((string) ($cfg['port'] ?? 3306)).' '
            .escapeshellarg((string) $cfg['database']);

        $xz = "xz -{$level} -T0";
        $openssl = 'openssl enc -'.self::CIPHER.' -pbkdf2 -iter '.self::PBKDF2_ITER.' -salt -pass env:WK_KEY';

        // bash -c with pipefail: any stage failing trips the whole
        // pipeline. Without pipefail, a mysqldump crash with xz still
        // running would leave the output file 0-byte and the exit
        // code 0 — silent corruption.
        $pipeline = "{$mysqldump} | {$xz} | {$openssl} > ".escapeshellarg($outPath);
        $cmd = '/bin/bash -c '.escapeshellarg('set -o pipefail; '.$pipeline);

        try {
            self::run($cmd, [
                'MYSQL_PWD' => (string) ($cfg['password'] ?? ''),
                'WK_KEY' => self::passphrase(),
            ]);
        } catch (RuntimeException $e) {
            // Don't leave a half-written encrypted file lying around —
            // it's neither valid plaintext (which we'd never want)
            // nor a complete backup, just confusing partial state.
            if (is_file($outPath)) {
                @unlink($outPath);
            }
            throw $e;
        }

        if (! file_exists($outPath) || filesize($outPath) === 0) {
            throw new RuntimeException("Backup pipeline produced no output at {$outPath}");
        }

        // Best-effort metadata sidecar — never fail a good backup over it.
        self::writeMeta($outPath, $cfg, $level);
    }

    /**
     * openssl dec → xz -d → mysql. Streams through the same pipeline
     * in reverse. Does no PHP-side decoding so the file size is bounded
     * only by disk I/O.
     *
     * A `SET FOREIGN_KEY_CHECKS=0;` line is prepended to the SQL stream so
     * that even a hand-rolled dump lacking mysqldump's own header imports
     * regardless of table order (forward FK references resolve once every
     * table exists). This is harmless for mysqldump output, which sets and
     * restores its own session vars, and the whole import is a single
     * short-lived mysql session so the relaxed checks never leak out.
     *
     * NOTE: this does NOT rescue a restore onto a *dirty* schema whose
     * leftover tables have incompatible column types — MySQL still raises
     * errno 150 / error 3780 at CREATE TABLE because the referenced table
     * already exists with the wrong type. For that case the target must be
     * empty first (see RestoreCommand --fresh). The openssl|xz stages stay
     * a real pipeline segment so `set -o pipefail` still surfaces a bad key
     * or corrupt file as a non-zero exit.
     */
    public static function decryptDecompressImport(string $inPath, array $cfg): void
    {
        self::requireBinary('mysql');
        self::requireBinary('xz');
        self::requireBinary('openssl');
        self::requireBinary('bash');

        self::assertLooksEncrypted($inPath);

        $openssl = self::opensslDecrypt($inPath);
        $mysql = self::mysqlClient($cfg);

        // The brace group's exit status is that of its last command (the
        // openssl|xz pipeline), so under `set -o pipefail` a decrypt/decompress
        // failure still propagates out as the head of the outer pipeline.
        $payload = "{ printf 'SET FOREIGN_KEY_CHECKS=0;\\n'; {$openssl} | xz -d -T0; }";
        $pipeline = "{$payload} | {$mysql}";
        $cmd = '/bin/bash -c '.escapeshellarg('set -o pipefail; '.$pipeline);

        self::run($cmd, [
            'MYSQL_PWD' => (string) ($cfg['password'] ?? ''),
            'WK_KEY' => self::passphrase(),
        ]);
    }

    /**
     * Drop every base table and view in the target database, then leave
     * the (empty) database in place. Built entirely from $cfg via the same
     * mysql client the import uses, and scoped to `DATABASE()` (the schema
     * the CLI connects to) so the wipe provably hits the same server +
     * schema the subsequent import writes to.
     *
     * Implemented as in-SQL prepared statements rather than DROP DATABASE
     * so it never needs CREATE privilege and preserves the database's
     * charset/collation and any grants attached to it.
     */
    public static function dropAllTables(array $cfg): void
    {
        self::requireBinary('mysql');
        self::requireBinary('bash');

        // Tables and views dropped in two single statements (GROUP_CONCAT
        // builds `a`,`b`,`c`); FK checks off so table order is irrelevant.
        $sql = <<<'SQL'
SET FOREIGN_KEY_CHECKS = 0;
SET SESSION group_concat_max_len = 1000000000;
SET @t := (SELECT GROUP_CONCAT(CONCAT('`', table_name, '`')) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_type = 'BASE TABLE');
SET @v := (SELECT GROUP_CONCAT(CONCAT('`', table_name, '`')) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_type = 'VIEW');
SET @dt := IF(@t IS NULL, 'SELECT 1', CONCAT('DROP TABLE IF EXISTS ', @t));
PREPARE s1 FROM @dt; EXECUTE s1; DEALLOCATE PREPARE s1;
SET @dv := IF(@v IS NULL, 'SELECT 1', CONCAT('DROP VIEW IF EXISTS ', @v));
PREPARE s2 FROM @dv; EXECUTE s2; DEALLOCATE PREPARE s2;
SET FOREIGN_KEY_CHECKS = 1;
SQL;

        $pipeline = 'printf %s '.escapeshellarg($sql).' | '.self::mysqlClient($cfg);
        $cmd = '/bin/bash -c '.escapeshellarg('set -o pipefail; '.$pipeline);

        self::run($cmd, ['MYSQL_PWD' => (string) ($cfg['password'] ?? '')]);
    }

    /**
     * Verify a backup without touching MySQL: decrypt + fully decompress and
     * inspect the tail. Returns a structured result rather than throwing so
     * callers (workkit:db:verify, the --fresh pre-drop gate) can branch on it.
     *
     * What it proves:
     *   - integrity_ok   : decrypts with the current APP_KEY AND is a valid,
     *                       non-truncated xz stream (xz -d fails otherwise).
     *   - looks_complete : the plaintext ends with mysqldump's
     *                       "-- Dump completed" marker — catches a dump that
     *                       was truncated AT SOURCE (mysqldump crashed mid-run)
     *                       yet compressed cleanly, which xz integrity can't.
     *   - checksum_matches: on-disk sha256 equals the value recorded in the
     *                       .meta.json sidecar (proves the file wasn't altered
     *                       on disk since it was written), or null if no sidecar.
     *
     * It does NOT prove the dump is restorable against a given schema — a
     * valid backup of the wrong/empty database still verifies "OK".
     *
     * @return array{file:string,bytes:int,integrity_ok:bool,looks_complete:bool,sha256:?string,recorded_sha256:?string,checksum_matches:?bool,message:string}
     */
    public static function verify(string $path): array
    {
        self::requireBinary('openssl');
        self::requireBinary('xz');
        self::requireBinary('bash');

        $result = [
            'file' => $path,
            'bytes' => is_file($path) ? (int) filesize($path) : 0,
            'integrity_ok' => false,
            'looks_complete' => false,
            'sha256' => null,
            'recorded_sha256' => null,
            'checksum_matches' => null,
            'message' => '',
        ];

        if (! is_file($path)) {
            $result['message'] = "Backup file not found: {$path}";

            return $result;
        }

        $head = (string) @file_get_contents($path, false, null, 0, 8);
        if ($head !== 'Salted__') {
            $result['message'] = "Not a streaming openssl backup (missing 'Salted__' header). "
                .'Legacy Crypt-envelope backups need a different restore path.';

            return $result;
        }

        // Decrypt + decompress in one streaming pass, keeping only the tail
        // for the completion-marker check. xz -d fails on truncation/bit-rot.
        $pipeline = self::opensslDecrypt($path).' | xz -d -T0 | tail -c 512';
        $cmd = '/bin/bash -c '.escapeshellarg('set -o pipefail; '.$pipeline);
        [$exit, $stdout, $stderr] = self::runResult($cmd, ['WK_KEY' => self::passphrase()]);

        if ($exit !== 0) {
            if (str_contains($stderr, 'bad decrypt') || str_contains($stderr, 'bad magic')) {
                $result['message'] = 'Decryption failed — APP_KEY likely does not match the host that produced this backup.';
            } else {
                // openssl can (rarely, ~1/256) yield valid PKCS#7 padding on a
                // wrong key and exit 0, leaving xz to choke on garbage — so a
                // post-decrypt xz failure is ambiguous between the two causes.
                $result['message'] = 'Decrypted but the stream is not valid xz — the backup is corrupt OR '
                    .'(rarely) the APP_KEY is wrong and the padding happened to validate. Verify APP_KEY first. '
                    .'stderr: '.trim($stderr);
            }

            return $result;
        }

        $result['integrity_ok'] = true;
        $result['looks_complete'] = str_contains($stdout, '-- Dump completed');

        // sha256 + sidecar comparison (best-effort; streams the file).
        $sha = @hash_file('sha256', $path);
        $result['sha256'] = $sha === false ? null : $sha;
        $meta = self::readMeta($path);
        if ($meta !== null && isset($meta['sha256']) && is_string($meta['sha256'])) {
            $result['recorded_sha256'] = $meta['sha256'];
            if ($result['sha256'] !== null) {
                $result['checksum_matches'] = hash_equals($meta['sha256'], $result['sha256']);
            }
        }

        if ($result['checksum_matches'] === false) {
            $result['message'] = 'CHECKSUM MISMATCH — the file changed on disk since it was written (integrity of the xz stream is still OK).';
        } elseif (! $result['looks_complete']) {
            $result['message'] = 'Integrity OK (decrypts + valid xz) but no mysqldump completion marker — '
                .'may be truncated at source or a non-mysqldump dump.';
        } else {
            $result['message'] = 'OK — decrypts, valid xz, mysqldump completion marker present'
                .($result['checksum_matches'] === true ? ', checksum matches sidecar.' : '.');
        }

        return $result;
    }

    /**
     * Count base tables + views in the target database, via the same mysql
     * client the import/drop use. Returns null if the query can't run (no
     * connection, database missing) so callers can treat it as "unknown".
     */
    public static function countTables(array $cfg): ?int
    {
        $val = self::scalarQuery(
            $cfg,
            'SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE()'
        );

        return $val === null ? null : (int) $val;
    }

    /**
     * Approximate on-disk size (bytes) of the target database from
     * information_schema. Null if unavailable. Used only for a best-effort
     * pre-snapshot disk-space sanity check.
     */
    public static function estimateDatabaseBytes(array $cfg): ?int
    {
        $val = self::scalarQuery(
            $cfg,
            'SELECT IFNULL(SUM(data_length + index_length), 0) FROM information_schema.tables WHERE table_schema = DATABASE()'
        );

        return $val === null ? null : (int) $val;
    }

    /**
     * Real backup files in the backup directory, newest first. Excludes
     * `.meta.json` sidecars but KEEPS custom `--out`-named backups (we only
     * deny the sidecar suffix rather than allow-listing one extension, so a
     * consumer's nightly.enc / db.bak is still found and pruned).
     *
     * @return string[] absolute paths, newest (by mtime) first
     */
    public static function backupFiles(): array
    {
        $base = self::backupDirectory();
        $files = glob($base.'/*') ?: [];
        $files = array_values(array_filter($files, function ($f) {
            return is_file($f) && ! str_ends_with($f, self::META_EXT);
        }));
        usort($files, fn ($a, $b) => filemtime($b) <=> filemtime($a));

        return $files;
    }

    /**
     * Like {@see backupFiles()} but also drops pre-restore safety snapshots,
     * so the "newest backup" auto-resolver never silently picks a snapshot of
     * a known-bad database taken moments before a wipe.
     *
     * @return string[]
     */
    public static function restoreCandidates(): array
    {
        return array_values(array_filter(
            self::backupFiles(),
            fn ($f) => ! str_contains(basename($f), self::PRE_RESTORE_MARK.'.')
        ));
    }

    /**
     * Derive the metadata sidecar path for a backup file. Always
     * backup → sidecar (never the reverse), so a backup with no sidecar is
     * still handled and a lone sidecar is recognisably an orphan.
     */
    public static function metaPathFor(string $backupPath): string
    {
        return $backupPath.self::META_EXT;
    }

    /**
     * Derive the openssl passphrase from APP_KEY. Laravel stores APP_KEY
     * as "base64:<random-bytes>"; we strip the prefix and feed the rest
     * straight to openssl, which runs PBKDF2 over it to get the AES key.
     * Same APP_KEY always derives the same key — restore is deterministic.
     */
    public static function passphrase(): string
    {
        $key = (string) config('app.key');
        if ($key === '') {
            throw new RuntimeException('APP_KEY is empty. Run `php artisan key:generate` before using the backup commands.');
        }
        if (str_starts_with($key, 'base64:')) {
            $key = substr($key, 7);
        }

        return $key;
    }

    /**
     * Path of the host's backup directory, created if missing. Defaults
     * to storage/backups; overridable via config('workkit.backup.path').
     *
     * Group-writable on purpose — the dir is often created once at deploy
     * time (as root or whoever ran the first artisan command) and then
     * written by www-data at runtime. If we can't make it writable for
     * the current user, we throw with the exact `chown`/`chmod` to run,
     * because the alternative is the bash redirect failing mid-pipeline
     * with `Permission denied` after mysqldump has already started.
     */
    public static function backupDirectory(): string
    {
        $path = config('workkit.backup.path') ?: storage_path('backups');

        if (! is_dir($path)) {
            // Suppress because a tight parent dir or umask can race here;
            // the is_dir() check below is the real gate.
            @mkdir($path, 0775, true);
            if (! is_dir($path)) {
                throw new RuntimeException("Failed to create backup directory: {$path}");
            }
        }

        // Best-effort widen — only succeeds when we own the dir, which is
        // exactly when the perms were already too narrow to begin with.
        @chmod($path, 0775);

        if (! is_writable($path)) {
            $owner = '?';
            $current = '?';
            if (function_exists('posix_geteuid') && function_exists('posix_getpwuid')) {
                $statOwner = @fileowner($path);
                if ($statOwner !== false) {
                    $owner = posix_getpwuid($statOwner)['name'] ?? (string) $statOwner;
                }
                $current = posix_getpwuid(posix_geteuid())['name'] ?? (string) posix_geteuid();
            }
            throw new RuntimeException(sprintf(
                "Backup directory not writable: %s\n"
                ."  owner=%s, current user=%s\n"
                .'  Fix as root:  chown -R %s %s && chmod 0775 %s',
                $path,
                $owner,
                $current,
                $current,
                $path,
                $path
            ));
        }

        return rtrim($path, '/');
    }

    /**
     * Bail loudly if a required system binary isn't on PATH. Done early
     * in each command so users get one clear message instead of a
     * cryptic exec failure halfway through.
     */
    public static function requireBinary(string $bin): void
    {
        $found = trim((string) @shell_exec('command -v '.escapeshellarg($bin)));
        if ($found === '') {
            throw new RuntimeException("Required binary `{$bin}` not found on PATH.");
        }
    }

    /**
     * Run a shell command and throw on non-zero exit, capturing stderr
     * for the error message.
     */
    public static function run(string $command, array $env = []): void
    {
        [$exit, , $stderr] = self::runResult($command, $env);

        if ($exit !== 0) {
            throw new RuntimeException(sprintf(
                "Command failed (exit %d):\n  %s\nstderr:\n%s",
                $exit,
                self::redactCommand($command),
                trim((string) $stderr) ?: '(empty)'
            ));
        }
    }

    /**
     * Run a shell command and return [exitCode, stdout, stderr] without
     * throwing. Env vars are passed via proc_open's env arg — scoped to the
     * child process, never visible in the host's `ps` listing.
     *
     * stdout is read fully; in this package the only stdout producers are
     * tiny (a `tail -c 512`, a scalar query, a row of SHOW output), so this
     * stays well within the zero-DB-bytes-in-PHP invariant. stdout and stderr
     * are drained concurrently (non-blocking + stream_select) so a child that
     * fills both pipe buffers at once can never deadlock against a serial read.
     */
    public static function runResult(string $command, array $env = []): array
    {
        $descriptors = [
            0 => ['pipe', 'r'],
            1 => ['pipe', 'w'],
            2 => ['pipe', 'w'],
        ];

        $envForProc = $env === [] ? null : array_merge($_ENV, $env);
        $proc = proc_open($command, $descriptors, $pipes, null, $envForProc);
        if (! is_resource($proc)) {
            throw new RuntimeException("Failed to start process: {$command}");
        }

        fclose($pipes[0]);
        stream_set_blocking($pipes[1], false);
        stream_set_blocking($pipes[2], false);

        $stdout = '';
        $stderr = '';
        $open = [1 => $pipes[1], 2 => $pipes[2]];

        while ($open !== []) {
            $read = array_values($open);
            $write = null;
            $except = null;
            if (@stream_select($read, $write, $except, null) === false) {
                break;
            }
            foreach ($read as $stream) {
                $chunk = fread($stream, 8192);
                if (($chunk === '' || $chunk === false) && feof($stream)) {
                    $key = array_search($stream, $open, true);
                    if ($key !== false) {
                        fclose($stream);
                        unset($open[$key]);
                    }

                    continue;
                }
                if ($chunk === false) {
                    continue;
                }
                if ($stream === $pipes[1]) {
                    $stdout .= $chunk;
                } else {
                    $stderr .= $chunk;
                }
            }
        }

        // Close anything stream_select bailed on, then reap the process.
        foreach ($open as $stream) {
            fclose($stream);
        }
        $exit = proc_close($proc);

        return [$exit, $stdout, $stderr];
    }

    /** Human-readable byte size, e.g. 13631488 → "13.00 MB". */
    public static function humanBytes(int $bytes): string
    {
        $units = ['B', 'KB', 'MB', 'GB', 'TB'];
        $i = 0;
        $value = (float) $bytes;
        while ($value >= 1024 && $i < count($units) - 1) {
            $value /= 1024;
            $i++;
        }

        return sprintf('%.2f %s', $value, $units[$i]);
    }

    // ---- internals ---------------------------------------------------------

    /**
     * Build the mysql client invocation from $cfg. Single source of truth
     * for the target server across import, drop and queries.
     */
    private static function mysqlClient(array $cfg): string
    {
        return 'mysql '
            .'--user='.escapeshellarg((string) ($cfg['username'] ?? 'root')).' '
            .'--host='.escapeshellarg((string) ($cfg['host'] ?? 'localhost')).' '
            .'--port='.escapeshellarg((string) ($cfg['port'] ?? 3306)).' '
            .escapeshellarg((string) $cfg['database']);
    }

    /** The openssl decrypt stage, reading from $inPath. */
    private static function opensslDecrypt(string $inPath): string
    {
        return 'openssl enc -d -'.self::CIPHER
            .' -pbkdf2 -iter '.self::PBKDF2_ITER
            .' -pass env:WK_KEY '
            .'-in '.escapeshellarg($inPath);
    }

    /** Sanity-check the openssl `-salt` magic header. */
    private static function assertLooksEncrypted(string $inPath): void
    {
        $head = (string) @file_get_contents($inPath, false, null, 0, 8);
        if ($head !== 'Salted__') {
            throw new RuntimeException(
                "File at {$inPath} doesn't look like a streaming openssl backup. "
                .'Legacy backups (made with the pre-streaming Crypt envelope) need '
                .'a separate restore path; see workkit:db:restore-legacy or restore '
                .'manually with `Crypt::decryptString` after raising memory_limit.'
            );
        }
    }

    /**
     * Run a single-row, single-column query against $cfg and return the
     * trimmed scalar, or null if the query fails (no connection / missing db).
     */
    private static function scalarQuery(array $cfg, string $sql): ?string
    {
        self::requireBinary('mysql');
        self::requireBinary('bash');

        $pipeline = self::mysqlClient($cfg).' -N -e '.escapeshellarg($sql);
        $cmd = '/bin/bash -c '.escapeshellarg('set -o pipefail; '.$pipeline);
        [$exit, $stdout] = self::runResult($cmd, ['MYSQL_PWD' => (string) ($cfg['password'] ?? '')]);

        if ($exit !== 0) {
            return null;
        }

        return trim($stdout);
    }

    /**
     * Write a best-effort metadata sidecar next to a freshly-created backup.
     * Never throws: a backup is valid with or without its sidecar.
     */
    private static function writeMeta(string $backupPath, array $cfg, int $xzLevel): void
    {
        try {
            $sha = @hash_file('sha256', $backupPath);
            $meta = [
                'file' => basename($backupPath),
                'database' => $cfg['database'] ?? null,
                'host' => $cfg['host'] ?? null,
                'created_at' => date('c'),
                'bytes' => (int) @filesize($backupPath),
                'sha256' => $sha === false ? null : $sha,
                'cipher' => self::CIPHER,
                'pbkdf2_iter' => self::PBKDF2_ITER,
                'xz_level' => $xzLevel,
            ];
            @file_put_contents(
                self::metaPathFor($backupPath),
                json_encode($meta, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES)
            );
        } catch (\Throwable $e) {
            // Sidecar is a nicety, not a requirement.
        }
    }

    /** Read a backup's metadata sidecar, or null if absent/unreadable. */
    private static function readMeta(string $backupPath): ?array
    {
        $metaPath = self::metaPathFor($backupPath);
        if (! is_file($metaPath)) {
            return null;
        }
        $decoded = json_decode((string) @file_get_contents($metaPath), true);

        return is_array($decoded) ? $decoded : null;
    }

    /**
     * Hide credential-looking flags in error messages so we don't dump
     * passwords to logs. The streaming pipeline doesn't put creds on
     * the CLI (everything goes via env vars), but defence in depth.
     */
    private static function redactCommand(string $command): string
    {
        return preg_replace('/(--password=)[^\s]+/', '$1***', $command);
    }
}