import 'dart:convert'; import 'package:chain_client_sdk/chain_client_sdk.dart' as chain; import 'hub_endpoint.dart'; import 'models.dart'; import 'repository.dart'; /// Live `EvaluationRepository` backed by a Ch∆In hub over gRPC. /// /// Reads flow.completed events from the hub's audit log, parses /// each event's `detail` JSON as a Recl∆Im-shaped output bag /// (see flows/durchstich-gewo-14.yaml), and converts the bag to /// an Evaluation the UI can render. Events from non-Recl∆Im /// flows are silently skipped (detected by absence of the /// `norm` + `skm` keys). /// /// Sort order matches MockRepository: highest €/Jahr first. /// If the hub has zero matching flow.completed events the list /// returns empty — the UI shows a "noch keine Auswertungen, ' /// `chain run flows/durchstich-gewo-14.yaml` aufrufen"-Hinweis. class HubRepository implements EvaluationRepository { HubRepository._(this._client, this.settings, this._healthy); final chain.HubClient _client; final HubSettings settings; final bool _healthy; static Future connect(HubSettings settings) async { final endpoint = chain.HubEndpoint( host: settings.host, port: settings.port, secure: settings.secure, ); final client = chain.HubClient( endpoint: endpoint, authToken: settings.authToken, ); // The probe must NEVER throw out of connect(): main() awaits // this before runApp(), so any uncaught error here leaves the // app on a blank page. A timeout maps to false; so does any // transport error (e.g. a gRPC-web call against a plain-gRPC // hub, or a refused connection) — the app then comes up in // 'hub-down' state and shows the empty-state hint. bool ok; try { ok = await client.healthy().timeout( const Duration(seconds: 4), onTimeout: () => false, ); } catch (_) { ok = false; } return HubRepository._(client, settings, ok); } bool get isHealthy => _healthy; chain.HubClient get client => _client; @override Future> list() async { if (!_healthy) return const []; final events = await _client.eventLog( limit: 200, types: const ['flow.completed'], ); final evals = []; for (final e in events) { final eval = _parse(e); if (eval != null) evals.add(eval); } evals.sort((a, b) => b.skmEurPerYear.compareTo(a.skmEurPerYear)); return evals; } @override Future byEli(String eli) async { final all = await list(); for (final e in all) { if (e.norm.eli == eli) return e; } return null; } @override String get mode => _healthy ? 'hub-live' : 'hub-down'; @override String get hubLabel => '${settings.url}' '${_healthy ? '' : ' — Verbindung gestört'}'; /// Decode one flow.completed event into an Evaluation, or /// return null when the event doesn't look like a Recl∆Im /// flow. We're permissive about field absence so a partially /// failed flow run still surfaces what it produced. Evaluation? _parse(chain.LoggedEvent event) { Map? bag; try { final decoded = jsonDecode(event.detail); if (decoded is Map) bag = decoded; } catch (_) { return null; } if (bag == null) return null; final normJson = _asMap(bag['norm']); if (normJson == null) return null; final skmJson = _asMap(bag['skm']); final eli = (normJson['eli'] as String?) ?? 'eli/unknown'; final norm = Norm( eli: eli, title: (normJson['title'] as String?) ?? '(ohne Titel)', jurabk: (normJson['jurabk'] as String?) ?? '', paragraph: (normJson['paragraph'] as String?) ?? '', jurisdiction: JurisdictionLabel.parse( normJson['scope'] as String?, eli: eli, ), standDate: _parseTimestamp(event.timestamp), sourceUrl: (normJson['source_url'] as String?) ?? '', sourceSha256: (normJson['source_sha256'] as String?) ?? '', freshness: _parseFreshness(normJson['freshness']), supersededNote: normJson['superseded_note'] as String?, paragraphs: _parseParagraphs(normJson['paragraphs']), ); final cohortJson = _asMap(bag['cohort']); final affected = _asInt(cohortJson?['count']); // SKM may be missing if the flow only ran the normalize-/ // cohort-arm (e.g. flows/durchstich-from-file.yaml). Default // to zero and let the tier fall back to the cohort source's // tier — that's still the most-honest reading. final skmEur = _asDouble(skmJson?['total_eur_per_year']); final tierLowest = skmJson != null ? _parseTier(skmJson['tier_lowest']) : _parseTier(cohortJson?['tier']); final benefitJson = _asMap(bag['benefit']); final benefit = _asDouble(benefitJson?['total']); final frustJson = _asMap(bag['frust']); final frust = _asDouble(frustJson?['composite_frust']); return Evaluation( norm: norm, skmEurPerYear: skmEur, benefitScore: benefit, affectedCount: affected, frustScore: frust, tierLowest: tierLowest, duties: _parseDuties(bag['duties']), sources: _parseSources(cohortJson), auditEventId: event.eventId, notes: 'Live aus Hub-Flow ${event.flowName} ' '(execution ${event.flowExecution.substring(0, 8.clamp(0, event.flowExecution.length))}).', // Jurist sign-off is recorded by the approval, not the flow // bag. Until the review-rewrite wires approvals in, these stay // null (= unreviewed) unless the bag carries them explicitly. reviewedBy: bag['reviewed_by'] as String?, reviewedAt: DateTime.tryParse( (bag['reviewed_at'] as String?) ?? ''), ); } static Freshness _parseFreshness(dynamic v) { switch ((v as String?)?.trim().toLowerCase()) { case 'superseded' || 'amended' || 'geändert' || 'stale': return Freshness.superseded; case 'unknown' || 'ungeprüft': return Freshness.unknown; default: return Freshness.current; } } static Map? _asMap(dynamic v) => v is Map ? v : null; static double _asDouble(dynamic v) { if (v is num) return v.toDouble(); if (v is String) return double.tryParse(v) ?? 0.0; return 0.0; } static int _asInt(dynamic v) { if (v is num) return v.toInt(); if (v is String) return int.tryParse(v) ?? 0; return 0; } static EvidenceTier _parseTier(dynamic v) { final s = (v as String?)?.toUpperCase() ?? 'T4'; return switch (s) { 'T1' => EvidenceTier.t1, 'T2' => EvidenceTier.t2, 'T3' => EvidenceTier.t3, _ => EvidenceTier.t4, }; } static DateTime _parseTimestamp(String iso) => DateTime.tryParse(iso) ?? DateTime.now(); static List _parseParagraphs(dynamic v) { if (v is! List) return const []; return v.whereType>().map((p) { return NormParagraph( number: (p['number'] as String?) ?? '', text: (p['text'] as String?) ?? '', ); }).toList(); } static List _parseDuties(dynamic v) { final outer = _asMap(v); final inner = outer?['duties']; if (inner is! List) return const []; return inner.whereType>().map((d) { return Duty( modality: (d['modality'] as String?) ?? '?', addressee: (d['addressee'] as String?) ?? '', action: (d['action'] as String?) ?? '', frequency: (d['frequency'] as String?) ?? '', sourceNorm: (d['source_norm'] as String?) ?? '', authority: d['authority'] as String?, consequence: d['consequence'] as String?, ); }).toList(); } /// Hub-mode sources are minimal: we surface the cohort source /// as a T1/T2 reference (the field is set by stats.cohort_size) /// plus an audit-log pointer. Richer source attribution comes /// when the flow has a dedicated `sources.collect` step. static List _parseSources(Map? cohort) { final out = []; final label = cohort?['source'] as String?; final tierStr = (cohort?['tier'] as String?)?.toUpperCase() ?? 'T2'; if (label != null && label.isNotEmpty) { final tier = switch (tierStr) { 'T1' => EvidenceTier.t1, 'T2' => EvidenceTier.t2, 'T3' => EvidenceTier.t3, _ => EvidenceTier.t4, }; out.add(Source(label: label, url: '', tier: tier)); } return out; } }