/** * MiroFlow Bridge para Apache Superset * * Adaptador que conecta Superset com MiroFlow para análises científicas. * Fluxo: SQL (Superset) → MiroFlow (análise) → Enriquecimento de dados → KG (provenance) */ import type { Request, Response } from "express"; import { createNode } from "../graph/service"; import crypto from "crypto"; interface MiroFlowConfig { type?: "exploratory" | "forecast" | "anomaly" | "hypothesis" | "causal"; horizon?: number; // dias para forecast confidence_level?: number; // 0.90, 0.95, 0.99 methods?: string[]; // ["arima", "prophet", "statistical_test", etc] } interface SupersetDataset { data: Record[]; columns: string[]; index?: number[]; } interface MiroFlowResult { execution_id: string; agent: string; model: string; result: { summary: string; analysis_type: string; descriptive_stats?: Record; anomalies?: Array<{ index: number; score: number; value: any }>; forecast?: { values: number[]; lower_bound: number[]; upper_bound: number[] }; hypothesis_test?: { p_value: number; result: string }; confidence_intervals?: Record; }; metadata: { execution_time_ms: number; rows_analyzed: number; columns_used: string[]; }; } export interface EnrichedDataset extends SupersetDataset { _miroflow_analysis: string; _miroflow_type: string; _confidence_intervals?: Record; _anomaly_scores?: Record; _forecast?: { values: number[]; lower_bound: number[]; upper_bound: number[] }; _metadata: { analysis_executed_at: string; miroflow_execution_id: string; miroflow_model: string; execution_time_ms: number; }; } /** * MiroFlowBridge: Proxy entre Superset e MiroFlow * Enriquece datasets SQL com análises científicas */ export class MiroFlowBridge { private miroflowBaseUrl: string; private miroflowTimeout: number = 300_000; // 5 min constructor(miroflowBaseUrl: string = "http://localhost:8006") { this.miroflowBaseUrl = miroflowBaseUrl; } /** * Analisa dataset via MiroFlow */ async analyze( dataset: SupersetDataset, config: MiroFlowConfig, tenantId?: number, userId?: number ): Promise { const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), this.miroflowTimeout); try { const payload = { dataset: JSON.stringify(dataset), analysis_type: config.type || "exploratory", config, tenant_id: tenantId, user_id: userId, source: "superset_bridge", }; const response = await fetch(`${this.miroflowBaseUrl}/analyze`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(payload), signal: controller.signal, }); clearTimeout(timeout); if (!response.ok) { const err = await response.json().catch(() => ({ detail: response.statusText })); throw new Error(`MiroFlow error: ${err.detail || response.status}`); } return await response.json(); } catch (err: any) { clearTimeout(timeout); if (err.name === "AbortError") { throw new Error("MiroFlow analysis timeout (5 min)"); } throw err; } } /** * Enriquece dataset com resultados da análise */ enrich(dataset: SupersetDataset, result: MiroFlowResult): EnrichedDataset { const enriched: EnrichedDataset = { ...dataset, _miroflow_analysis: result.result.summary, _miroflow_type: result.result.analysis_type, _metadata: { analysis_executed_at: new Date().toISOString(), miroflow_execution_id: result.execution_id, miroflow_model: result.model, execution_time_ms: result.metadata.execution_time_ms, }, }; // Adicionar intervalos de confiança if (result.result.confidence_intervals) { enriched._confidence_intervals = result.result.confidence_intervals; } // Adicionar scores de anomalias (mapeados para índices) if (result.result.anomalies && result.result.anomalies.length > 0) { enriched._anomaly_scores = {}; result.result.anomalies.forEach((anom) => { enriched._anomaly_scores![anom.index] = anom.score; }); } // Adicionar forecast if (result.result.forecast) { enriched._forecast = result.result.forecast; } return enriched; } /** * Registra análise no Knowledge Graph para provenance */ async registerInKG( result: MiroFlowResult, dataset: SupersetDataset, config: MiroFlowConfig, req: Request ): Promise { try { const analysisHash = crypto .createHash("sha256") .update( JSON.stringify({ execution_id: result.execution_id, dataset_columns: dataset.columns, config, result: result.result, }) ) .digest("hex"); await createNode({ type: "miroflow_analysis", tenantId: (req.user as any)?.tenantId ?? null, data: { execution_id: result.execution_id, agent: result.agent, model: result.model, analysis_type: result.result.analysis_type, dataset_shape: `${dataset.data.length}x${dataset.columns.length}`, columns_analyzed: dataset.columns, config, summary: result.result.summary, execution_time_ms: result.metadata.execution_time_ms, analysisHash, immutable: true, source: "superset_bridge", registeredAt: new Date().toISOString(), }, }); } catch (kgErr: any) { console.error("[MiroFlow Bridge] Falha ao registrar análise no KG:", kgErr.message); // Não bloqueia a resposta — logging apenas } } /** * Handler Express para POST /api/superset/miroflow/analyze * Superset envia dataset SQL-resultante e config de análise */ async handleAnalyzeRequest(req: Request, res: Response): Promise { if (!req.isAuthenticated()) { res.status(401).json({ error: "Não autenticado" }); return; } try { const { dataset, config }: { dataset: SupersetDataset; config: MiroFlowConfig } = req.body; if (!dataset || !dataset.data || !dataset.columns) { res.status(400).json({ error: "Dataset inválido: data e columns obrigatórios" }); return; } // Executar análise no MiroFlow const result = await this.analyze(dataset, config, (req.user as any)?.tenantId, (req.user as any)?.id); // Enriquecer dataset const enriched = this.enrich(dataset, result); // Registrar no KG (assíncrono, não bloqueia) this.registerInKG(result, dataset, config, req).catch(() => {}); // Retornar dados enriquecidos res.json({ success: true, data: enriched, analysis: result.result, execution_id: result.execution_id, }); } catch (err: any) { console.error("[MiroFlow Bridge] Erro:", err.message); res.status(502).json({ error: err.message, source: "miroflow_bridge", }); } } } /** * Factory para criar instância do bridge */ export function createMiroFlowBridge(miroflowUrl?: string): MiroFlowBridge { return new MiroFlowBridge(miroflowUrl || process.env.MIROFLOW_URL || "http://localhost:8006"); }