portal/app/app/Console/Commands/AuditRebuildChain.php

<?php

declare(strict_types=1);

namespace App\Console\Commands;

use App\Services\Audit\AuditChainConfig;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;

/**
 * Пересчитывает hash-цепь в указанной партиции аудит-таблицы начиная с заданного id.
 *
 * ADR-018: воспроизводит per-tenant scope триггера audit_chain_hash() (через RLS).
 * Для tenant-таблиц (activity_log/balance_transactions/tenant_operations_log/
 * pd_processing_log) — отдельная цепочка на каждый tenant. Для BYPASSRLS-таблиц
 * (auth_log/saas_admin_audit_log) — единая цепочка в пределах партиции.
 *
 * Алгоритм (Вариант B — PHP-iteration с partition awareness):
 *   1. SET session_replication_role = replica отключает BEFORE-триггеры.
 *   2. Determine partition_clause из AuditChainConfig::TABLES[parent_table].
 *   3. Для per-tenant таблиц: получить distinct tenant_ids в range, для каждого:
 *        - prev_hash = log_hash of last row with id<from-id AND tenant_id=X
 *        - iterate rows ordered by id, UPDATE + propagate prev_hash forward
 *      Для BYPASSRLS-таблиц: одна iteration без tenant scope.
 *   4. Возвращаем session_replication_role = origin.
 *
 * NB: row-by-row PHP loop сохранён намеренно (вариант с одиночным CTE и
 * LAG страдает snapshot-isolation bug — downstream rows используют OLD stored
 * prev_hash вместо новых хешей текущего UPDATE'а; chain ломается через >1 row).
 *
 * Ref: docs/adr/ADR-018-audit-chain-per-tenant-semantics.md
 *      docs/superpowers/plans/2026-05-29-audit-rebuild-per-tenant-fix.md
 */
final class AuditRebuildChain extends Command
{
    protected $signature = 'audit:rebuild-chain
        {--partition= : Имя партиции, например activity_log_y2026_m05}
        {--from-id= : ID с которого начать пересчёт (включительно)}
        {--dry-run : Показать сколько строк затронет, без UPDATE}
        {--force : Пропустить интерактивное подтверждение (для CI/тестов)}';

    protected $description = 'Пересчитать hash-цепь партиции аудит-таблицы (per-tenant per ADR-018)';

    public function handle(): int
    {
        $partition = (string) $this->option('partition');
        $fromId = (int) $this->option('from-id');
        $dryRun = (bool) $this->option('dry-run');
        $force = (bool) $this->option('force');

        if ($partition === '' || $fromId <= 0) {
            $this->error('--partition и --from-id обязательны');

            return self::FAILURE;
        }

        $parentTable = (string) preg_replace('/_y\d{4}_m\d{2}$/', '', $partition);

        if (! array_key_exists($parentTable, AuditChainConfig::TABLES)) {
            $this->error("Partition '{$partition}' не относится к поддерживаемым аудит-таблицам.");
            $this->line('Поддерживаемые: '.implode(', ', array_keys(AuditChainConfig::TABLES)));

            return self::FAILURE;
        }

        $partitionClause = AuditChainConfig::TABLES[$parentTable]['partition'];
        $rowExpr = AuditChainConfig::rowExpression($parentTable);

        $count = DB::connection('pgsql_supplier')
            ->table($partition)
            ->where('id', '>=', $fromId)
            ->count();

        $scopeLabel = $partitionClause !== '' ? $partitionClause : 'global (within partition)';

        $this->info("Партиция : {$partition}");
        $this->info("Родитель : {$parentTable}");
        $this->info("Scope    : {$scopeLabel}");
        $this->info("От id    : {$fromId}");
        $this->info("Строк    : {$count}");

        if ($count === 0) {
            $this->warn('Нет строк с id >= '.$fromId.'. Пересчёт не нужен.');

            return self::SUCCESS;
        }

        if ($dryRun) {
            $this->warn('--dry-run: UPDATE не выполнен.');

            return self::SUCCESS;
        }

        if (! $force && ! $this->confirm(
            "Пересчитать log_hash для {$count} строк в {$partition} (scope: {$scopeLabel})? Это изменит данные в проде.",
            false,
        )) {
            $this->warn('Отменено.');

            return self::FAILURE;
        }

        // Пересчёт цепочки = UPDATE по append-only таблицам. Вместо superuser-параметра
        // session_replication_role (недоступен в Managed PG — Путь А) используем метку
        // app.audit_rebuild='on', которую чтит триггер audit_block_mutation. SET LOCAL
        // внутри транзакции — Odyssey-safe: метка живёт ровно на время пересчёта и
        // сбрасывается на commit. В тестах (DatabaseTransactions + SharesSupplierPdo)
        // это savepoint внутри внешней транзакции — метка применяется ко всем UPDATE.
        $totalUpdated = 0;
        DB::connection('pgsql_supplier')->transaction(function () use ($partition, $partitionClause, $rowExpr, $fromId, &$totalUpdated) {
            DB::connection('pgsql_supplier')->statement("SET LOCAL app.audit_rebuild = 'on'");

            if ($partitionClause === 'PARTITION BY tenant_id') {
                // Per-tenant rebuild — separate scope iteration per tenant.
                $tenantIds = DB::connection('pgsql_supplier')
                    ->table($partition)
                    ->where('id', '>=', $fromId)
                    ->distinct()
                    ->pluck('tenant_id')
                    ->all();

                foreach ($tenantIds as $tenantId) {
                    $totalUpdated += $this->rebuildScope(
                        $partition,
                        $rowExpr,
                        $fromId,
                        'tenant_id',
                        (int) $tenantId,
                    );
                }
            } else {
                // global scope (auth_log, saas_admin_audit_log).
                $totalUpdated = $this->rebuildScope($partition, $rowExpr, $fromId, null, null);
            }
        });

        $this->info("Обновлено {$totalUpdated} строк в {$partition}.");

        $this->info('Готово. Запустите audit:verify-chains для проверки целостности.');

        return self::SUCCESS;
    }

    /**
     * Пересчитывает chain для одного scope (tenant или global).
     *
     * Iterative PHP loop: prev_hash propagate'ится forward через каждый row,
     * UPDATE применяется immediately чтобы snapshot для следующей iteration
     * был свежий (default PG READ COMMITTED — own writes visible immediately).
     *
     * @param  string|null  $tenantColumn  'tenant_id' для per-tenant scope, null для global
     * @param  int|null  $tenantValue  значение tenant_id для этого scope (если применимо)
     */
    private function rebuildScope(
        string $partition,
        string $rowExpr,
        int $fromId,
        ?string $tenantColumn,
        ?int $tenantValue,
    ): int {
        // Find prev_hash (last row before fromId within scope).
        $prevQuery = DB::connection('pgsql_supplier')
            ->table($partition)
            ->where('id', '<', $fromId);
        if ($tenantColumn !== null) {
            $prevQuery->where($tenantColumn, $tenantValue);
        }
        $prevHashRow = $prevQuery->orderByDesc('id')->first(['log_hash']);
        $prevHashHex = $this->bytesToHex($prevHashRow?->log_hash);

        // Get rows to rebuild ordered by id.
        $rowsQuery = DB::connection('pgsql_supplier')
            ->table($partition)
            ->where('id', '>=', $fromId);
        if ($tenantColumn !== null) {
            $rowsQuery->where($tenantColumn, $tenantValue);
        }
        $rows = $rowsQuery->orderBy('id')->get(['id']);

        $updated = 0;
        foreach ($rows as $row) {
            $prevHashExpr = $prevHashHex !== null
                ? "'{$prevHashHex}'::bytea"
                : "''::bytea";

            $sql = "
                UPDATE {$partition}
                SET log_hash = (
                    SELECT digest(
                        COALESCE({$prevHashExpr}, ''::bytea)
                        || (SELECT {$rowExpr}::text::bytea FROM {$partition} t WHERE t.id = ?)
                        , 'sha256'
                    )
                )
                WHERE id = ?
                RETURNING log_hash
            ";

            $result = DB::connection('pgsql_supplier')->selectOne($sql, [$row->id, $row->id]);
            $updated++;

            $prevHashHex = $this->bytesToHex($result?->log_hash);
        }

        return $updated;
    }

    /**
     * Convert a BYTEA value (PHP resource or string) to hex literal for SQL.
     * PostgreSQL PDO driver returns BYTEA as a PHP stream resource.
     */
    private function bytesToHex(mixed $value): ?string
    {
        if ($value === null) {
            return null;
        }
        $bin = is_resource($value) ? stream_get_contents($value) : (string) $value;
        if ($bin === '' || $bin === false) {
            return null;
        }

        return '\\x'.bin2hex($bin);
    }
}