[FEAT] Connettori discovery rete/cloud per-org → auto-popola Inventario (backend, mig 064)

Più connettori per azienda; ogni connettore ha una api_key dedicata (scope ingest:assets)
che l'agente esterno usa per mappare e auto-popolare NIS2.
- mig 064: discovery_connectors (per-org, multi) + discovery_runs (storico) + network_flows (ID.AM-03).
- DiscoveryConnectorController (org_admin): CRUD connettori + emissione/rotazione api_key
  (mostrata 1 volta) + dettaglio con ultimi run.
- ServicesController::ingestAssets esteso: accetta anche "flows" (upsert network_flows,
  dedup external_ref) e, se la chiave appartiene a un connettore, registra discovery_run
  + aggiorna last_run del connettore. Auto-scoring rilevanza NIS2 già in bulkUpsert.
- AssetController::bulkUpsert: ora salva anche "dependencies" → popola la Mappa Dipendenze.
- Router: /api/discovery-connectors (list/create/{id}/update/delete/rotateKey).
Smoke E2E (HTTP 201): 2 asset scorati + 1 flusso + run tracciato + dipendenze persistite.

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

application/controllers/ServicesController.php

@@ -2645,9 +2645,79 @@ class ServicesController extends BaseController
         }
         require_once __DIR__ . '/AssetController.php';
         $result = AssetController::bulkUpsert($this->currentOrgId, $items, $source, null);
+
+        // Flussi di rete (ID.AM-03) opzionali nello stesso payload
+        $flowsCount = 0;
+        if (!empty($body['flows']) && is_array($body['flows'])) {
+            $flowsCount = $this->upsertFlows($this->currentOrgId, $body['flows'], $source);
+        }
+        $result['flows_ingested'] = $flowsCount;
+
+        // Se la chiave appartiene a un connettore discovery → registra il run + last_run
+        $this->recordDiscoveryRun((int) ($this->apiKeyRecord['id'] ?? 0), $result, $flowsCount);
+
         $this->jsonSuccess($result, 'Asset importati', 201);
     }
 
+    /** Upsert idempotente dei flussi di rete (dedup su organization_id+external_ref). */
+    private function upsertFlows(int $orgId, array $flows, string $source): int
+    {
+        $n = 0;
+        foreach ($flows as $f) {
+            if (!is_array($f)) continue;
+            $src = trim((string) ($f['src'] ?? ''));
+            $dst = trim((string) ($f['dst'] ?? ''));
+            if ($src === '' || $dst === '') continue;
+            $port  = isset($f['port']) ? (int) $f['port'] : null;
+            $proto = isset($f['protocol']) ? substr((string) $f['protocol'], 0, 12) : null;
+            $dir   = in_array(($f['direction'] ?? ''), ['internal', 'inbound', 'outbound'], true) ? $f['direction'] : 'internal';
+            $ref   = isset($f['external_ref'])
+                ? substr((string) $f['external_ref'], 0, 190)
+                : substr($src . '>' . $dst . ':' . $port . '/' . $proto, 0, 190);
+            try {
+                Database::query(
+                    "INSERT INTO network_flows (organization_id, src, dst, port, protocol, direction, discovery_source, external_ref, last_seen_at)
+                     VALUES (?,?,?,?,?,?,?,?,NOW())
+                     ON DUPLICATE KEY UPDATE port=VALUES(port), protocol=VALUES(protocol), direction=VALUES(direction), last_seen_at=NOW()",
+                    [$orgId, substr($src, 0, 190), substr($dst, 0, 190), $port, $proto, $dir, substr($source, 0, 40), $ref]
+                );
+                $n++;
+            } catch (Throwable $e) {
+                error_log('[flows-ingest] ' . $e->getMessage());
+            }
+        }
+        return $n;
+    }
+
+    /** Registra il run del connettore discovery legato all'api_key, se esiste. */
+    private function recordDiscoveryRun(int $apiKeyId, array $result, int $flowsCount): void
+    {
+        if ($apiKeyId <= 0) return;
+        try {
+            $conn = Database::fetchOne(
+                'SELECT id FROM discovery_connectors WHERE api_key_id = ? AND organization_id = ?',
+                [$apiKeyId, $this->currentOrgId]
+            );
+            if (!$conn) return;
+            $assets = (int) ($result['imported'] ?? 0) + (int) ($result['updated'] ?? 0);
+            Database::insert('discovery_runs', [
+                'connector_id'      => $conn['id'],
+                'organization_id'   => $this->currentOrgId,
+                'finished_at'       => date('Y-m-d H:i:s'),
+                'status'            => 'ok',
+                'discovered_assets' => $assets,
+                'discovered_flows'  => $flowsCount,
+                'message'           => 'ingest via api',
+            ]);
+            Database::query(
+                'UPDATE discovery_connectors SET last_run_at=NOW(), last_status=?, last_discovered=?, last_message=? WHERE id=?',
+                ['ok', $assets, "asset {$assets}, flussi {$flowsCount}", $conn['id']]
+            );
+        } catch (Throwable $e) {
+            error_log('[discovery-run] ' . $e->getMessage());
+        }
+    }
+
     /**
      * GET /api/services/controls-monitoring
      * Auth: X-API-Key con scope read:compliance