fix(audit): AuditRebuildChain per-tenant rebuild (ADR-018, closes Stage 5 #1)

Task 4 плана 2026-05-29-audit-rebuild-per-tenant-fix.md.
Переписан AuditRebuildChain под per-tenant semantics ADR-018:

- Drop private COLUMN_CONFIG → читаем AuditChainConfig::TABLES + rowExpression()
- Для tenant-таблиц (partition_clause='PARTITION BY tenant_id'): отдельная
  iteration на каждый tenant. prev_hash scoped to last row with id<from-id
  AND tenant_id=X. Iterate rows of that tenant ordered by id, UPDATE +
  propagate prev_hash forward.
- Для BYPASSRLS-таблиц (auth_log/saas_admin_audit_log, partition_clause=''):
  одна global iteration без tenant scope.
- Информационный output показывает scope ('PARTITION BY tenant_id' или
  'global (within partition)').

NB: deviates from plan SQL (CTE с LAG+UPDATE) — той СтратегиЯ страдает
snapshot-isolation bug. PostgreSQL CTE executes on single snapshot, LAG
видит OLD stored log_hash, не propagate'ит новые хеши downstream. Chain
ломается через >1 row. Существующая PHP-loop архитектура iterating prev_hash
через переменную — корректна и сохранена. Tests подтверждают:

- AuditRebuildChainTest: 7/7 GREEN (включая 3 новых Task 3 теста +
  существующие 4 repair/balance/dry-run/reject — multi-tenant flipped
  RED→GREEN с post-rebuild PARTITION BY tenant_id matching).
- tests/Feature/Audit/: 16 tests / 13 passed / 0 failed / 2 errors / 1 skipped.
- 2 errors orthogonal к Task 4 (deal_id NOT NULL bug в AuditChainRace test +
  webhook_log undefined в OperationalFullFlow) — pre-existing baseline noise.

Ref: docs/adr/ADR-018-audit-chain-per-tenant-semantics.md
     docs/superpowers/plans/2026-05-29-audit-rebuild-per-tenant-fix.md

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

app/app/Console/Commands/AuditRebuildChain.php

@@ -4,27 +4,33 @@ declare(strict_types=1);
 
 namespace App\Console\Commands;
 
+use App\Services\Audit\AuditChainConfig;
 use Illuminate\Console\Command;
 use Illuminate\Support\Facades\DB;
 
 /**
  * Пересчитывает hash-цепь в указанной партиции аудит-таблицы начиная с заданного id.
  *
- * Используется для восстановления целостности после race condition в
- * audit_chain_hash() trigger (когда concurrent INSERT в одну партицию
- * создавали ветвление цепочки).
+ * 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) — единая цепочка в пределах партиции.
  *
- * Алгоритм (pure-SQL, Вариант А):
+ * Алгоритм (Вариант B — PHP-iteration с partition awareness):
  *   1. SET session_replication_role = replica отключает BEFORE-триггеры.
- *   2. Берём prev_hash строки с id < from-id (NULL для первой строки партиции).
- *   3. Для каждой строки от from-id вычисляем:
- *        new_hash = digest(COALESCE(prev_hash, x)  || ROW(...)::text::bytea, sha256)
- *      где ROW(...) имеет NULL::bytea на позиции log_hash.
- *   4. UPDATE партиции SET log_hash = new_hash WHERE id = cur_id.
- *   5. prev_hash = new_hash для следующей строки.
+ *   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.
  *
- * Ref: docs/superpowers/plans/2026-05-29-audit-chain-race-fix.md Task 3
- *      app/app/Console/Commands/VerifyAuditChains.php (TABLE_CONFIG)
+ * 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
 {
@@ -34,43 +40,7 @@ final class AuditRebuildChain extends Command
         {--dry-run : Показать сколько строк затронет, без UPDATE}
         {--force : Пропустить интерактивное подтверждение (для CI/тестов)}';
 
-    protected $description = 'Пересчитать hash-цепь в партиции аудит-таблицы начиная с указанного id';
-
-    /** @var array<string, list<string>> */
-    private const COLUMN_CONFIG = [
-        'activity_log' => [
-            'id', 'tenant_id', 'user_id', 'deal_id', 'event',
-            'old_value', 'new_value', 'context', 'ip_address', 'user_agent',
-            '__log_hash__', 'created_at',
-        ],
-        'balance_transactions' => [
-            'id', 'tenant_id', 'type', 'amount_rub', 'amount_leads',
-            'balance_rub_after', 'balance_leads_after', 'description',
-            'related_type', 'related_id', 'user_id', 'admin_user_id',
-            '__log_hash__', 'created_at',
-        ],
-        'auth_log' => [
-            'id', 'actor_type', 'tenant_id', 'user_id', 'saas_admin_user_id',
-            'email', 'event', 'ip_address', 'user_agent', 'failure_reason',
-            '__log_hash__', 'created_at',
-        ],
-        'tenant_operations_log' => [
-            'id', 'tenant_id', 'user_id', 'entity_type', 'entity_id',
-            'event', 'payload_before', 'payload_after', 'ip_address', 'user_agent',
-            '__log_hash__', 'created_at',
-        ],
-        'pd_processing_log' => [
-            'id', 'tenant_id', 'subject_type', 'subject_id', 'action',
-            'purpose', 'actor_tenant_user_id', 'actor_admin_user_id', 'ip_address',
-            '__log_hash__', 'created_at',
-        ],
-        'saas_admin_audit_log' => [
-            'id', 'admin_user_id', 'action', 'target_type', 'target_id',
-            'target_tenant_id', 'payload_before', 'payload_after', 'reason',
-            'ip_address', 'user_agent', 'requires_approval', 'approved_by', 'approved_at',
-            '__log_hash__', 'created_at',
-        ],
-    ];
+    protected $description = 'Пересчитать hash-цепь партиции аудит-таблицы (per-tenant per ADR-018)';
 
     public function handle(): int
     {
@@ -87,22 +57,26 @@ final class AuditRebuildChain extends Command
 
         $parentTable = (string) preg_replace('/_y\d{4}_m\d{2}$/', '', $partition);
 
-        if (! array_key_exists($parentTable, self::COLUMN_CONFIG)) {
+        if (! array_key_exists($parentTable, AuditChainConfig::TABLES)) {
             $this->error("Partition '{$partition}' не относится к поддерживаемым аудит-таблицам.");
-            $this->line('Поддерживаемые: '.implode(', ', array_keys(self::COLUMN_CONFIG)));
+            $this->line('Поддерживаемые: '.implode(', ', array_keys(AuditChainConfig::TABLES)));
 
             return self::FAILURE;
         }
 
-        $columns = self::COLUMN_CONFIG[$parentTable];
+        $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}");
 
@@ -119,7 +93,7 @@ final class AuditRebuildChain extends Command
         }
 
         if (! $force && ! $this->confirm(
-            "Пересчитать log_hash для {$count} строк в {$partition}? Это изменит данные в проде.",
+            "Пересчитать log_hash для {$count} строк в {$partition} (scope: {$scopeLabel})? Это изменит данные в проде.",
             false,
         )) {
             $this->warn('Отменено.');
@@ -127,54 +101,38 @@ final class AuditRebuildChain extends Command
             return self::FAILURE;
         }
 
-        $rowExpr = $this->buildRowExpression($columns);
-
         // Disable BEFORE triggers (audit_block_mutation blocks UPDATE).
         // Use session-level SET so it works even inside a wrapping transaction
         // (e.g. DatabaseTransactions in tests). Reset in finally.
         DB::connection('pgsql_supplier')->statement("SET session_replication_role = 'replica'");
 
         try {
-            $prevHashRow = DB::connection('pgsql_supplier')
-                ->table($partition)
-                ->where('id', '<', $fromId)
-                ->orderByDesc('id')
-                ->first(['log_hash']);
+            $totalUpdated = 0;
 
-            $prevHashHex = $this->bytesToHex($prevHashRow?->log_hash);
+            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();
 
-            $rows = DB::connection('pgsql_supplier')
-                ->table($partition)
-                ->where('id', '>=', $fromId)
-                ->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);
+                foreach ($tenantIds as $tenantId) {
+                    $totalUpdated += $this->rebuildScope(
+                        $partition,
+                        $rowExpr,
+                        $fromId,
+                        'tenant_id',
+                        (int) $tenantId,
+                    );
+                }
+            } else {
+                // BYPASSRLS-таблицы (auth_log, saas_admin_audit_log) — global scope.
+                $totalUpdated = $this->rebuildScope($partition, $rowExpr, $fromId, null, null);
             }
 
-            $this->info("Обновлено {$updated} строк в {$partition}.");
+            $this->info("Обновлено {$totalUpdated} строк в {$partition}.");
         } finally {
             DB::connection('pgsql_supplier')->statement("SET session_replication_role = 'origin'");
         }
@@ -184,6 +142,70 @@ final class AuditRebuildChain extends Command
         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.
@@ -200,19 +222,4 @@ final class AuditRebuildChain extends Command
 
         return '\\x'.bin2hex($bin);
     }
-
-    /**
-     * Build ROW(col1, ..., NULL::bytea, ..., coln) expression.
-     *
-     * @param  list<string>  $columns
-     */
-    private function buildRowExpression(array $columns): string
-    {
-        $parts = [];
-        foreach ($columns as $col) {
-            $parts[] = ($col === '__log_hash__') ? 'NULL::bytea' : "t.{$col}";
-        }
-
-        return 'ROW('.implode(', ', $parts).')';
-    }
 }