nis2-agile/application/cli/db_integrity_probe.php

<?php
/**
 * db_integrity_probe.php — Sonda READ-ONLY di integrità referenziale del DB.
 *
 * Diagnostico: SOLO SELECT + information_schema. NON modifica nulla.
 * Estende l'audit della migrazione 039 alle tabelle aggiunte dopo (040-057).
 *
 * Uso (dal container app, connessione TLS via config app):
 *   docker exec nis2-app php /var/www/nis2-agile/application/cli/db_integrity_probe.php
 *
 * Output: report testuale a sezioni + riepilogo finale. Exit 0 sempre (è una sonda).
 */
if (PHP_SAPI !== 'cli') { http_response_code(403); exit("Solo da CLI.\n"); }
require_once __DIR__ . '/../config/env.php';
require_once __DIR__ . '/../config/database.php';

$pdo = Database::getInstance();
$db  = $pdo->query('SELECT DATABASE()')->fetchColumn();

function q(PDO $pdo, string $sql, array $p = []): array {
    $st = $pdo->prepare($sql); $st->execute($p); return $st->fetchAll(PDO::FETCH_ASSOC);
}
function h(string $t): void { echo "\n=== $t ===\n"; }

echo "SONDA INTEGRITÀ DB — schema: $db\n";
echo "Modalità: READ-ONLY (information_schema + SELECT). Nessuna scrittura.\n";

$issues = 0;

/* 1) Tabelle senza PRIMARY KEY ----------------------------------------- */
h('1) Tabelle SENZA PRIMARY KEY');
$noPk = q($pdo, "
  SELECT t.TABLE_NAME
  FROM information_schema.TABLES t
  LEFT JOIN information_schema.TABLE_CONSTRAINTS c
    ON c.TABLE_SCHEMA=t.TABLE_SCHEMA AND c.TABLE_NAME=t.TABLE_NAME AND c.CONSTRAINT_TYPE='PRIMARY KEY'
  WHERE t.TABLE_SCHEMA=? AND t.TABLE_TYPE='BASE TABLE' AND c.CONSTRAINT_NAME IS NULL
  ORDER BY t.TABLE_NAME", [$db]);
if (!$noPk) echo "  OK: tutte le tabelle hanno PK.\n";
else { foreach ($noPk as $r) echo "  [!] {$r['TABLE_NAME']}\n"; $issues += count($noPk); }

/* 2) Tabelle non-InnoDB (niente FK) ------------------------------------ */
h('2) Tabelle con engine != InnoDB (FK non supportate)');
$eng = q($pdo, "
  SELECT TABLE_NAME, ENGINE FROM information_schema.TABLES
  WHERE TABLE_SCHEMA=? AND TABLE_TYPE='BASE TABLE' AND ENGINE<>'InnoDB' ORDER BY TABLE_NAME", [$db]);
if (!$eng) echo "  OK: tutte InnoDB.\n";
else { foreach ($eng as $r) echo "  [!] {$r['TABLE_NAME']} = {$r['ENGINE']}\n"; $issues += count($eng); }

/* 3) Conteggio chiavi: PK / UNIQUE / FK per tabella -------------------- */
h('3) Tabelle senza alcun indice UNIQUE oltre alla PK (informativo)');
$rows = q($pdo, "
  SELECT t.TABLE_NAME,
    SUM(CASE WHEN s.NON_UNIQUE=0 AND s.INDEX_NAME<>'PRIMARY' THEN 1 ELSE 0 END) AS uniq_idx
  FROM information_schema.TABLES t
  LEFT JOIN information_schema.STATISTICS s
    ON s.TABLE_SCHEMA=t.TABLE_SCHEMA AND s.TABLE_NAME=t.TABLE_NAME
  WHERE t.TABLE_SCHEMA=? AND t.TABLE_TYPE='BASE TABLE'
  GROUP BY t.TABLE_NAME HAVING uniq_idx=0 ORDER BY t.TABLE_NAME", [$db]);
echo '  ' . count($rows) . " tabelle senza UNIQUE secondario (normale per tabelle figlie/log).\n";

/* 4) Colonne 'code'/'token'/'*_hash' candidate a UNIQUE senza unico ----- */
h('4) Colonne chiave-naturale (code/token/hash/slug/uuid) SENZA UNIQUE');
$cands = q($pdo, "
  SELECT c.TABLE_NAME, c.COLUMN_NAME
  FROM information_schema.COLUMNS c
  WHERE c.TABLE_SCHEMA=?
    AND (c.COLUMN_NAME IN ('code','token','slug','uuid','key_hash','token_hash','qdrant_doc_uuid')
         OR c.COLUMN_NAME LIKE '%_code' OR c.COLUMN_NAME LIKE '%_token' OR c.COLUMN_NAME LIKE '%_uuid')
    AND NOT EXISTS (
      SELECT 1 FROM information_schema.STATISTICS s
      WHERE s.TABLE_SCHEMA=c.TABLE_SCHEMA AND s.TABLE_NAME=c.TABLE_NAME
        AND s.COLUMN_NAME=c.COLUMN_NAME AND s.NON_UNIQUE=0)
  ORDER BY c.TABLE_NAME, c.COLUMN_NAME", [$db]);
if (!$cands) echo "  OK: nessuna colonna chiave-naturale priva di UNIQUE.\n";
else { foreach ($cands as $r) echo "  [?] {$r['TABLE_NAME']}.{$r['COLUMN_NAME']} (valutare se è chiave naturale)\n"; }
echo "  NB: molti '%_code' sono riferimenti (es. ref_code, control_code) e NON vanno resi unici.\n";

/* 5) Verifica mirata: duplicati su (organization_id, code) per i moduli a codice progressivo */
h('5) Duplicati su (organization_id, code) — pre-check sicurezza UNIQUE (mig.058)');
foreach (['internal_audits','management_reviews','periodic_controls'] as $tbl) {
    $exists = q($pdo, "SELECT 1 FROM information_schema.TABLES WHERE TABLE_SCHEMA=? AND TABLE_NAME=?", [$db,$tbl]);
    if (!$exists) { echo "  - $tbl: assente\n"; continue; }
    $dup = q($pdo, "SELECT organization_id, code, COUNT(*) n FROM `$tbl`
                    WHERE code IS NOT NULL GROUP BY organization_id, code HAVING n>1");
    $uq = q($pdo, "SELECT INDEX_NAME FROM information_schema.STATISTICS
                   WHERE TABLE_SCHEMA=? AND TABLE_NAME=? AND NON_UNIQUE=0 AND INDEX_NAME<>'PRIMARY'
                   GROUP BY INDEX_NAME", [$db,$tbl]);
    $hasUq = $uq ? implode(',', array_column($uq,'INDEX_NAME')) : '—';
    if ($dup) { echo "  [!] $tbl: " . count($dup) . " coppie (org,code) DUPLICATE → UNIQUE da bonificare prima! UNIQUE attuali: $hasUq\n"; $issues += count($dup); }
    else      { echo "  OK $tbl: 0 duplicati (org,code) → UNIQUE applicabile in sicurezza. UNIQUE attuali: $hasUq\n"; }
}

/* 6) Colonne *_id senza FK (candidate) -------------------------------- */
h('6) Colonne *_id SENZA FOREIGN KEY (candidate / da valutare)');
$nofk = q($pdo, "
  SELECT c.TABLE_NAME, c.COLUMN_NAME
  FROM information_schema.COLUMNS c
  WHERE c.TABLE_SCHEMA=? AND (c.COLUMN_NAME LIKE '%\\_id' OR c.COLUMN_NAME='id' AND 1=0)
    AND c.COLUMN_NAME NOT IN ('id')
    AND NOT EXISTS (
      SELECT 1 FROM information_schema.KEY_COLUMN_USAGE k
      WHERE k.TABLE_SCHEMA=c.TABLE_SCHEMA AND k.TABLE_NAME=c.TABLE_NAME
        AND k.COLUMN_NAME=c.COLUMN_NAME AND k.REFERENCED_TABLE_NAME IS NOT NULL)
  ORDER BY c.TABLE_NAME, c.COLUMN_NAME", [$db]);
if (!$nofk) echo "  OK: ogni colonna *_id ha una FK.\n";
else {
  echo "  " . count($nofk) . " colonne *_id senza FK (alcune legittime: polimorfiche object_id, sso_identity_id cross-DB, sentinel 0):\n";
  foreach ($nofk as $r) echo "    - {$r['TABLE_NAME']}.{$r['COLUMN_NAME']}\n";
}

/* 7) Orphan check su organization_id senza FK (integrità multi-tenant) */
h('7) Righe ORFANE su organization_id privo di FK (integrità multi-tenant)');
$orgCols = q($pdo, "
  SELECT c.TABLE_NAME
  FROM information_schema.COLUMNS c
  WHERE c.TABLE_SCHEMA=? AND c.COLUMN_NAME='organization_id'
    AND NOT EXISTS (
      SELECT 1 FROM information_schema.KEY_COLUMN_USAGE k
      WHERE k.TABLE_SCHEMA=c.TABLE_SCHEMA AND k.TABLE_NAME=c.TABLE_NAME
        AND k.COLUMN_NAME='organization_id' AND k.REFERENCED_TABLE_NAME='organizations')
  ORDER BY c.TABLE_NAME", [$db]);
// Pattern "by design": FK volutamente assente (trail immutabile / sentinel). NON bloccanti.
$byDesign = ['audit_logs' => 'audit trail immutabile (mig.006), deve sopravvivere alla cancellazione org'];
if (!$orgCols) echo "  OK: ogni organization_id ha FK verso organizations.\n";
else foreach ($orgCols as $r) {
    $t = $r['TABLE_NAME'];
    $o = q($pdo, "SELECT COUNT(*) n FROM `$t` x
                  WHERE x.organization_id IS NOT NULL AND x.organization_id<>0
                    AND NOT EXISTS (SELECT 1 FROM organizations o WHERE o.id=x.organization_id)");
    $n = (int)($o[0]['n'] ?? 0);
    $zero = (int)(q($pdo, "SELECT COUNT(*) n FROM `$t` WHERE organization_id=0")[0]['n'] ?? 0);
    $msg = "  $t: orfani=$n" . ($zero ? " (+$zero righe org_id=0 sentinel)" : '') . " — FK assente";
    if (isset($byDesign[$t])) { echo "  [by-design]$msg ({$byDesign[$t]})\n"; }
    elseif ($n>0) { echo "  [!]$msg\n"; $issues += $n; }
    else echo "$msg\n";
}

/* 8) Riepilogo FK totali ---------------------------------------------- */
h('8) Riepilogo');
$fkCount = (int)(q($pdo, "SELECT COUNT(*) n FROM information_schema.TABLE_CONSTRAINTS
                          WHERE TABLE_SCHEMA=? AND CONSTRAINT_TYPE='FOREIGN KEY'", [$db])[0]['n'] ?? 0);
$tblCount = (int)(q($pdo, "SELECT COUNT(*) n FROM information_schema.TABLES
                           WHERE TABLE_SCHEMA=? AND TABLE_TYPE='BASE TABLE'", [$db])[0]['n'] ?? 0);
echo "  Tabelle: $tblCount | Foreign key totali: $fkCount\n";
echo "  Problemi BLOCCANTI rilevati (PK mancanti / orfani reali / duplicati su code): $issues\n";
echo ($issues===0 ? "  ESITO: nessun problema bloccante.\n" : "  ESITO: rivedere le voci [!] sopra.\n");
exit(0);