diff --git a/kb-app-demo-desktop/Cargo.toml b/kb-app-demo-desktop/Cargo.toml index 78105c5..e380f5d 100644 --- a/kb-app-demo-desktop/Cargo.toml +++ b/kb-app-demo-desktop/Cargo.toml @@ -25,6 +25,7 @@ chrono.workspace = true fs2.workspace = true kb-config = { path = "../kb-config" } kb-core = { path = "../kb-core" } +kb-lib = { path = "../kb-lib" } kb-logging = { path = "../kb-logging" } kb-program-ids = { path = "../kb-program-ids" } kb-store = { path = "../kb-store" } diff --git a/kb-app-demo-desktop/capabilities/default.json b/kb-app-demo-desktop/capabilities/default.json index 0ff3083..a77c31c 100644 --- a/kb-app-demo-desktop/capabilities/default.json +++ b/kb-app-demo-desktop/capabilities/default.json @@ -12,10 +12,12 @@ "demo_sql_pg_raw", "demo_sql_pg_core", "demo_sql_replay_candidates", - "demo_config" + "demo_config", + "demo_core_extraction", + "demo_decode_replay" ], "permissions": [ "core:default", "tracing:default" ] -} \ No newline at end of file +} diff --git a/kb-app-demo-desktop/frontend/demo_core_extraction.html b/kb-app-demo-desktop/frontend/demo_core_extraction.html new file mode 100644 index 0000000..b4490a6 --- /dev/null +++ b/kb-app-demo-desktop/frontend/demo_core_extraction.html @@ -0,0 +1,160 @@ + + + + + + + + + Khadhroony Bot2 - Extraction canonical vers core + + + + +
+ +
+ +
+
+
+
+
+

Extraction transactionnelle canonical vers core

+

Chaque transaction canonique sélectionnée est validée, normalisée dans les tables core et enregistrée dans le ledger de traitement au sein d'une seule transaction PostgreSQL.

+
+
+ +
+
+

+ +

+
+
+
+
+ + +
+
+ + +
+
+
+ + +
+
+
+
+
+
+ +
+

+ +

+
+
+ + + +
+
+
+ +
+

+ +

+
+
+

Sélectionne les transactions canoniques dont l'état de traitement est received, dans l'ordre des slots.

+ +
+
+
+ +
+

+ +

+
+
+
+
+ + +
+
+ + +
+
+ +
+
+
+ +
+

+ +

+
+
+

Sélectionne les transactions raw ayant déjà au moins une instruction core résolue pour ce programme. Ce mode sert aux replays après changement de version ou de hash.

+ + + +
+
+
+ +
+

+ +

+
+
+
+

Journal

+
+ +

Résumé JSON

+ +
+
+
+
+
+
+
+ + + + + + \ No newline at end of file diff --git a/kb-app-demo-desktop/frontend/demo_decode_replay.html b/kb-app-demo-desktop/frontend/demo_decode_replay.html new file mode 100644 index 0000000..e3d95a6 --- /dev/null +++ b/kb-app-demo-desktop/frontend/demo_decode_replay.html @@ -0,0 +1,166 @@ + + + + + + + + + Décodage et matérialisation + + + + + +
+ +
+
+
+
+
+
+
+
+

Sélection bornée

+
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+
+
+
+
+

Dispatch et replay

+
+
+
+ + +
+
+
+ + +
+
+ + +
+
+ + +
+
Sans signatures explicites, cocher Toutes les signatures pour autoriser un force replay borné. Le filtre d’état est alors ignoré, mais les programmes, paths et la limite restent appliqués.
+
+ + +
+
Chargement de la disponibilité des matérialiseurs…
+
+
+ +
+
+
+
+
+
+
+
+
+ + + +
+

Journal

+ +

Résumé

+ +

Diagnostics

+ +
+
+
+
+
+
+
+
+ + +
+
+ + +
+ +
+

Annotations non chargées.

+
+ + + + + + + + + + + + + + +
SlotSignaturePathGénérationTexteOctetsSignataires vérifiésSHA-256 / provenance
+
+
+
+
+
+
+
+
+ + + + + diff --git a/kb-app-demo-desktop/frontend/main.html b/kb-app-demo-desktop/frontend/main.html index 1f49a2f..5a7c871 100644 --- a/kb-app-demo-desktop/frontend/main.html +++ b/kb-app-demo-desktop/frontend/main.html @@ -1,5 +1,5 @@ - + @@ -28,6 +28,9 @@
  • HTTP JSON-RPC
  • Backfill HTTP
    • +
  • Extraction core
    • +
  • Décodage contextualisé
    • +
  • SQL diagnostics
  • SQL PostgreSQL raw
  • SQL PostgreSQL core
    • diff --git a/kb-app-demo-desktop/frontend/ts/demo_core_extraction.ts b/kb-app-demo-desktop/frontend/ts/demo_core_extraction.ts new file mode 100644 index 0000000..747628c --- /dev/null +++ b/kb-app-demo-desktop/frontend/ts/demo_core_extraction.ts @@ -0,0 +1,185 @@ +// file: kb-app-demo-desktop/frontend/ts/demo_core_extraction.ts +// version: 4 + +import * as bootstrap from "bootstrap"; +import "simplebar"; +import ResizeObserver from "resize-observer-polyfill"; +import { invoke } from "@tauri-apps/api/core"; +import { listen } from "@tauri-apps/api/event"; +import { frontendDebug, frontendError, installFrontendConsoleBridge } from "./frontend_log.ts"; +import type { DemoCoreExtractionOptionsPayload } from "./bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionOptionsPayload.ts"; +import type { DemoCoreExtractionProgressPayload } from "./bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionProgressPayload.ts"; +import type { DemoCoreExtractionRequest } from "./bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionRequest.ts"; +import type { DemoCoreExtractionSummaryPayload } from "./bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionSummaryPayload.ts"; + +(window as Window & typeof globalThis & { bootstrap?: typeof bootstrap }).bootstrap = bootstrap; +(window as Window & typeof globalThis & { ResizeObserver?: typeof ResizeObserver }).ResizeObserver = ResizeObserver; + +const logLines: string[] = []; +const maximumLogLines = 1000; +let running = false; + +function element(selector: string): T { + const value = document.querySelector(selector); + if (!value) { + throw new Error(`Missing UI element: ${selector}`); + } + return value; +} + +function inputValue(selector: string): string { + return element(selector).value.trim(); +} + +function integerValue(selector: string): number { + const parsed = Number.parseInt(inputValue(selector), 10); + if (!Number.isFinite(parsed)) { + throw new Error(`Valeur numérique invalide pour ${selector}`); + } + return parsed; +} + +function nullableIntegerValue(selector: string): number | null { + const value = inputValue(selector); + if (value.length === 0) { + return null; + } + const parsed = Number.parseInt(value, 10); + if (!Number.isFinite(parsed)) { + throw new Error(`Valeur numérique invalide pour ${selector}`); + } + return parsed; +} + +function appendLog(payload: DemoCoreExtractionProgressPayload | { timestamp: string; level: string; message: string }): void { + const progress = "completed" in payload ? ` [${payload.completed}/${payload.total}]` : ""; + logLines.push(`${payload.timestamp} ${payload.level.toUpperCase()}${progress} ${payload.message}`); + while (logLines.length > maximumLogLines) { + logLines.shift(); + } + const output = element("#coreExtractionLogOutput"); + output.value = logLines.join("\n"); + output.scrollTop = output.scrollHeight; +} + +function setRunning(value: boolean): void { + running = value; + document.querySelectorAll(".core-extraction-start-button").forEach(button => { + button.disabled = value; + }); + element("#cancelCoreExtractionButton").disabled = !value; + const badge = element("#coreExtractionStatusBadge"); + badge.textContent = value ? "Extraction en cours" : "Prêt"; + badge.className = value ? "badge text-bg-warning" : "badge text-bg-success"; +} + +function buildRequest(mode: string): DemoCoreExtractionRequest { + return { + mode, + signaturesText: mode === "signatures" ? element("#coreExtractionSignaturesTextarea").value : null, + programId: mode === "program_id" ? inputValue("#coreExtractionProgramIdInput") : null, + minSlot: mode === "slot_range" ? nullableIntegerValue("#coreExtractionMinSlotInput") : null, + maxSlot: mode === "slot_range" ? nullableIntegerValue("#coreExtractionMaxSlotInput") : null, + limit: integerValue("#coreExtractionLimitInput"), + maxConcurrentExtractions: integerValue("#coreExtractionConcurrencyInput"), + forceReplay: element("#coreExtractionForceReplayInput").checked, + }; +} + +async function executeCoreExtraction(mode: string): Promise { + if (running) { + return; + } + try { + const request = buildRequest(mode); + setRunning(true); + element("#coreExtractionSummaryOutput").value = "Exécution en cours..."; + appendLog({ timestamp: new Date().toISOString(), level: "info", message: `Démarrage du mode ${mode}` }); + const summary = await invoke("demo_core_extraction_execute", { request }); + element("#coreExtractionSummaryOutput").value = JSON.stringify(summary, null, 2); + appendLog({ + timestamp: new Date().toISOString(), + level: summary.cancelled || Number(summary.failed) > 0 ? "warn" : "info", + message: `Campagne terminée : extracted=${summary.extracted}, skipped=${summary.skipped}, failed=${summary.failed}, cancelled=${summary.cancelledCandidates}, notStarted=${summary.notStarted}`, + }); + } catch (caughtError) { + const message = caughtError instanceof Error ? caughtError.message : String(caughtError); + element("#coreExtractionSummaryOutput").value = JSON.stringify({ error: message }, null, 2); + appendLog({ timestamp: new Date().toISOString(), level: "error", message }); + frontendError("kb_app_demo_desktop.frontend.demo_core_extraction", `Core extraction failed: ${message}`); + } finally { + setRunning(false); + } +} + +async function cancelCoreExtraction(): Promise { + try { + const accepted = await invoke("demo_core_extraction_cancel"); + appendLog({ + timestamp: new Date().toISOString(), + level: accepted ? "warn" : "info", + message: accepted ? "Demande d'arrêt envoyée." : "Aucune campagne active.", + }); + } catch (caughtError) { + const message = caughtError instanceof Error ? caughtError.message : String(caughtError); + frontendError("kb_app_demo_desktop.frontend.demo_core_extraction", `Cancellation failed: ${message}`); + } +} + + +async function openDiagnostics(command: "open_demo_sql_pg_raw_window" | "open_demo_sql_pg_core_window" | "open_demo_sql_replay_candidates_window", label: string): Promise { + try { + await invoke(command); + appendLog({ timestamp: new Date().toISOString(), level: "info", message: `${label} ouvert.` }); + } catch (caughtError) { + const message = caughtError instanceof Error ? caughtError.message : String(caughtError); + appendLog({ timestamp: new Date().toISOString(), level: "error", message }); + frontendError("kb_app_demo_desktop.frontend.demo_core_extraction", `${label} opening failed: ${message}`); + } +} + +async function loadOptions(): Promise { + const options = await invoke("demo_core_extraction_options"); + element("#coreExtractionVersionBadge").textContent = `Version ${options.processorVersion}`; + element("#coreExtractionLimitInput").value = String(options.defaultLimit); + element("#coreExtractionConcurrencyInput").value = String(options.defaultMaxConcurrentExtractions); + setRunning(options.running); +} + +document.addEventListener("DOMContentLoaded", () => { + installFrontendConsoleBridge("kb_app_demo_desktop.frontend.demo_core_extraction"); + frontendDebug("kb_app_demo_desktop.frontend.demo_core_extraction", "core extraction demo window loaded"); + document.querySelectorAll('[data-bs-toggle="tooltip"]').forEach(item => new bootstrap.Tooltip(item)); + document.querySelectorAll(".core-extraction-start-button").forEach(button => { + button.addEventListener("click", () => { + const mode = button.dataset.coreExtractionMode; + if (mode) { + void executeCoreExtraction(mode); + } + }); + }); + element("#openRawDiagnosticsButton").addEventListener("click", () => { + void openDiagnostics("open_demo_sql_pg_raw_window", "Diagnostic raw"); + }); + element("#openCoreDiagnosticsButton").addEventListener("click", () => { + void openDiagnostics("open_demo_sql_pg_core_window", "Diagnostic core"); + }); + element("#openReplayCandidatesButton").addEventListener("click", () => { + void openDiagnostics("open_demo_sql_replay_candidates_window", "Sélecteur de candidats replay"); + }); + element("#cancelCoreExtractionButton").addEventListener("click", () => { + void cancelCoreExtraction(); + }); + element("#clearCoreExtractionLogButton").addEventListener("click", () => { + logLines.length = 0; + element("#coreExtractionLogOutput").value = ""; + }); + void listen("demo-core-extraction-progress", event => { + appendLog(event.payload); + }); + void loadOptions().catch(caughtError => { + const message = caughtError instanceof Error ? caughtError.message : String(caughtError); + appendLog({ timestamp: new Date().toISOString(), level: "error", message }); + frontendError("kb_app_demo_desktop.frontend.demo_core_extraction", `Options loading failed: ${message}`); + }); +}); diff --git a/kb-app-demo-desktop/frontend/ts/demo_decode_replay.ts b/kb-app-demo-desktop/frontend/ts/demo_decode_replay.ts new file mode 100644 index 0000000..b5e8315 --- /dev/null +++ b/kb-app-demo-desktop/frontend/ts/demo_decode_replay.ts @@ -0,0 +1,340 @@ +// file: kb-app-demo-desktop/frontend/ts/demo_decode_replay.ts +// version: 5 + +import * as bootstrap from "bootstrap"; +import "simplebar"; +import ResizeObserver from "resize-observer-polyfill"; +import { invoke } from "@tauri-apps/api/core"; +import { listen } from "@tauri-apps/api/event"; +import { frontendDebug, frontendError, installFrontendConsoleBridge } from "./frontend_log.ts"; +import type { DemoDecodeDiagnosticsPayload } from "./bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeDiagnosticsPayload.ts"; +import type { DemoDecodeReplayOptionsPayload } from "./bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayOptionsPayload.ts"; +import type { DemoDecodeReplayProgressPayload } from "./bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayProgressPayload.ts"; +import type { DemoDecodeReplayRequest } from "./bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayRequest.ts"; +import type { DemoDecodeReplaySummaryPayload } from "./bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplaySummaryPayload.ts"; +import type { DemoTransactionAnnotationRequest } from "./bindings/kb_app_demo_desktop/demo_decode_replay/DemoTransactionAnnotationRequest.ts"; +import type { DemoTransactionAnnotationRow } from "./bindings/kb_app_demo_desktop/demo_decode_replay/DemoTransactionAnnotationRow.ts"; + +(window as Window & typeof globalThis & { bootstrap?: typeof bootstrap }).bootstrap = bootstrap; +(window as Window & typeof globalThis & { ResizeObserver?: typeof ResizeObserver }).ResizeObserver = ResizeObserver; + +const logLines: string[] = []; +let running = false; +let materializationAvailable = false; + +function element(selector: string): T { + const value = document.querySelector(selector); + if (!value) { + throw new Error(`Missing UI element: ${selector}`); + } + return value; +} + +function inputValue(selector: string): string { + return element(selector).value.trim(); +} + +function integerValue(selector: string): number { + const value = Number.parseInt(inputValue(selector), 10); + if (!Number.isFinite(value) || value < 1) { + throw new Error(`Valeur numérique invalide pour ${selector}`); + } + return value; +} + +function appendLog(payload: DemoDecodeReplayProgressPayload | { timestamp: string; level: string; message: string }): void { + const progress = "completed" in payload ? ` [${payload.completed}/${payload.total}]` : ""; + const campaign = "campaignId" in payload ? ` [${payload.campaignId}]` : ""; + logLines.push(`${payload.timestamp} ${payload.level.toUpperCase()}${campaign}${progress} ${payload.message}`); + while (logLines.length > 1000) { + logLines.shift(); + } + const output = element("#decodeReplayLogOutput"); + output.value = logLines.join("\n"); + output.scrollTop = output.scrollHeight; +} + +function explicitSignatureCount(): number { + return element("#decodeReplaySignaturesTextarea") + .value + .split(/\r?\n/) + .filter(value => value.trim().length > 0) + .length; +} + +function updateReplayScopeControls(): void { + const signaturesPresent = explicitSignatureCount() > 0; + const incompleteSignatures = inputValue("#decodeReplayStateSelect") === "incomplete_signatures"; + const allMatchingInput = element("#decodeReplayAllSignaturesInput"); + const forceInput = element("#decodeReplayForceInput"); + const materializeInput = element("#decodeReplayMaterializeInput"); + if (!running && (signaturesPresent || incompleteSignatures) && allMatchingInput.checked) { + allMatchingInput.checked = false; + } + allMatchingInput.disabled = running || signaturesPresent || incompleteSignatures; + const forceScopeMissing = !signaturesPresent && !allMatchingInput.checked; + forceInput.disabled = running || forceScopeMissing || incompleteSignatures; + if (!running && (forceScopeMissing || incompleteSignatures)) { + forceInput.checked = false; + } + materializeInput.disabled = running || !materializationAvailable; + if (!running && !materializationAvailable) { + materializeInput.checked = false; + } +} + +function setRunning(value: boolean): void { + running = value; + element("#startDecodeReplayButton").disabled = value; + element("#cancelDecodeReplayButton").disabled = !value; + const badge = element("#decodeReplayStatusBadge"); + badge.textContent = value ? "Replay en cours" : "Prêt"; + badge.className = value ? "badge text-bg-warning" : "badge text-bg-success"; + updateReplayScopeControls(); +} + +function selectedDecoderNames(): string[] { + return Array.from(document.querySelectorAll(".decode-replay-decoder-input:checked")) + .map(input => input.value); +} + +function buildRequest(): DemoDecodeReplayRequest { + return { + signaturesText: element("#decodeReplaySignaturesTextarea").value || null, + programId: inputValue("#decodeReplayProgramIdInput") || null, + instructionState: inputValue("#decodeReplayStateSelect"), + instructionPathsText: element("#decodeReplayPathsTextarea").value || null, + decoderNames: selectedDecoderNames(), + limit: integerValue("#decodeReplayLimitInput"), + maxConcurrentInputs: integerValue("#decodeReplayConcurrencyInput"), + allCompatible: element("#decodeReplayAllCompatibleInput").checked, + forceReplay: element("#decodeReplayForceInput").checked, + forceReplayAllMatching: element("#decodeReplayAllSignaturesInput").checked, + materializeAfterDecode: element("#decodeReplayMaterializeInput").checked, + }; +} + +function validationMessage(request: DemoDecodeReplayRequest): string | null { + const signatureCount = request.signaturesText + ? request.signaturesText.split(/\r?\n/).filter(value => value.trim().length > 0).length + : 0; + if (request.decoderNames.length === 0) { + return "Sélectionner au moins un décodeur."; + } + if (request.instructionState === "incomplete_signatures" && request.forceReplay) { + return "Le mode signatures incomplètes rejoue son périmètre sans Force replay global."; + } + if (request.forceReplay && signatureCount === 0 && !request.forceReplayAllMatching) { + return "Le force replay nécessite des signatures explicites ou l’autorisation Toutes les signatures."; + } + if (request.forceReplayAllMatching && signatureCount > 0) { + return "Le mode « Toutes les signatures » ne peut pas être combiné avec une liste de signatures explicites."; + } + if (request.forceReplayAllMatching && !request.forceReplay) { + return "Le mode « Toutes les signatures » est uniquement disponible avec Force replay."; + } + if (request.materializeAfterDecode && !materializationAvailable) { + return "Aucun matérialiseur n’est actuellement enregistré dans cette application."; + } + return null; +} + +function reportValidationWarning(message: string): void { + element("#decodeReplaySummaryOutput").value = JSON.stringify({ validation: message }, null, 2); + appendLog({ timestamp: new Date().toISOString(), level: "warn", message }); + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", `Decode replay request rejected locally: ${message}`); +} + +async function executeDecodeReplay(): Promise { + if (running) { + return; + } + try { + const request = buildRequest(); + const invalid = validationMessage(request); + if (invalid) { + reportValidationWarning(invalid); + return; + } + setRunning(true); + element("#decodeReplaySummaryOutput").value = "Exécution en cours..."; + const signatureCount = request.signaturesText + ? request.signaturesText.split(/\r?\n/).filter(value => value.trim().length > 0).length + : 0; + const requestSummary = `signatures=${signatureCount}, programId=${request.programId ?? "auto"}, state=${request.instructionState}, decoders=${request.decoderNames.join(",")}, limit=${request.limit}, concurrency=${request.maxConcurrentInputs}, allCompatible=${request.allCompatible}, forceReplay=${request.forceReplay}, allSignatures=${request.forceReplayAllMatching}, materialize=${request.materializeAfterDecode}`; + appendLog({ timestamp: new Date().toISOString(), level: "info", message: `Démarrage du replay de décodage contextualisé : ${requestSummary}` }); + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", `Invoke demo_decode_replay_execute ${requestSummary}`); + const summary = await invoke("demo_decode_replay_execute", { request }); + element("#decodeReplaySummaryOutput").value = JSON.stringify(summary, null, 2); + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", `Invoke demo_decode_replay_execute completed summary=${JSON.stringify(summary)}`); + appendLog({ + timestamp: new Date().toISOString(), + level: summary.cancelled || summary.failedInputs > 0 ? "warn" : "info", + message: `Campagne ${summary.campaignId} terminée : completed=${summary.completed}, unmatched=${summary.unmatched}, failed=${summary.failedInputs}, notStarted=${summary.notStarted}`, + }); + } catch (caughtError) { + const message = caughtError instanceof Error ? caughtError.message : String(caughtError); + element("#decodeReplaySummaryOutput").value = JSON.stringify({ error: message }, null, 2); + appendLog({ timestamp: new Date().toISOString(), level: "error", message }); + frontendError("kb_app_demo_desktop.frontend.demo_decode_replay", `Decode replay failed: ${message}`); + } finally { + setRunning(false); + } +} + +async function cancelDecodeReplay(): Promise { + try { + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", "Invoke demo_decode_replay_cancel"); + const accepted = await invoke("demo_decode_replay_cancel"); + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", `Invoke demo_decode_replay_cancel completed accepted=${accepted}`); + appendLog({ + timestamp: new Date().toISOString(), + level: accepted ? "warn" : "info", + message: accepted ? "Demande d'arrêt envoyée." : "Aucune campagne active.", + }); + } catch (caughtError) { + const message = caughtError instanceof Error ? caughtError.message : String(caughtError); + frontendError("kb_app_demo_desktop.frontend.demo_decode_replay", `Cancellation failed: ${message}`); + } +} + +async function loadDiagnostics(): Promise { + try { + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", "Invoke demo_decode_replay_diagnostics"); + const diagnostics = await invoke("demo_decode_replay_diagnostics"); + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", `Invoke demo_decode_replay_diagnostics completed tables=${diagnostics.tables.length} coverage=${diagnostics.coverage.length}`); + element("#decodeReplayDiagnosticsOutput").value = JSON.stringify(diagnostics, null, 2); + } catch (caughtError) { + const message = caughtError instanceof Error ? caughtError.message : String(caughtError); + element("#decodeReplayDiagnosticsOutput").value = JSON.stringify({ error: message }, null, 2); + frontendError("kb_app_demo_desktop.frontend.demo_decode_replay", `Diagnostics loading failed: ${message}`); + } +} + +function tableCell(value: string, title?: string): HTMLTableCellElement { + const cell = document.createElement("td"); + cell.textContent = value; + if (title) { + cell.title = title; + } + return cell; +} + +function shortened(value: string, visible: number): string { + return value.length > visible ? `${value.slice(0, visible)}…` : value; +} + +function renderAnnotations(rows: DemoTransactionAnnotationRow[]): void { + const body = element("#annotationJournalBody"); + body.replaceChildren(); + for (const row of rows) { + const tableRow = document.createElement("tr"); + tableRow.append( + tableCell(row.slot), + tableCell(shortened(row.signature, 18), row.signature), + tableCell(row.instructionPath), + tableCell(row.generation, row.programId), + tableCell(row.text), + tableCell(String(row.payloadLengthBytes)), + tableCell(row.verifiedSigners.join(", ") || "—"), + tableCell(`${shortened(row.payloadSha256, 12)} · ${row.decoder}`, `${row.payloadSha256}\n${row.idempotenceKey}\ntransaction_annotations@${row.processorVersion}`), + ); + body.append(tableRow); + } + element("#annotationJournalStatus").textContent = `${rows.length} annotation(s) committed chargée(s).`; +} + +async function loadAnnotations(): Promise { + const request: DemoTransactionAnnotationRequest = { + signatureContains: inputValue("#annotationSignatureInput") || null, + limit: integerValue("#annotationLimitInput"), + }; + try { + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", `Invoke demo_decode_replay_annotations limit=${request.limit}`); + const rows = await invoke("demo_decode_replay_annotations", { request }); + renderAnnotations(rows); + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", `Invoke demo_decode_replay_annotations completed rows=${rows.length}`); + } catch (caughtError) { + const message = caughtError instanceof Error ? caughtError.message : String(caughtError); + element("#annotationJournalStatus").textContent = `Erreur : ${message}`; + frontendError("kb_app_demo_desktop.frontend.demo_decode_replay", `Annotation journal loading failed: ${message}`); + } +} + +async function openWindow(command: "open_demo_sql_pg_core_window" | "open_demo_sql_replay_candidates_window"): Promise { + try { + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", `Invoke ${command}`); + await invoke(command); + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", `Invoke ${command} completed`); + } catch (caughtError) { + const message = caughtError instanceof Error ? caughtError.message : String(caughtError); + frontendError("kb_app_demo_desktop.frontend.demo_decode_replay", `Window opening failed: ${message}`); + } +} + +async function loadOptions(): Promise { + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", "Invoke demo_decode_replay_options"); + const options = await invoke("demo_decode_replay_options"); + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", `Invoke demo_decode_replay_options completed options=${JSON.stringify(options)}`); + element("#decodeReplayVersionBadge").textContent = `Pipeline ${options.pipelineVersion}`; + element("#decodeReplayLimitInput").value = String(options.defaultLimit); + element("#decodeReplayConcurrencyInput").value = String(options.defaultMaxConcurrentInputs); + const container = element("#decodeReplayDecoderList"); + container.innerHTML = ""; + for (const decoder of options.decoders) { + const wrapper = document.createElement("div"); + wrapper.className = "form-check border rounded p-3 ps-5 mb-2"; + const input = document.createElement("input"); + input.className = "form-check-input decode-replay-decoder-input"; + input.type = "checkbox"; + input.value = decoder.name; + input.id = `decode-${decoder.name}`; + input.checked = true; + const label = document.createElement("label"); + label.className = "form-check-label"; + label.htmlFor = input.id; + label.textContent = `${decoder.name} v${decoder.version} — ${decoder.programIds.length} programmes`; + wrapper.append(input, label); + container.append(wrapper); + } + materializationAvailable = options.materializerNames.length > 0; + const materializeHelp = element("#decodeReplayMaterializeHelp"); + materializeHelp.textContent = materializationAvailable + ? `Matérialiseurs disponibles : ${options.materializerNames.join(", ")}` + : "Aucun matérialiseur enregistré : option désactivée."; + setRunning(options.running); +} + +document.addEventListener("DOMContentLoaded", () => { + installFrontendConsoleBridge("kb_app_demo_desktop.frontend.demo_decode_replay"); + frontendDebug("kb_app_demo_desktop.frontend.demo_decode_replay", "decode replay demo window loaded"); + document.querySelectorAll('[data-bs-toggle="tooltip"]').forEach(item => new bootstrap.Tooltip(item)); + element("#decodeReplaySignaturesTextarea").addEventListener("input", () => updateReplayScopeControls()); + element("#decodeReplayStateSelect").addEventListener("change", () => updateReplayScopeControls()); + element("#decodeReplayAllSignaturesInput").addEventListener("change", event => { + const input = event.currentTarget as HTMLInputElement; + if (input.checked) { + element("#decodeReplayForceInput").checked = true; + } + updateReplayScopeControls(); + }); + element("#decodeReplayForceInput").addEventListener("change", event => { + const input = event.currentTarget as HTMLInputElement; + if (!input.checked) { + element("#decodeReplayAllSignaturesInput").checked = false; + } + updateReplayScopeControls(); + }); + element("#startDecodeReplayButton").addEventListener("click", () => void executeDecodeReplay()); + element("#cancelDecodeReplayButton").addEventListener("click", () => void cancelDecodeReplay()); + element("#loadDecodeDiagnosticsButton").addEventListener("click", () => void loadDiagnostics()); + element("#loadAnnotationsButton").addEventListener("click", () => void loadAnnotations()); + element("#openReplayCandidatesFromDecodeButton").addEventListener("click", () => void openWindow("open_demo_sql_replay_candidates_window")); + element("#openCoreDiagnosticsFromDecodeButton").addEventListener("click", () => void openWindow("open_demo_sql_pg_core_window")); + void listen("demo-decode-replay-progress", event => appendLog(event.payload)); + void loadOptions().catch(caughtError => { + const message = caughtError instanceof Error ? caughtError.message : String(caughtError); + appendLog({ timestamp: new Date().toISOString(), level: "error", message }); + frontendError("kb_app_demo_desktop.frontend.demo_decode_replay", `Options loading failed: ${message}`); + }); +}); diff --git a/kb-app-demo-desktop/frontend/ts/main.ts b/kb-app-demo-desktop/frontend/ts/main.ts index 459e5a4..88d1b11 100644 --- a/kb-app-demo-desktop/frontend/ts/main.ts +++ b/kb-app-demo-desktop/frontend/ts/main.ts @@ -1,5 +1,5 @@ // file: kb-app-demo-desktop/frontend/ts/main.ts -// version: 3 +// version: 4 import * as bootstrap from "bootstrap"; import MarkdownIt from "markdown-it"; @@ -75,6 +75,20 @@ if (openDemoBackfillLink) { }); } +const pipelineDemoLinks: ReadonlyArray = [ + ["openDemoCoreExtractionLink", "open_demo_core_extraction_window"], + ["openDemoDecodeReplayLink", "open_demo_decode_replay_window"], +]; +for (const [elementId, command] of pipelineDemoLinks) { + const link = document.querySelector(`#${elementId}`); + if (link) { + link.addEventListener("click", event => { + event.preventDefault(); + void invoke(command); + }); + } +} + const sqlDemoLinks: ReadonlyArray = [ ["openDemoSqlDiagLink", "open_demo_sql_diag_window"], ["openDemoSqlPgRawLink", "open_demo_sql_pg_raw_window"], diff --git a/kb-app-demo-desktop/src/app_state.rs b/kb-app-demo-desktop/src/app_state.rs index 46d2026..39dbb8b 100644 --- a/kb-app-demo-desktop/src/app_state.rs +++ b/kb-app-demo-desktop/src/app_state.rs @@ -1,5 +1,5 @@ // file: kb-app-demo-desktop/src/app_state.rs -// version: 4 +// version: 5 //! Shared Tauri application state and startup initialization. @@ -10,10 +10,17 @@ pub(crate) struct AppState { active_profile: kb_config::ProfileConfig, logging_guard: std::sync::Mutex, http_pool: kb_onchain_transport::HttpEndpointPool, - ws_pool: std::sync::Mutex>>, - demo_ws_session: tokio::sync::Mutex>>, + ws_pool: + std::sync::Mutex>>, + demo_ws_session: + tokio::sync::Mutex>>, demo_backfill_running: std::sync::atomic::AtomicBool, demo_backfill_cancel_requested: std::sync::atomic::AtomicBool, + demo_core_extraction_running: std::sync::atomic::AtomicBool, + demo_core_extraction_cancel_requested: std::sync::atomic::AtomicBool, + demo_decode_replay_running: std::sync::atomic::AtomicBool, + demo_decode_replay_cancel_requested: std::sync::atomic::AtomicBool, + demo_decode_replay_campaign_id: std::sync::Mutex>, } impl crate::AppState { @@ -33,7 +40,8 @@ impl crate::AppState { std::result::Result::Ok(guard) => guard, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let http_pool = match kb_onchain_transport::HttpEndpointPool::from_profile(&active_profile) { + let http_pool = match kb_onchain_transport::HttpEndpointPool::from_profile(&active_profile) + { std::result::Result::Ok(pool) => pool, std::result::Result::Err(error) => return std::result::Result::Err(error), }; @@ -47,6 +55,11 @@ impl crate::AppState { demo_ws_session: tokio::sync::Mutex::new(std::option::Option::None), demo_backfill_running: std::sync::atomic::AtomicBool::new(false), demo_backfill_cancel_requested: std::sync::atomic::AtomicBool::new(false), + demo_core_extraction_running: std::sync::atomic::AtomicBool::new(false), + demo_core_extraction_cancel_requested: std::sync::atomic::AtomicBool::new(false), + demo_decode_replay_running: std::sync::atomic::AtomicBool::new(false), + demo_decode_replay_cancel_requested: std::sync::atomic::AtomicBool::new(false), + demo_decode_replay_campaign_id: std::sync::Mutex::new(std::option::Option::None), }); } @@ -97,9 +110,8 @@ impl crate::AppState { /// Returns the persistent WebSocket demo session slot. pub(crate) fn demo_ws_session( &self, - ) -> &tokio::sync::Mutex< - std::option::Option>, - > { + ) -> &tokio::sync::Mutex>> + { return &self.demo_ws_session; } @@ -113,6 +125,33 @@ impl crate::AppState { return &self.demo_backfill_cancel_requested; } + /// Returns the core extraction running flag. + pub(crate) fn demo_core_extraction_running(&self) -> &std::sync::atomic::AtomicBool { + return &self.demo_core_extraction_running; + } + + /// Returns the core extraction cancellation flag. + pub(crate) fn demo_core_extraction_cancel_requested(&self) -> &std::sync::atomic::AtomicBool { + return &self.demo_core_extraction_cancel_requested; + } + + /// Returns the decode replay running flag. + pub(crate) fn demo_decode_replay_running(&self) -> &std::sync::atomic::AtomicBool { + return &self.demo_decode_replay_running; + } + + /// Returns the decode replay cancellation flag. + pub(crate) fn demo_decode_replay_cancel_requested(&self) -> &std::sync::atomic::AtomicBool { + return &self.demo_decode_replay_cancel_requested; + } + + /// Returns the active decode replay campaign slot. + pub(crate) fn demo_decode_replay_campaign_id( + &self, + ) -> &std::sync::Mutex> { + return &self.demo_decode_replay_campaign_id; + } + /// Returns the number of logging routes held by the logging guard. pub(crate) fn logging_route_count(&self) -> usize { let lock_result = self.logging_guard.lock(); diff --git a/kb-app-demo-desktop/src/demo_config.rs b/kb-app-demo-desktop/src/demo_config.rs index 1fb4a52..d3c2265 100644 --- a/kb-app-demo-desktop/src/demo_config.rs +++ b/kb-app-demo-desktop/src/demo_config.rs @@ -1,5 +1,5 @@ // file: kb-app-demo-desktop/src/demo_config.rs -// version: 1 +// version: 2 //! Configuration demo payload and state projection. diff --git a/kb-app-demo-desktop/src/demo_core_extraction.rs b/kb-app-demo-desktop/src/demo_core_extraction.rs new file mode 100644 index 0000000..bf404ea --- /dev/null +++ b/kb-app-demo-desktop/src/demo_core_extraction.rs @@ -0,0 +1,267 @@ +// file: kb-app-demo-desktop/src/demo_core_extraction.rs +// version: 9 + +//! Tauri commands and UI payloads for canonical transaction to core extraction. + +use tauri::Emitter; // rust-rules: trait-import +use ts_rs::TS; // rust-rules: derive-import + +/// Initial options shown by the core extraction demo. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionOptionsPayload.ts" +)] +pub(crate) struct DemoCoreExtractionOptionsPayload { + /// Current extractor implementation version. + pub(crate) processor_version: std::string::String, + /// Default maximum selected canonical transactions. + pub(crate) default_limit: u32, + /// Default maximum concurrent extractions. + pub(crate) default_max_concurrent_extractions: u32, + /// Whether one extraction campaign is currently running. + pub(crate) running: bool, +} + +/// UI request for one bounded core extraction campaign. +#[derive(Clone, Debug, serde::Deserialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionRequest.ts" +)] +pub(crate) struct DemoCoreExtractionRequest { + /// Source mode: signatures, pending or slot_range. + pub(crate) mode: std::string::String, + /// Newline-separated signatures for exact selection. + pub(crate) signatures_text: std::option::Option, + /// Optional program id already present in core instructions. + pub(crate) program_id: std::option::Option, + /// Optional inclusive minimum slot. + #[ts(type = "number | null")] + pub(crate) min_slot: std::option::Option, + /// Optional inclusive maximum slot. + #[ts(type = "number | null")] + pub(crate) max_slot: std::option::Option, + /// Maximum selected canonical transactions. + pub(crate) limit: u32, + /// Maximum concurrent extraction operations. + pub(crate) max_concurrent_extractions: u32, + /// Forces replacement of already current core rows. + pub(crate) force_replay: bool, +} + +/// One progress event emitted to the core extraction window. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionProgressPayload.ts" +)] +pub(crate) struct DemoCoreExtractionProgressPayload { + /// RFC 3339 timestamp. + pub(crate) timestamp: std::string::String, + /// Stable severity code. + pub(crate) level: std::string::String, + /// Human-readable message. + pub(crate) message: std::string::String, + /// Number of terminal candidates. + #[ts(type = "number")] + pub(crate) completed: u64, + /// Total selected candidates. + #[ts(type = "number")] + pub(crate) total: u64, +} + +/// Final UI-safe summary for one extraction campaign. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionSummaryPayload.ts" +)] +pub(crate) struct DemoCoreExtractionSummaryPayload { + /// Current extractor implementation version. + pub(crate) processor_version: std::string::String, + /// Number of selected canonical transactions. + #[ts(type = "number")] + pub(crate) selected: u64, + /// Number admitted to the bounded execution queue. + #[ts(type = "number")] + pub(crate) started: u64, + /// Number reaching a terminal result. + #[ts(type = "number")] + pub(crate) completed: u64, + /// Number skipped by version/hash idempotence. + #[ts(type = "number")] + pub(crate) skipped: u64, + /// Number extracted and committed. + #[ts(type = "number")] + pub(crate) extracted: u64, + /// Number failed. + #[ts(type = "number")] + pub(crate) failed: u64, + /// Number admitted but cancelled before a terminal result. + #[ts(type = "number")] + pub(crate) cancelled_candidates: u64, + /// Number selected but not started after cancellation. + #[ts(type = "number")] + pub(crate) not_started: u64, + /// Whether the campaign was cancelled. + pub(crate) cancelled: bool, + /// Campaign start timestamp. + pub(crate) started_at: std::string::String, + /// Campaign finish timestamp. + pub(crate) finished_at: std::string::String, +} + +pub(crate) struct DemoCoreExtractionObserver<'a> { + pub(crate) app_handle: tauri::AppHandle, + pub(crate) cancel_requested: &'a std::sync::atomic::AtomicBool, +} + +impl kb_pipeline::CoreExtractionObserver for crate::DemoCoreExtractionObserver<'_> { + fn on_progress(&self, event: &kb_pipeline::CoreExtractionProgressEvent) { + let payload = crate::DemoCoreExtractionProgressPayload { + timestamp: event.timestamp.clone(), + level: event.level.code().to_string(), + message: event.message.clone(), + completed: event.completed, + total: event.total, + }; + let emit_result = self.app_handle.emit_to( + "demo_core_extraction", + "demo-core-extraction-progress", + payload, + ); + if let std::result::Result::Err(error) = emit_result { + tracing::warn!( + target: crate::TRACING_TARGET, + "cannot emit core extraction progress: {error}" + ); + } + } + + fn is_cancelled(&self) -> bool { + return self.cancel_requested.load(std::sync::atomic::Ordering::Relaxed); + } +} + +pub(crate) struct DemoCoreExtractionRunGuard<'a> { + pub(crate) running: &'a std::sync::atomic::AtomicBool, +} + +impl std::ops::Drop for crate::DemoCoreExtractionRunGuard<'_> { + fn drop(&mut self) { + self.running.store(false, std::sync::atomic::Ordering::Release); + } +} + +pub(crate) fn build_demo_core_extraction_pipeline_request( + request: crate::DemoCoreExtractionRequest, +) -> std::result::Result { + let source_result = match request.mode.trim() { + "signatures" => { + let signatures = split_signatures(request.signatures_text.as_deref()); + std::result::Result::Ok(kb_pipeline::CoreExtractionSource::Signatures(signatures)) + }, + "pending" => std::result::Result::Ok(kb_pipeline::CoreExtractionSource::Pending), + "program_id" => { + let program_id = match request.program_id { + std::option::Option::Some(value) if !value.trim().is_empty() => value, + _ => { + return std::result::Result::Err( + "program id is required for program extraction".to_string(), + ); + }, + }; + std::result::Result::Ok(kb_pipeline::CoreExtractionSource::ProgramId { program_id }) + }, + "slot_range" => { + let min_slot = match request.min_slot { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err( + "minimum slot is required for slot range extraction".to_string(), + ); + }, + }; + let max_slot = match request.max_slot { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err( + "maximum slot is required for slot range extraction".to_string(), + ); + }, + }; + std::result::Result::Ok(kb_pipeline::CoreExtractionSource::SlotRange { + min_slot, + max_slot, + }) + }, + _ => std::result::Result::Err("unsupported core extraction mode".to_string()), + }; + let source = match source_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let pipeline_request = kb_pipeline::CoreExtractionRequest { + source, + limit: request.limit, + max_concurrent_extractions: request.max_concurrent_extractions, + force_replay: request.force_replay, + }; + let validation_result = pipeline_request.validate(); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error.to_string()); + } + return std::result::Result::Ok(pipeline_request); +} + +fn split_signatures(text: std::option::Option<&str>) -> std::vec::Vec { + let source = match text { + std::option::Option::Some(value) => value, + std::option::Option::None => "", + }; + let mut unique = std::collections::BTreeSet::::new(); + let mut output = std::vec::Vec::new(); + for line in source.lines() { + let signature = line.trim(); + if signature.is_empty() { + continue; + } + if unique.insert(signature.to_string()) { + output.push(signature.to_string()); + } + } + return output; +} + +pub(crate) fn demo_core_extraction_summary_payload( + summary: kb_pipeline::CoreExtractionSummary, +) -> crate::DemoCoreExtractionSummaryPayload { + return crate::DemoCoreExtractionSummaryPayload { + processor_version: summary.processor_version, + selected: summary.selected, + started: summary.started, + completed: summary.completed, + skipped: summary.skipped, + extracted: summary.extracted, + failed: summary.failed, + cancelled_candidates: summary.cancelled_candidates, + not_started: summary.not_started, + cancelled: summary.cancelled, + started_at: summary.started_at, + finished_at: summary.finished_at, + }; +} + +#[cfg(test)] +mod tests { + #[test] + fn signature_source_splits_deduplicates_and_ignores_empty_rows() { + let values = super::split_signatures(std::option::Option::Some("alpha\n\n beta \nalpha\n")); + assert_eq!(values, std::vec!["alpha".to_string(), "beta".to_string()]); + } +} diff --git a/kb-app-demo-desktop/src/demo_decode_replay.rs b/kb-app-demo-desktop/src/demo_decode_replay.rs new file mode 100644 index 0000000..912cbbd --- /dev/null +++ b/kb-app-demo-desktop/src/demo_decode_replay.rs @@ -0,0 +1,947 @@ +// file: kb-app-demo-desktop/src/demo_decode_replay.rs +// version: 24 + +//! Tauri commands and UI payloads for contextual instruction decode replay. + +use tauri::Emitter; // rust-rules: trait-import +use ts_rs::TS; // rust-rules: derive-import + +/// One decoder selectable by the decode replay demo. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayDecoderOption.ts" +)] +pub(crate) struct DemoDecodeReplayDecoderOption { + /// Stable decoder name. + pub(crate) name: std::string::String, + /// Stable decoder version. + pub(crate) version: std::string::String, + /// Exact supported program identifiers. + pub(crate) program_ids: std::vec::Vec, +} + +/// Initial options shown by the contextual decode replay demo. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayOptionsPayload.ts" +)] +pub(crate) struct DemoDecodeReplayOptionsPayload { + /// Common orchestration version. + pub(crate) pipeline_version: std::string::String, + /// Available contextual decoders. + pub(crate) decoders: std::vec::Vec, + /// Stable names of materializers currently registered by the demo. + pub(crate) materializer_names: std::vec::Vec, + /// Default bounded selection limit. + pub(crate) default_limit: u32, + /// Default maximum concurrent contextual inputs. + pub(crate) default_max_concurrent_inputs: u32, + /// Whether one replay campaign is running. + pub(crate) running: bool, +} + +/// UI request for one bounded contextual decode replay campaign. +#[derive(Clone, Debug, serde::Deserialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayRequest.ts" +)] +pub(crate) struct DemoDecodeReplayRequest { + /// Newline-separated exact transaction signatures. + pub(crate) signatures_text: std::option::Option, + /// Optional exact program identifier. + pub(crate) program_id: std::option::Option, + /// Instruction processing state code, actionable or incomplete_signatures. + pub(crate) instruction_state: std::string::String, + /// Newline-separated exact stable instruction paths. + pub(crate) instruction_paths_text: std::option::Option, + /// Explicit selected decoder names. + pub(crate) decoder_names: std::vec::Vec, + /// Maximum selected contextual inputs. + pub(crate) limit: u32, + /// Maximum concurrent contextual inputs. + pub(crate) max_concurrent_inputs: u32, + /// Whether every compatible decoder must run. + pub(crate) all_compatible: bool, + /// Replaces only processor-owned outputs for the same version and input. + pub(crate) force_replay: bool, + /// Explicitly authorizes a bounded force replay without exact signatures. + pub(crate) force_replay_all_matching: bool, + /// Runs compatible materializers after decoded event persistence. + pub(crate) materialize_after_decode: bool, +} + +/// One progress event emitted to the decode replay window. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayProgressPayload.ts" +)] +pub(crate) struct DemoDecodeReplayProgressPayload { + /// Stable process-local campaign identifier. + pub(crate) campaign_id: std::string::String, + /// RFC 3339 timestamp. + pub(crate) timestamp: std::string::String, + /// Stable severity code. + pub(crate) level: std::string::String, + /// Human-readable message. + pub(crate) message: std::string::String, + /// Number of terminal contextual inputs. + #[ts(type = "number")] + pub(crate) completed: u64, + /// Total selected contextual inputs. + #[ts(type = "number")] + pub(crate) total: u64, +} + +/// One processor counter row returned to the UI. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeProcessorSummaryPayload.ts" +)] +pub(crate) struct DemoDecodeProcessorSummaryPayload { + /// Stable processor name. + pub(crate) processor_name: std::string::String, + /// Stable processor version. + pub(crate) processor_version: std::string::String, + /// Compatible dispatch count. + #[ts(type = "number")] + pub(crate) dispatched: u64, + /// Version and hash skip count. + #[ts(type = "number")] + pub(crate) skipped: u64, + /// Decoded input count. + #[ts(type = "number")] + pub(crate) decoded: u64, + /// Ignored input count. + #[ts(type = "number")] + pub(crate) ignored: u64, + /// Unsupported input count. + #[ts(type = "number")] + pub(crate) unsupported: u64, + /// Failed input count. + #[ts(type = "number")] + pub(crate) failed: u64, + /// Materialized output count. + #[ts(type = "number")] + pub(crate) materialized_outputs: u64, + /// Materialization refusal count. + #[ts(type = "number")] + pub(crate) materialization_refused: u64, +} + +/// Final UI-safe summary for one contextual decode replay campaign. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplaySummaryPayload.ts" +)] +pub(crate) struct DemoDecodeReplaySummaryPayload { + /// Stable process-local campaign identifier. + pub(crate) campaign_id: std::string::String, + /// Common pipeline version. + pub(crate) pipeline_version: std::string::String, + /// Number of selected contextual inputs. + #[ts(type = "number")] + pub(crate) selected: u64, + /// Number admitted to execution. + #[ts(type = "number")] + pub(crate) started: u64, + /// Number reaching a terminal state. + #[ts(type = "number")] + pub(crate) completed: u64, + /// Number with no compatible enabled decoder. + #[ts(type = "number")] + pub(crate) unmatched: u64, + /// Number never started after cancellation. + #[ts(type = "number")] + pub(crate) not_started: u64, + /// Number of failed contextual inputs. + #[ts(type = "number")] + pub(crate) failed_inputs: u64, + /// Whether cancellation was observed. + pub(crate) cancelled: bool, + /// Per-processor counters. + pub(crate) processors: std::vec::Vec, + /// Campaign start timestamp. + pub(crate) started_at: std::string::String, + /// Campaign finish timestamp. + pub(crate) finished_at: std::string::String, +} + +/// One aggregated coverage row returned to the UI. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeCoverageSummaryPayload.ts" +)] +pub(crate) struct DemoDecodeCoverageSummaryPayload { + /// Stable processor name. + pub(crate) processor_name: std::string::String, + /// Stable processor version. + pub(crate) processor_version: std::string::String, + /// Exact program identifier. + pub(crate) program_id: std::string::String, + /// Optional stable surface code. + pub(crate) surface_code: std::option::Option, + /// Stable entry classifier. + pub(crate) entry_code: std::string::String, + /// Declared count. + #[ts(type = "number")] + pub(crate) declared_count: i64, + /// Observed count. + #[ts(type = "number")] + pub(crate) observed_count: i64, + /// Recognized count. + #[ts(type = "number")] + pub(crate) recognized_count: i64, + /// Decoded observation count. + #[ts(type = "number")] + pub(crate) decoded_count: i64, + /// Materialized output count. + #[ts(type = "number")] + pub(crate) materialized_count: i64, + /// Error count. + #[ts(type = "number")] + pub(crate) error_count: i64, + /// Unknown or unsupported observation count. + #[ts(type = "number")] + pub(crate) unknown_count: i64, + /// Successful source transaction count. + #[ts(type = "number")] + pub(crate) successful_transaction_count: i64, + /// Failed source transaction count. + #[ts(type = "number")] + pub(crate) failed_transaction_count: i64, +} + +/// Read-only decode, ledger and coverage diagnostics. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeDiagnosticsPayload.ts" +)] +pub(crate) struct DemoDecodeDiagnosticsPayload { + /// Decode, materialization and ledger table diagnostics. + pub(crate) tables: std::vec::Vec, + /// Aggregated declared and observed coverage. + pub(crate) coverage: std::vec::Vec, +} + +/// Bounded read request for committed transaction annotations. +#[derive(Clone, Debug, serde::Deserialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoTransactionAnnotationRequest.ts" +)] +pub(crate) struct DemoTransactionAnnotationRequest { + /// Optional partial transaction signature. + pub(crate) signature_contains: std::option::Option, + /// Maximum returned rows. + pub(crate) limit: u32, +} + +/// UI-safe committed transaction annotation row. +#[derive(Clone, Debug, serde::Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts( + export, + export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoTransactionAnnotationRow.ts" +)] +pub(crate) struct DemoTransactionAnnotationRow { + /// Materializer processor version. + pub(crate) processor_version: std::string::String, + /// Source decoder and version. + pub(crate) decoder: std::string::String, + /// Transaction signature. + pub(crate) signature: std::string::String, + /// Decimal slot rendered as text to preserve JSON precision. + pub(crate) slot: std::string::String, + /// Stable outer or inner instruction path. + pub(crate) instruction_path: std::string::String, + /// Exact Memo generation. + pub(crate) generation: std::string::String, + /// Exact Memo Program ID. + pub(crate) program_id: std::string::String, + /// Complete bounded UTF-8 Memo text. + pub(crate) text: std::string::String, + /// Exact payload byte length. + pub(crate) payload_length_bytes: u32, + /// Canonical payload SHA-256. + pub(crate) payload_sha256: std::string::String, + /// Runtime-verified ordered signer keys. + pub(crate) verified_signers: std::vec::Vec, + /// Stable materializer idempotence key. + pub(crate) idempotence_key: std::string::String, + /// Database creation timestamp. + pub(crate) created_at: std::string::String, + /// Database replacement timestamp. + pub(crate) updated_at: std::string::String, +} + +pub(crate) struct DemoDecodeReplayObserver<'a> { + pub(crate) app_handle: tauri::AppHandle, + pub(crate) campaign_id: std::string::String, + pub(crate) cancel_requested: &'a std::sync::atomic::AtomicBool, +} + +impl kb_pipeline::DecodeReplayObserver for crate::DemoDecodeReplayObserver<'_> { + fn on_progress(&self, event: &kb_pipeline::DecodeReplayProgressEvent) { + let payload = DemoDecodeReplayProgressPayload { + campaign_id: self.campaign_id.clone(), + timestamp: event.timestamp.clone(), + level: event.level.code().to_string(), + message: event.message.clone(), + completed: event.completed, + total: event.total, + }; + let emit_result = + self.app_handle + .emit_to("demo_decode_replay", "demo-decode-replay-progress", payload); + if let std::result::Result::Err(error) = emit_result { + tracing::warn!(target: crate::TRACING_TARGET, action = "emit_progress", campaign_id = %self.campaign_id, error = %error, "cannot emit decode replay progress"); + } + } + + fn is_cancelled(&self) -> bool { + return self.cancel_requested.load(std::sync::atomic::Ordering::Relaxed); + } +} + +pub(crate) struct DemoDecodeReplayRunGuard<'a> { + pub(crate) running: &'a std::sync::atomic::AtomicBool, + pub(crate) campaign_id: &'a std::sync::Mutex>, +} + +impl std::ops::Drop for crate::DemoDecodeReplayRunGuard<'_> { + fn drop(&mut self) { + self.running.store(false, std::sync::atomic::Ordering::Release); + let lock_result = self.campaign_id.lock(); + if let std::result::Result::Ok(mut active_campaign_id) = lock_result { + *active_campaign_id = std::option::Option::None; + } + } +} + +pub(crate) fn register_active_campaign( + campaign_slot: &std::sync::Mutex>, + campaign_id: &str, +) -> std::result::Result<(), std::string::String> { + let lock_result = campaign_slot.lock(); + let mut active_campaign_id = match lock_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => { + return std::result::Result::Err( + "cannot register active contextual decode campaign".to_string(), + ); + }, + }; + *active_campaign_id = std::option::Option::Some(campaign_id.to_string()); + return std::result::Result::Ok(()); +} + +pub(crate) fn available_materializers() +-> std::vec::Vec> { + return std::vec![ + std::sync::Arc::new(kb_lib::MtAdminMaterializer), + std::sync::Arc::new(kb_lib::MtComplianceAuditMaterializer), + std::sync::Arc::new(kb_lib::MtFeesMaterializer), + std::sync::Arc::new(kb_lib::MtLifecycleMaterializer), + std::sync::Arc::new(kb_lib::MtStakingMaterializer), + std::sync::Arc::new(kb_lib::MtTokenAccountsMaterializer), + std::sync::Arc::new(kb_lib::MtRiskMaterializer), + std::sync::Arc::new(kb_lib::MtTransactionAnnotationMaterializer,), + ]; +} + +pub(crate) fn available_decoders() +-> std::vec::Vec> { + return std::vec![ + std::sync::Arc::new(kb_lib::DcSolanaCoreDecoder), + std::sync::Arc::new(kb_lib::DcSplAssociatedTokenAccountDecoder,), + std::sync::Arc::new(kb_lib::DcSplElgamalRegistryDecoder), + std::sync::Arc::new(kb_lib::DcSplMemoDecoder), + std::sync::Arc::new(kb_lib::DcSplTokenDecoder), + std::sync::Arc::new(kb_lib::DcSplToken2022Decoder), + ]; +} + +pub(crate) fn build_demo_decode_replay_pipeline_request( + request: crate::DemoDecodeReplayRequest, + campaign_id: std::string::String, +) -> std::result::Result { + let states_result = processing_states(request.instruction_state.as_str()); + let states = match states_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let signatures = split_lines(request.signatures_text.as_deref()); + let instruction_paths = split_lines(request.instruction_paths_text.as_deref()); + let incomplete_signatures = request.instruction_state.trim() == "incomplete_signatures"; + let program_ids = match request.program_id { + std::option::Option::Some(value) if !value.trim().is_empty() => { + std::vec![value.trim().to_string()] + }, + _ => std::vec::Vec::new(), + }; + let selection_result = kb_store::DecodeSelectionFilter::new( + signatures, + states, + std::option::Option::None, + std::option::Option::None, + program_ids, + instruction_paths, + incomplete_signatures, + request.limit, + ); + let selection = match selection_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + let dispatch_policy = if request.all_compatible { + kb_pipeline::DecodeDispatchPolicy::AllCompatible + } else { + kb_pipeline::DecodeDispatchPolicy::HighestPriority + }; + let pipeline_request = kb_pipeline::DecodeReplayRequest { + campaign_id, + selection, + decoder_names: request.decoder_names, + dispatch_policy, + max_concurrent_inputs: request.max_concurrent_inputs, + force_replay: request.force_replay, + force_replay_all_matching: request.force_replay_all_matching, + materialize_after_decode: request.materialize_after_decode, + }; + let validation_result = pipeline_request.validate(); + if let std::result::Result::Err(error) = validation_result { + return std::result::Result::Err(error.to_string()); + } + tracing::debug!( + target: crate::TRACING_TARGET, + action = "build_request", + campaign_id = %pipeline_request.campaign_id, + signature_count = pipeline_request.selection.signatures.len(), + signature_sample = ?text_sample(pipeline_request.selection.signatures.as_slice(), 5), + processing_states = ?pipeline_request.selection.processing_states, + program_ids = ?pipeline_request.selection.program_ids, + instruction_paths = ?pipeline_request.selection.instruction_paths, + incomplete_signatures = pipeline_request.selection.incomplete_signatures, + limit = pipeline_request.selection.limit, + decoder_names = ?pipeline_request.decoder_names, + dispatch_policy = ?pipeline_request.dispatch_policy, + max_concurrent_inputs = pipeline_request.max_concurrent_inputs, + force_replay = pipeline_request.force_replay, + force_replay_all_matching = pipeline_request.force_replay_all_matching, + materialize_after_decode = pipeline_request.materialize_after_decode, + "built normalized contextual decode replay request" + ); + return std::result::Result::Ok(pipeline_request); +} + +pub(crate) fn optional_line_count(value: &std::option::Option) -> usize { + return match value { + std::option::Option::Some(text) => { + text.lines().filter(|line| return !line.trim().is_empty()).count() + }, + std::option::Option::None => 0, + }; +} + +pub(crate) fn text_sample(values: &[std::string::String], limit: usize) -> std::vec::Vec<&str> { + return values.iter().take(limit).map(std::string::String::as_str).collect(); +} + +fn processing_states( + value: &str, +) -> std::result::Result, std::string::String> +{ + return match value.trim() { + "incomplete_signatures" | "actionable" => std::result::Result::Ok(std::vec![ + kb_store::CoreInstructionProcessingState::Pending, + kb_store::CoreInstructionProcessingState::Failed, + kb_store::CoreInstructionProcessingState::ReplayRequested, + ]), + "pending" => { + std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Pending]) + }, + "failed" => { + std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Failed]) + }, + "replay_requested" => std::result::Result::Ok(std::vec![ + kb_store::CoreInstructionProcessingState::ReplayRequested + ]), + "decoded" => { + std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Decoded]) + }, + "ignored" => { + std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Ignored]) + }, + "materialized" => std::result::Result::Ok(std::vec![ + kb_store::CoreInstructionProcessingState::Materialized + ]), + _ => std::result::Result::Err("unsupported instruction processing state".to_string()), + }; +} + +fn split_lines(text: std::option::Option<&str>) -> std::vec::Vec { + let source = match text { + std::option::Option::Some(value) => value, + std::option::Option::None => "", + }; + let mut unique = std::collections::BTreeSet::::new(); + let mut output = std::vec::Vec::new(); + for line in source.lines() { + let value = line.trim(); + if value.is_empty() { + continue; + } + if unique.insert(value.to_string()) { + output.push(value.to_string()); + } + } + return output; +} + +pub(crate) fn demo_decode_replay_summary_payload( + summary: kb_pipeline::DecodeReplaySummary, +) -> crate::DemoDecodeReplaySummaryPayload { + return crate::DemoDecodeReplaySummaryPayload { + campaign_id: summary.campaign_id, + pipeline_version: summary.pipeline_version, + selected: summary.selected, + started: summary.started, + completed: summary.completed, + unmatched: summary.unmatched, + not_started: summary.not_started, + failed_inputs: summary.failed_inputs, + cancelled: summary.cancelled, + processors: summary + .processors + .into_iter() + .map(|processor| { + return crate::DemoDecodeProcessorSummaryPayload { + processor_name: processor.processor_name, + processor_version: processor.processor_version, + dispatched: processor.dispatched, + skipped: processor.skipped, + decoded: processor.decoded, + ignored: processor.ignored, + unsupported: processor.unsupported, + failed: processor.failed, + materialized_outputs: processor.materialized_outputs, + materialization_refused: processor.materialization_refused, + }; + }) + .collect(), + started_at: summary.started_at, + finished_at: summary.finished_at, + }; +} + +pub(crate) fn coverage_payload( + value: kb_store::DecodeCoverageSummaryRow, +) -> crate::DemoDecodeCoverageSummaryPayload { + return crate::DemoDecodeCoverageSummaryPayload { + processor_name: value.processor_name, + processor_version: value.processor_version, + program_id: value.program_id, + surface_code: value.surface_code, + entry_code: value.entry_code, + declared_count: value.declared_count, + observed_count: value.observed_count, + recognized_count: value.recognized_count, + decoded_count: value.decoded_count, + materialized_count: value.materialized_count, + error_count: value.error_count, + unknown_count: value.unknown_count, + successful_transaction_count: value.successful_transaction_count, + failed_transaction_count: value.failed_transaction_count, + }; +} + +pub(crate) fn annotation_payload( + row: kb_store::MaterializedEventQueryRow, +) -> std::result::Result { + if row.processor_name != "transaction_annotations" + || row.materialized_family != "transaction_annotation" + { + return std::result::Result::Err( + "materialized row is not a transaction annotation".to_string(), + ); + } + let payload = &row.payload_json; + let byte_length = match payload.get("payloadLengthBytes").and_then(serde_json::Value::as_u64) { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err( + "transaction annotation payloadLengthBytes is absent".to_string(), + ); + }, + }; + let byte_length_result = u32::try_from(byte_length); + let payload_length_bytes = match byte_length_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + return std::result::Result::Err(format!( + "transaction annotation payloadLengthBytes is not UI-safe: {error}" + )); + }, + }; + let signers = match payload.get("signers") { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err( + "transaction annotation signers are absent".to_string(), + ); + }, + }; + let verified_values = match signers.get("verified").and_then(serde_json::Value::as_array) { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err( + "transaction annotation verified signers are invalid".to_string(), + ); + }, + }; + let mut verified_signers = std::vec::Vec::with_capacity(verified_values.len()); + for value in verified_values { + match value.as_str() { + std::option::Option::Some(signer) if !signer.trim().is_empty() => { + verified_signers.push(signer.to_string()); + }, + _ => { + return std::result::Result::Err( + "transaction annotation verified signer is invalid".to_string(), + ); + }, + } + } + let instruction_path = match required_annotation_text(payload, "instructionPath") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let generation = match required_annotation_text(payload, "generation") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let program_id = match required_annotation_text(payload, "programId") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let text = match required_annotation_text(payload, "text") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let payload_sha256 = match required_annotation_text(payload, "payloadSha256") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let idempotence_key = match required_annotation_text(payload, "idempotenceKey") { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + return std::result::Result::Ok(crate::DemoTransactionAnnotationRow { + processor_version: row.processor_version, + decoder: format!("{}@{}", row.source_decoder_name, row.source_decoder_version), + signature: row.signature, + slot: row.slot.to_string(), + instruction_path, + generation, + program_id, + text, + payload_length_bytes, + payload_sha256, + verified_signers, + idempotence_key, + created_at: row.created_at, + updated_at: row.updated_at, + }); +} + +fn required_annotation_text( + payload: &serde_json::Value, + field: &str, +) -> std::result::Result { + return match payload.get(field).and_then(serde_json::Value::as_str) { + std::option::Option::Some(value) => std::result::Result::Ok(value.to_string()), + std::option::Option::None => { + std::result::Result::Err(format!("transaction annotation {field} is absent")) + }, + }; +} + +#[cfg(test)] +mod tests { + fn ata_observation(entry: &str) -> kb_lib::DcApiDecodedObservation { + return kb_lib::DcApiDecodedObservation { + event_key: format!("ata:{entry}:0"), + event: kb_lib::MdDecodedProtocolEvent { + signature: kb_lib::MdSignature("signature".to_string()), + slot: kb_lib::MdSlot(1), + instruction_path: kb_lib::MdInstructionPath("0".to_string()), + program_id: kb_lib::MdProgramId( + kb_program_ids::ASSOCIATED_TOKEN_PROGRAM_ID.to_string(), + ), + protocol_code: kb_lib::MdProtocolCode("spl_associated_token_account".to_string()), + surface_code: kb_lib::MdSurfaceCode("spl_associated_token_account".to_string()), + event_code: kb_lib::MdEventCode(format!("spl_associated_token_account.{entry}")), + event_name: kb_lib::MdEventName(entry.to_string()), + event_family: kb_lib::MdEventFamily::Lifecycle, + source_kind: kb_lib::MdEventSourceKind::Instruction, + confidence: kb_lib::MdDecoderConfidence::ManualExact, + }, + payload_json: serde_json::json!({}), + transaction_failed: false, + transaction_error: std::option::Option::None, + observation_committed: true, + proof: kb_lib::DcApiDecoderProof { + kind: kb_lib::DcApiDecoderProofKind::Manual, + confidence: kb_lib::MdDecoderConfidence::ManualExact, + evidence: std::vec!["fixture".to_string()], + }, + }; + } + + #[test] + fn native_memo_token_2022_elgamal_classic_token_and_ata_decoders_are_registered() { + let decoders = crate::available_decoders(); + assert_eq!(decoders.len(), 6); + let names = decoders + .iter() + .map(|decoder| return decoder.identity().name) + .collect::>(); + assert_eq!( + names, + std::vec![ + "solana_native_classifier".to_string(), + "spl_associated_token_account".to_string(), + "spl_elgamal_registry".to_string(), + "spl_memo".to_string(), + "spl_token".to_string(), + "spl_token_2022".to_string(), + ] + ); + } + + #[test] + fn runtime_materializer_registry_is_complete_for_current_instructional_surfaces() { + let materializers = crate::available_materializers(); + assert_eq!(materializers.len(), 8); + let names = materializers + .iter() + .map(|materializer| return materializer.identity().name) + .collect::>(); + assert_eq!( + names, + std::vec![ + "solana_native_admin".to_string(), + "solana_native_compliance_audit".to_string(), + "fees".to_string(), + "solana_native_lifecycle".to_string(), + "solana_native_staking".to_string(), + "spl_token_accounts".to_string(), + "spl_token_risk".to_string(), + "transaction_annotations".to_string(), + ] + ); + } + + #[test] + fn ata_materializer_ownership_is_exact_in_the_runtime_registry() { + for (entry, expected) in [ + ("create", std::vec!["spl_token_accounts".to_string()]), + ( + "recover_nested", + std::vec!["spl_token_accounts".to_string(), "spl_token_risk".to_string(),], + ), + ] { + let observation = ata_observation(entry); + let owners = crate::available_materializers() + .iter() + .filter(|materializer| return materializer.accepts_observation(&observation)) + .map(|materializer| return materializer.identity().name) + .collect::>(); + assert_eq!(owners, expected); + } + } + + #[test] + fn line_selection_is_trimmed_and_deduplicated() { + let values = super::split_lines(std::option::Option::Some(" a \n\na\nb\n")); + assert_eq!(values, std::vec!["a".to_string(), "b".to_string()]); + } + + fn request( + signatures_text: std::option::Option<&str>, + force_replay: bool, + force_replay_all_matching: bool, + materialize_after_decode: bool, + ) -> crate::DemoDecodeReplayRequest { + return crate::DemoDecodeReplayRequest { + signatures_text: signatures_text.map(|value| return value.to_string()), + program_id: std::option::Option::None, + instruction_state: "actionable".to_string(), + instruction_paths_text: std::option::Option::None, + decoder_names: std::vec!["solana_native_classifier".to_string()], + limit: 10, + max_concurrent_inputs: 2, + all_compatible: false, + force_replay, + force_replay_all_matching, + materialize_after_decode, + }; + } + + #[test] + fn force_replay_requires_signatures_or_all_matching_authorization() { + let result = crate::build_demo_decode_replay_pipeline_request( + request(std::option::Option::None, true, false, false), + "decode-test".to_string(), + ); + assert!(result.is_err()); + } + + #[test] + fn force_replay_all_matching_is_forwarded() { + let result = crate::build_demo_decode_replay_pipeline_request( + request(std::option::Option::None, true, true, false), + "decode-test".to_string(), + ); + let request = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("request mapping failed: {error}"), + }; + assert!(request.force_replay_all_matching); + assert!(request.selection.signatures.is_empty()); + } + + #[test] + fn run_guard_clears_active_campaign_state() { + let running = std::sync::atomic::AtomicBool::new(true); + let campaign_id = + std::sync::Mutex::new(std::option::Option::Some("decode-test".to_string())); + { + let _guard = crate::DemoDecodeReplayRunGuard { + running: &running, + campaign_id: &campaign_id, + }; + } + assert!(!running.load(std::sync::atomic::Ordering::Acquire)); + let lock_result = campaign_id.lock(); + let active_campaign_id = match lock_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => panic!("campaign id mutex must not be poisoned"), + }; + assert!(active_campaign_id.is_none()); + } + + #[test] + fn register_active_campaign_sets_slot_without_exposing_guard() { + let campaign_id = std::sync::Mutex::new(std::option::Option::None); + let result = crate::register_active_campaign(&campaign_id, "decode-test"); + assert!(result.is_ok()); + let lock_result = campaign_id.lock(); + let active_campaign_id = match lock_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => panic!("campaign id mutex must not be poisoned"), + }; + assert_eq!(active_campaign_id.as_deref(), std::option::Option::Some("decode-test")); + } + + #[test] + fn incomplete_signature_state_uses_actionable_states_and_expansion() { + let mut request = request(std::option::Option::None, false, false, true); + request.instruction_state = "incomplete_signatures".to_string(); + let result = crate::build_demo_decode_replay_pipeline_request( + request, + "decode-incomplete".to_string(), + ); + let pipeline_request = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("request mapping failed: {error}"), + }; + assert!(pipeline_request.selection.incomplete_signatures); + assert_eq!(pipeline_request.selection.processing_states.len(), 3); + assert!(!pipeline_request.force_replay); + assert!(pipeline_request.materialize_after_decode); + } + + #[test] + fn actionable_state_contains_pending_failed_and_replay_requested() { + let result = super::processing_states("actionable"); + let states = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("state mapping failed: {error}"), + }; + assert_eq!(states.len(), 3); + } + + #[test] + fn committed_annotation_row_is_mapped_to_ui_safe_contract() { + let row = kb_store::MaterializedEventQueryRow { + processor_name: "transaction_annotations".to_string(), + processor_version: "0.4.3".to_string(), + input_key: "input".to_string(), + output_key: "output".to_string(), + source_event_key: "memo:0".to_string(), + source_decoder_name: "spl_memo".to_string(), + source_decoder_version: "0.4.3".to_string(), + signature: "signature".to_string(), + slot: u64::MAX, + materialized_family: "transaction_annotation".to_string(), + payload_json: serde_json::json!({ + "instructionPath": "1/0", + "generation": "v4", + "programId": kb_program_ids::SPL_MEMO_V4_PROGRAM_ID, + "text": "annotation", + "payloadLengthBytes": 10, + "payloadSha256": "11", + "signers": {"verified": ["signer"]}, + "idempotenceKey": "stable" + }), + created_at: "created".to_string(), + updated_at: "updated".to_string(), + }; + let result = crate::annotation_payload(row); + let mapped = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => panic!("annotation mapping failed: {error}"), + }; + assert_eq!(mapped.slot, u64::MAX.to_string()); + assert_eq!(mapped.text, "annotation"); + assert_eq!(mapped.verified_signers, std::vec!["signer".to_string()]); + } + + #[test] + fn malformed_annotation_payload_fails_closed() { + let row = kb_store::MaterializedEventQueryRow { + processor_name: "transaction_annotations".to_string(), + processor_version: "0.4.3".to_string(), + input_key: "input".to_string(), + output_key: "output".to_string(), + source_event_key: "memo:0".to_string(), + source_decoder_name: "spl_memo".to_string(), + source_decoder_version: "0.4.3".to_string(), + signature: "signature".to_string(), + slot: 1, + materialized_family: "transaction_annotation".to_string(), + payload_json: serde_json::json!({}), + created_at: "created".to_string(), + updated_at: "updated".to_string(), + }; + assert!(crate::annotation_payload(row).is_err()); + } +} diff --git a/kb-app-demo-desktop/src/demo_sql_common.rs b/kb-app-demo-desktop/src/demo_sql_common.rs index cdab135..b72d833 100644 --- a/kb-app-demo-desktop/src/demo_sql_common.rs +++ b/kb-app-demo-desktop/src/demo_sql_common.rs @@ -87,7 +87,6 @@ pub(crate) fn open_sql_demo_window( }; } - /// Builds PostgreSQL options from the active profile without coupling kb-store to kb-config. pub(crate) fn postgres_store_options_from_profile( profile: &kb_config::ProfileConfig, diff --git a/kb-app-demo-desktop/src/demo_sql_replay_candidates.rs b/kb-app-demo-desktop/src/demo_sql_replay_candidates.rs index dddd6af..bf56ab2 100644 --- a/kb-app-demo-desktop/src/demo_sql_replay_candidates.rs +++ b/kb-app-demo-desktop/src/demo_sql_replay_candidates.rs @@ -326,10 +326,8 @@ pub(crate) fn program_scope_from_code( pub(crate) fn optional_entity_kind_from_code( code: std::option::Option<&str>, -) -> std::result::Result< - std::option::Option, - std::string::String, -> { +) -> std::result::Result, std::string::String> +{ return match code { std::option::Option::Some(value) => { let result = entity_kind_from_code(value); diff --git a/kb-app-demo-desktop/src/demo_ws.rs b/kb-app-demo-desktop/src/demo_ws.rs index 3bb6f70..7ebea29 100644 --- a/kb-app-demo-desktop/src/demo_ws.rs +++ b/kb-app-demo-desktop/src/demo_ws.rs @@ -338,13 +338,19 @@ async fn ensure_demo_ws_session( } else { kb_onchain_transport::WsReconnectPolicy::disabled() }; - let capabilities = - kb_onchain_transport::StandardWsCapabilities::from_endpoint(selected_client.endpoint_config()); - let session = - match kb_onchain_transport::WsSession::connect(selected_client, capabilities, reconnect_policy).await { - std::result::Result::Ok(session) => std::sync::Arc::new(session), - std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), - }; + let capabilities = kb_onchain_transport::StandardWsCapabilities::from_endpoint( + selected_client.endpoint_config(), + ); + let session = match kb_onchain_transport::WsSession::connect( + selected_client, + capabilities, + reconnect_policy, + ) + .await + { + std::result::Result::Ok(session) => std::sync::Arc::new(session), + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; spawn_demo_ws_event_bridge(app_handle.clone(), session.clone()); { let mut guard = state.demo_ws_session().lock().await; @@ -442,7 +448,10 @@ fn spawn_demo_ws_event_bridge( kb_onchain_transport::WsSessionEvent::Diagnostic { code, message } => { emit_demo_ws_message(&app_handle, &code, truncate_payload(message)); }, - kb_onchain_transport::WsSessionEvent::Reconnecting { attempt, maximum_attempts } => { + kb_onchain_transport::WsSessionEvent::Reconnecting { + attempt, + maximum_attempts, + } => { emit_demo_ws_message( &app_handle, "reconnecting", @@ -473,7 +482,10 @@ fn spawn_demo_ws_event_bridge( }); } -fn emit_demo_ws_status(app_handle: &tauri::AppHandle, snapshot: kb_onchain_transport::WsSessionSnapshot) { +fn emit_demo_ws_status( + app_handle: &tauri::AppHandle, + snapshot: kb_onchain_transport::WsSessionSnapshot, +) { let window = match app_handle.get_webview_window("demo_ws") { std::option::Option::Some(window) => window, std::option::Option::None => return, @@ -501,7 +513,9 @@ fn emit_demo_ws_message( } } -fn status_from_snapshot(snapshot: kb_onchain_transport::WsSessionSnapshot) -> crate::DemoWsStatusPayload { +fn status_from_snapshot( + snapshot: kb_onchain_transport::WsSessionSnapshot, +) -> crate::DemoWsStatusPayload { let mut subscriptions = std::vec::Vec::new(); for subscription in &snapshot.subscriptions { if let std::option::Option::Some(remote_subscription_id) = @@ -602,7 +616,9 @@ pub(crate) fn build_ws_method_options() -> std::vec::Vec target, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - let config = match parse_optional_json_as::( - config_json, - "configJson", - ) { + let config = match parse_optional_json_as::< + kb_onchain_transport::WsAccountSubscribeConfig, + >(config_json, "configJson") + { std::result::Result::Ok(config) => config, std::result::Result::Err(error) => return std::result::Result::Err(error), }; @@ -644,11 +660,13 @@ fn build_standard_ws_request( if let std::result::Result::Err(error) = reject_present(target, "target", method) { return std::result::Result::Err(error); } - let filter = - match parse_required_json_as::(filter_json, "filterJson") { - std::result::Result::Ok(filter) => filter, - std::result::Result::Err(error) => return std::result::Result::Err(error), - }; + let filter = match parse_required_json_as::( + filter_json, + "filterJson", + ) { + std::result::Result::Ok(filter) => filter, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; let config = match parse_optional_json_as::( config_json, "configJson", @@ -664,11 +682,13 @@ fn build_standard_ws_request( if let std::result::Result::Err(error) = reject_present(target, "target", method) { return std::result::Result::Err(error); } - let filter = - match parse_required_json_as::(filter_json, "filterJson") { - std::result::Result::Ok(filter) => filter, - std::result::Result::Err(error) => return std::result::Result::Err(error), - }; + let filter = match parse_required_json_as::( + filter_json, + "filterJson", + ) { + std::result::Result::Ok(filter) => filter, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; let config = match parse_optional_json_as::( config_json, "configJson", @@ -676,10 +696,9 @@ fn build_standard_ws_request( std::result::Result::Ok(config) => config, std::result::Result::Err(error) => return std::result::Result::Err(error), }; - std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Logs(kb_onchain_transport::LogsSubscribeRequest { - filter, - config, - })) + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Logs( + kb_onchain_transport::LogsSubscribeRequest { filter, config }, + )) }, "programSubscribe" => { let target = match required_target(target, method) { @@ -691,10 +710,10 @@ fn build_standard_ws_request( { return std::result::Result::Err(error); } - let config = match parse_optional_json_as::( - config_json, - "configJson", - ) { + let config = match parse_optional_json_as::< + kb_onchain_transport::WsProgramSubscribeConfig, + >(config_json, "configJson") + { std::result::Result::Ok(config) => config, std::result::Result::Err(error) => return std::result::Result::Err(error), }; @@ -712,10 +731,10 @@ fn build_standard_ws_request( { return std::result::Result::Err(error); } - let config = match parse_optional_json_as::( - config_json, - "configJson", - ) { + let config = match parse_optional_json_as::< + kb_onchain_transport::WsSignatureSubscribeConfig, + >(config_json, "configJson") + { std::result::Result::Ok(config) => config, std::result::Result::Err(error) => return std::result::Result::Err(error), }; @@ -729,7 +748,9 @@ fn build_standard_ws_request( { return std::result::Result::Err(error); } - std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Root(kb_onchain_transport::RootSubscribeRequest)) + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Root( + kb_onchain_transport::RootSubscribeRequest, + )) }, "slotSubscribe" => { if let std::result::Result::Err(error) = @@ -737,7 +758,9 @@ fn build_standard_ws_request( { return std::result::Result::Err(error); } - std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Slot(kb_onchain_transport::SlotSubscribeRequest)) + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Slot( + kb_onchain_transport::SlotSubscribeRequest, + )) }, "slotsUpdatesSubscribe" => { if let std::result::Result::Err(error) = @@ -755,7 +778,9 @@ fn build_standard_ws_request( { return std::result::Result::Err(error); } - std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Vote(kb_onchain_transport::VoteSubscribeRequest)) + std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Vote( + kb_onchain_transport::VoteSubscribeRequest, + )) }, _ => std::result::Result::Err(format!("unsupported standard WebSocket method '{method}'")), }; diff --git a/kb-app-demo-desktop/src/frontend_log.rs b/kb-app-demo-desktop/src/frontend_log.rs index cded103..20a4d9b 100644 --- a/kb-app-demo-desktop/src/frontend_log.rs +++ b/kb-app-demo-desktop/src/frontend_log.rs @@ -1,5 +1,5 @@ // file: kb-app-demo-desktop/src/frontend_log.rs -// version: 3 +// version: 4 //! Frontend logging bridge used by Tauri WebView scripts. @@ -23,11 +23,21 @@ pub(crate) struct FrontendLogPayload { pub(crate) fn emit_frontend_log(payload: crate::FrontendLogPayload) { let target = normalize_frontend_target(payload.target.as_str()); match payload.level.trim().to_ascii_lowercase().as_str() { - "trace" => tracing::trace!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message), - "debug" => tracing::debug!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message), - "warn" => tracing::warn!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message), - "error" => tracing::error!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message), - _ => tracing::info!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message), + "trace" => { + tracing::trace!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message) + }, + "debug" => { + tracing::debug!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message) + }, + "warn" => { + tracing::warn!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message) + }, + "error" => { + tracing::error!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message) + }, + _ => { + tracing::info!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message) + }, } } @@ -48,6 +58,12 @@ fn normalize_frontend_target(target: &str) -> std::string::String { if trimmed == "kb-app-demo-desktop.frontend.demo_ws" { return trimmed.to_string(); } + if trimmed == "kb-app-demo-desktop.frontend.demo_core_extraction" { + return trimmed.to_string(); + } + if trimmed == "kb-app-demo-desktop.frontend.demo_decode_replay" { + return trimmed.to_string(); + } return "kb-app-demo-desktop.frontend".to_string(); } @@ -82,6 +98,13 @@ mod tests { super::normalize_frontend_target("kb-app-demo-desktop.frontend.demo_ws"), "kb-app-demo-desktop.frontend.demo_ws" ); + assert_eq!( + super::normalize_frontend_target("kb-app-demo-desktop.frontend.demo_core_extraction"), + "kb-app-demo-desktop.frontend.demo_core_extraction" + ); + assert_eq!( + super::normalize_frontend_target("kb-app-demo-desktop.frontend.demo_decode_replay"), + "kb-app-demo-desktop.frontend.demo_decode_replay" + ); } - } diff --git a/kb-app-demo-desktop/src/lib.rs b/kb-app-demo-desktop/src/lib.rs index 658f051..ad6bfe0 100644 --- a/kb-app-demo-desktop/src/lib.rs +++ b/kb-app-demo-desktop/src/lib.rs @@ -1,5 +1,5 @@ // file: kb-app-demo-desktop/src/lib.rs -// version: 6 +// version: 7 //! Tauri desktop demo application for `khadhroony-bot3`. @@ -11,13 +11,15 @@ mod app_state; mod constants; mod demo_backfill; mod demo_config; +mod demo_core_extraction; +mod demo_decode_replay; mod demo_http; -mod demo_ws; mod demo_sql_common; mod demo_sql_diag; mod demo_sql_pg_core; mod demo_sql_pg_raw; mod demo_sql_replay_candidates; +mod demo_ws; mod frontend_log; mod main_window; mod splash; @@ -68,36 +70,6 @@ pub(crate) use self::demo_http::build_http_method_options; pub(crate) use self::demo_http::build_http_role_options; /// Executes one raw HTTP JSON-RPC request. pub(crate) use self::demo_http::demo_http_execute_request_inner; -/// WebSocket execution response payload. -pub(crate) use self::demo_ws::DemoWsExecutionPayload; -/// WebSocket message emitted to the frontend. -pub(crate) use self::demo_ws::DemoWsMessagePayload; -/// One selectable WebSocket method. -pub(crate) use self::demo_ws::DemoWsMethodOption; -/// WebSocket demo options payload. -pub(crate) use self::demo_ws::DemoWsOptionsPayload; -/// WebSocket subscription request. -pub(crate) use self::demo_ws::DemoWsRequest; -/// One selectable WebSocket role. -pub(crate) use self::demo_ws::DemoWsRoleOption; -/// Current WebSocket session status. -pub(crate) use self::demo_ws::DemoWsStatusPayload; -/// One active WebSocket subscription status. -pub(crate) use self::demo_ws::DemoWsSubscriptionStatusPayload; -/// WebSocket unsubscribe request. -pub(crate) use self::demo_ws::DemoWsUnsubscribeRequest; -/// Builds the standard WebSocket method inventory. -pub(crate) use self::demo_ws::build_ws_method_options; -/// Builds selectable WebSocket roles from pool snapshots. -pub(crate) use self::demo_ws::build_ws_role_options; -/// Connects and subscribes through the persistent WebSocket session. -pub(crate) use self::demo_ws::demo_ws_connect_inner; -/// Disconnects the persistent WebSocket session. -pub(crate) use self::demo_ws::demo_ws_disconnect_inner; -/// Returns the current WebSocket session status. -pub(crate) use self::demo_ws::demo_ws_status_inner; -/// Unsubscribes one active WebSocket subscription. -pub(crate) use self::demo_ws::demo_ws_unsubscribe_inner; /// UI-safe SQL table diagnostic snapshot. pub(crate) use self::demo_sql_common::DemoSqlTableSnapshot; /// Connects to the configured PostgreSQL store. @@ -152,6 +124,36 @@ pub(crate) use self::demo_sql_replay_candidates::program_scope_from_code; pub(crate) use self::demo_sql_replay_candidates::transaction_row_from_pg; /// Validates a CSV filename. pub(crate) use self::demo_sql_replay_candidates::validated_csv_file_name; +/// WebSocket execution response payload. +pub(crate) use self::demo_ws::DemoWsExecutionPayload; +/// WebSocket message emitted to the frontend. +pub(crate) use self::demo_ws::DemoWsMessagePayload; +/// One selectable WebSocket method. +pub(crate) use self::demo_ws::DemoWsMethodOption; +/// WebSocket demo options payload. +pub(crate) use self::demo_ws::DemoWsOptionsPayload; +/// WebSocket subscription request. +pub(crate) use self::demo_ws::DemoWsRequest; +/// One selectable WebSocket role. +pub(crate) use self::demo_ws::DemoWsRoleOption; +/// Current WebSocket session status. +pub(crate) use self::demo_ws::DemoWsStatusPayload; +/// One active WebSocket subscription status. +pub(crate) use self::demo_ws::DemoWsSubscriptionStatusPayload; +/// WebSocket unsubscribe request. +pub(crate) use self::demo_ws::DemoWsUnsubscribeRequest; +/// Builds the standard WebSocket method inventory. +pub(crate) use self::demo_ws::build_ws_method_options; +/// Builds selectable WebSocket roles from pool snapshots. +pub(crate) use self::demo_ws::build_ws_role_options; +/// Connects and subscribes through the persistent WebSocket session. +pub(crate) use self::demo_ws::demo_ws_connect_inner; +/// Disconnects the persistent WebSocket session. +pub(crate) use self::demo_ws::demo_ws_disconnect_inner; +/// Returns the current WebSocket session status. +pub(crate) use self::demo_ws::demo_ws_status_inner; +/// Unsubscribes one active WebSocket subscription. +pub(crate) use self::demo_ws::demo_ws_unsubscribe_inner; /// Frontend logging payload. pub(crate) use self::frontend_log::FrontendLogPayload; /// Emits one normalized frontend log event. @@ -173,6 +175,64 @@ pub(crate) use self::splash::emit_splash_order; /// Waits for the minimum splash duration. pub(crate) use self::splash::wait_until_minimum; +/// Internal demo core extraction item. +pub(crate) use self::demo_core_extraction::DemoCoreExtractionObserver; +/// Internal demo core extraction item. +pub(crate) use self::demo_core_extraction::DemoCoreExtractionOptionsPayload; +/// Internal demo core extraction item. +pub(crate) use self::demo_core_extraction::DemoCoreExtractionProgressPayload; +/// Internal demo core extraction item. +pub(crate) use self::demo_core_extraction::DemoCoreExtractionRequest; +/// Internal demo core extraction item. +pub(crate) use self::demo_core_extraction::DemoCoreExtractionRunGuard; +/// Internal demo core extraction item. +pub(crate) use self::demo_core_extraction::DemoCoreExtractionSummaryPayload; +/// Internal demo core extraction item. +pub(crate) use self::demo_core_extraction::build_demo_core_extraction_pipeline_request; +/// Internal demo core extraction item. +pub(crate) use self::demo_core_extraction::demo_core_extraction_summary_payload; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoDecodeCoverageSummaryPayload; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoDecodeDiagnosticsPayload; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoDecodeProcessorSummaryPayload; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoDecodeReplayDecoderOption; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoDecodeReplayObserver; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoDecodeReplayOptionsPayload; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoDecodeReplayProgressPayload; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoDecodeReplayRequest; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoDecodeReplayRunGuard; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoDecodeReplaySummaryPayload; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoTransactionAnnotationRequest; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::DemoTransactionAnnotationRow; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::annotation_payload; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::available_decoders; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::available_materializers; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::build_demo_decode_replay_pipeline_request; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::coverage_payload; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::demo_decode_replay_summary_payload; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::optional_line_count; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::register_active_campaign; +/// Internal demo decode replay item. +pub(crate) use self::demo_decode_replay::text_sample; // Keep the canonical tracing target as the final facade export. /// Canonical tracing target for the desktop demo application. pub(crate) use self::constants::TRACING_TARGET; diff --git a/kb-app-demo-desktop/src/tauri.rs b/kb-app-demo-desktop/src/tauri.rs index cea7008..967dfd6 100644 --- a/kb-app-demo-desktop/src/tauri.rs +++ b/kb-app-demo-desktop/src/tauri.rs @@ -1,5 +1,5 @@ // file: kb-app-demo-desktop/src/tauri.rs -// version: 8 +// version: 9 //! Tauri runtime assembly and private command wrappers. @@ -64,6 +64,16 @@ pub fn run() -> kb_core::Result<()> { load_demo_sql_replay_programs, load_demo_sql_replay_entities, export_demo_sql_replay_csv, + open_demo_core_extraction_window, + demo_core_extraction_options, + demo_core_extraction_execute, + demo_core_extraction_cancel, + open_demo_decode_replay_window, + demo_decode_replay_options, + demo_decode_replay_execute, + demo_decode_replay_cancel, + demo_decode_replay_diagnostics, + demo_decode_replay_annotations, open_demo_config_window, load_demo_config, ]); @@ -164,9 +174,9 @@ pub fn run() -> kb_core::Result<()> { let run_result = builder.run(tauri::generate_context!()); return match run_result { std::result::Result::Ok(()) => std::result::Result::Ok(()), - std::result::Result::Err(error) => std::result::Result::Err(kb_core::Error::tauri(format!( - "cannot run desktop demo application: {error:?}" - ))), + std::result::Result::Err(error) => std::result::Result::Err(kb_core::Error::tauri( + format!("cannot run desktop demo application: {error:?}"), + )), }; } @@ -182,6 +192,436 @@ fn install_default_rustls_provider() -> kb_core::Result<()> { )), }; } +/// Opens or focuses the canonical to core extraction demo window. +#[tauri::command] +fn open_demo_core_extraction_window( + app_handle: tauri::AppHandle, +) -> std::result::Result<(), std::string::String> { + tracing::info!(target: crate::TRACING_TARGET, "open core extraction demo window"); + let existing_window = app_handle.get_webview_window("demo_core_extraction"); + if let std::option::Option::Some(window) = existing_window { + let show_result = window.show(); + if let std::result::Result::Err(error) = show_result { + return std::result::Result::Err(error.to_string()); + } + let focus_result = window.set_focus(); + if let std::result::Result::Err(error) = focus_result { + return std::result::Result::Err(error.to_string()); + } + return std::result::Result::Ok(()); + } + let build_result = tauri::WebviewWindowBuilder::new( + &app_handle, + "demo_core_extraction", + tauri::WebviewUrl::App("demo_core_extraction.html".into()), + ) + .title("Khadhroony Bot2 - Extraction canonical vers core") + .inner_size(1280.0, 860.0) + .min_inner_size(960.0, 620.0) + .resizable(true) + .visible(true) + .build(); + return match build_result { + std::result::Result::Ok(window) => { + let focus_result = window.set_focus(); + if let std::result::Result::Err(error) = focus_result { + return std::result::Result::Err(error.to_string()); + } + std::result::Result::Ok(()) + }, + std::result::Result::Err(error) => std::result::Result::Err(error.to_string()), + }; +} + +/// Returns the initial core extraction demo options. +#[tauri::command] +fn demo_core_extraction_options( + state: tauri::State<'_, crate::AppState>, +) -> crate::DemoCoreExtractionOptionsPayload { + return crate::DemoCoreExtractionOptionsPayload { + processor_version: kb_pipeline::CORE_EXTRACTION_PROCESSOR_VERSION.to_string(), + default_limit: 100, + default_max_concurrent_extractions: 4, + running: state.demo_core_extraction_running().load(std::sync::atomic::Ordering::Acquire), + }; +} + +/// Executes one bounded canonical transaction to core extraction campaign. +#[tauri::command] +async fn demo_core_extraction_execute( + app_handle: tauri::AppHandle, + state: tauri::State<'_, crate::AppState>, + request: crate::DemoCoreExtractionRequest, +) -> std::result::Result { + let acquire_result = state.demo_core_extraction_running().compare_exchange( + false, + true, + std::sync::atomic::Ordering::AcqRel, + std::sync::atomic::Ordering::Acquire, + ); + if acquire_result.is_err() { + return std::result::Result::Err( + "a core extraction campaign is already running".to_string(), + ); + } + let _run_guard = crate::DemoCoreExtractionRunGuard { + running: state.demo_core_extraction_running(), + }; + state + .demo_core_extraction_cancel_requested() + .store(false, std::sync::atomic::Ordering::Release); + let pipeline_request_result = crate::build_demo_core_extraction_pipeline_request(request); + let pipeline_request = match pipeline_request_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let store_result = crate::connect_postgres_store(state.active_profile()).await; + let store = match store_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let observer = crate::DemoCoreExtractionObserver { + app_handle, + cancel_requested: state.demo_core_extraction_cancel_requested(), + }; + let summary_result = + kb_pipeline::execute_core_extraction(&store, &pipeline_request, &observer).await; + let summary = match summary_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + return std::result::Result::Ok(crate::demo_core_extraction_summary_payload(summary)); +} + +/// Requests cooperative cancellation of the current core extraction campaign. +#[tauri::command] +fn demo_core_extraction_cancel(state: tauri::State<'_, crate::AppState>) -> bool { + let running = state.demo_core_extraction_running().load(std::sync::atomic::Ordering::Acquire); + state + .demo_core_extraction_cancel_requested() + .store(true, std::sync::atomic::Ordering::Release); + return running; +} + +/// Opens or focuses the contextual decode replay window. +#[tauri::command] +fn open_demo_decode_replay_window( + app_handle: tauri::AppHandle, +) -> std::result::Result<(), std::string::String> { + tracing::info!(target: crate::TRACING_TARGET, action = "open_window", window = "demo_decode_replay", "open contextual decode replay window"); + let existing_window = app_handle.get_webview_window("demo_decode_replay"); + if let std::option::Option::Some(window) = existing_window { + let show_result = window.show(); + if let std::result::Result::Err(error) = show_result { + return std::result::Result::Err(error.to_string()); + } + let focus_result = window.set_focus(); + if let std::result::Result::Err(error) = focus_result { + return std::result::Result::Err(error.to_string()); + } + return std::result::Result::Ok(()); + } + let build_result = tauri::WebviewWindowBuilder::new( + &app_handle, + "demo_decode_replay", + tauri::WebviewUrl::App("demo_decode_replay.html".into()), + ) + .title("Khadhroony Bot2 - Décodage et matérialisation") + .inner_size(1320.0, 900.0) + .min_inner_size(980.0, 660.0) + .resizable(true) + .visible(true) + .build(); + return match build_result { + std::result::Result::Ok(window) => { + let focus_result = window.set_focus(); + if let std::result::Result::Err(error) = focus_result { + return std::result::Result::Err(error.to_string()); + } + std::result::Result::Ok(()) + }, + std::result::Result::Err(error) => std::result::Result::Err(error.to_string()), + }; +} + +/// Returns available decoders and default replay bounds. +#[tauri::command] +fn demo_decode_replay_options( + state: tauri::State<'_, crate::AppState>, +) -> crate::DemoDecodeReplayOptionsPayload { + let decoders = crate::available_decoders(); + let decoder_options = decoders + .iter() + .map(|decoder| { + let identity = decoder.identity(); + return crate::DemoDecodeReplayDecoderOption { + name: identity.name, + version: identity.version, + program_ids: decoder + .surfaces() + .iter() + .map(|surface| return surface.program_id.to_string()) + .collect(), + }; + }) + .collect::>(); + let materializers = crate::available_materializers(); + let materializer_names = materializers + .iter() + .map(|materializer| { + let identity = materializer.identity(); + return format!("{}@{}", identity.name, identity.version); + }) + .collect::>(); + let running = state.demo_decode_replay_running().load(std::sync::atomic::Ordering::Acquire); + tracing::debug!( + target: crate::TRACING_TARGET, + action = "load_options", + pipeline_version = kb_pipeline::DECODE_PIPELINE_VERSION, + decoder_count = decoder_options.len(), + materializer_names = ?materializer_names, + default_limit = 100_u32, + default_max_concurrent_inputs = 4_u32, + running, + "return contextual decode replay options" + ); + return crate::DemoDecodeReplayOptionsPayload { + pipeline_version: kb_pipeline::DECODE_PIPELINE_VERSION.to_string(), + decoders: decoder_options, + materializer_names, + default_limit: 100, + default_max_concurrent_inputs: 4, + running, + }; +} + +/// Executes one bounded contextual decode and optional materialization campaign. +#[tauri::command] +async fn demo_decode_replay_execute( + app_handle: tauri::AppHandle, + state: tauri::State<'_, crate::AppState>, + request: crate::DemoDecodeReplayRequest, +) -> std::result::Result { + let campaign_id = kb_pipeline::new_decode_campaign_id(); + tracing::debug!( + target: crate::TRACING_TARGET, + action = "execute", + campaign_id = %campaign_id, + instruction_state = %request.instruction_state, + signature_line_count = crate::optional_line_count(&request.signatures_text), + program_id = ?request.program_id, + instruction_paths_text = ?request.instruction_paths_text, + decoder_names = ?request.decoder_names, + limit = request.limit, + max_concurrent_inputs = request.max_concurrent_inputs, + all_compatible = request.all_compatible, + force_replay = request.force_replay, + force_replay_all_matching = request.force_replay_all_matching, + materialize_after_decode = request.materialize_after_decode, + "received contextual decode replay command" + ); + let acquire_result = state.demo_decode_replay_running().compare_exchange( + false, + true, + std::sync::atomic::Ordering::AcqRel, + std::sync::atomic::Ordering::Acquire, + ); + if acquire_result.is_err() { + tracing::warn!(target: crate::TRACING_TARGET, action = "execute", campaign_id = %campaign_id, accepted = false, reason = "already_running", "reject contextual decode replay command"); + return std::result::Result::Err( + "a contextual decode replay campaign is already running".to_string(), + ); + } + let _run_guard = crate::DemoDecodeReplayRunGuard { + running: state.demo_decode_replay_running(), + campaign_id: state.demo_decode_replay_campaign_id(), + }; + let register_result = crate::register_active_campaign( + state.demo_decode_replay_campaign_id(), + campaign_id.as_str(), + ); + if let std::result::Result::Err(error) = register_result { + tracing::error!(target: crate::TRACING_TARGET, action = "register_campaign", campaign_id = %campaign_id, error = %error, "cannot lock active contextual decode campaign slot"); + return std::result::Result::Err(error); + } + state + .demo_decode_replay_cancel_requested() + .store(false, std::sync::atomic::Ordering::Release); + let pipeline_request_result = + crate::build_demo_decode_replay_pipeline_request(request, campaign_id.clone()); + let pipeline_request = match pipeline_request_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "build_request", campaign_id = %campaign_id, error = %error, "cannot build contextual decode replay request"); + return std::result::Result::Err(error); + }, + }; + tracing::debug!( + target: crate::TRACING_TARGET, + action = "execute_pipeline", + campaign_id = %pipeline_request.campaign_id, + signature_count = pipeline_request.selection.signatures.len(), + signature_sample = ?crate::text_sample(pipeline_request.selection.signatures.as_slice(), 5), + processing_states = ?pipeline_request.selection.processing_states, + min_slot = ?pipeline_request.selection.min_slot, + max_slot = ?pipeline_request.selection.max_slot, + program_ids = ?pipeline_request.selection.program_ids, + instruction_paths = ?pipeline_request.selection.instruction_paths, + incomplete_signatures = pipeline_request.selection.incomplete_signatures, + limit = pipeline_request.selection.limit, + decoder_names = ?pipeline_request.decoder_names, + dispatch_policy = ?pipeline_request.dispatch_policy, + max_concurrent_inputs = pipeline_request.max_concurrent_inputs, + force_replay = pipeline_request.force_replay, + force_replay_all_matching = pipeline_request.force_replay_all_matching, + materialize_after_decode = pipeline_request.materialize_after_decode, + "start contextual decode replay pipeline" + ); + let store_result = crate::connect_postgres_store(state.active_profile()).await; + let store = match store_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "connect_store", campaign_id = %pipeline_request.campaign_id, error = %error, "cannot connect contextual decode replay store"); + return std::result::Result::Err(error); + }, + }; + let decoders = crate::available_decoders(); + let materializers = crate::available_materializers(); + let observer = crate::DemoDecodeReplayObserver { + app_handle, + campaign_id: pipeline_request.campaign_id.clone(), + cancel_requested: state.demo_decode_replay_cancel_requested(), + }; + let summary_result = kb_pipeline::execute_decode_replay( + &store, + &pipeline_request, + decoders.as_slice(), + materializers.as_slice(), + &observer, + ) + .await; + let summary = match summary_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => { + tracing::error!(target: crate::TRACING_TARGET, action = "execute_pipeline", campaign_id = %pipeline_request.campaign_id, error = %error, "contextual decode replay pipeline failed"); + return std::result::Result::Err(error.to_string()); + }, + }; + tracing::debug!( + target: crate::TRACING_TARGET, + action = "execute_pipeline", + campaign_id = %summary.campaign_id, + selected = summary.selected, + started = summary.started, + completed = summary.completed, + unmatched = summary.unmatched, + not_started = summary.not_started, + failed_inputs = summary.failed_inputs, + cancelled = summary.cancelled, + processors = ?summary.processors, + "contextual decode replay command completed" + ); + return std::result::Result::Ok(crate::demo_decode_replay_summary_payload(summary)); +} + +/// Requests cooperative cancellation of the current decode replay campaign. +#[tauri::command] +fn demo_decode_replay_cancel(state: tauri::State<'_, crate::AppState>) -> bool { + let running = state.demo_decode_replay_running().load(std::sync::atomic::Ordering::Acquire); + state + .demo_decode_replay_cancel_requested() + .store(true, std::sync::atomic::Ordering::Release); + let campaign_lock_result = state.demo_decode_replay_campaign_id().lock(); + let campaign_id = match campaign_lock_result { + std::result::Result::Ok(active_campaign_id) => active_campaign_id.clone(), + std::result::Result::Err(_) => std::option::Option::None, + }; + tracing::debug!(target: crate::TRACING_TARGET, action = "cancel", campaign_id = ?campaign_id, running, cancellation_requested = true, "contextual decode replay cancellation command handled"); + return running; +} + +/// Loads read-only decode, materialization, ledger and coverage diagnostics. +#[tauri::command] +async fn demo_decode_replay_diagnostics( + state: tauri::State<'_, crate::AppState>, +) -> std::result::Result { + tracing::debug!(target: crate::TRACING_TARGET, action = "load_diagnostics", coverage_limit = 500_u32, "load contextual decode replay diagnostics"); + let store_result = crate::connect_postgres_store(state.active_profile()).await; + let store = match store_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let table_result = store.known_table_diagnostics().await; + let all_tables = match table_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + let selected_tables: std::vec::Vec = all_tables + .iter() + .filter(|table| { + return table.table_name.starts_with("kb_sol_decode_") + || table.table_name.starts_with("kb_sol_mat_") + || table.table_name == "kb_sol_ops_processing_ledger"; + }) + .map(crate::table_snapshot_from_pg) + .collect(); + let coverage_result = kb_store::DecodePipelineStore::list_decode_coverage_summary( + &store, + std::option::Option::None, + std::option::Option::None, + 500, + ) + .await; + let coverage = match coverage_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + tracing::debug!(target: crate::TRACING_TARGET, action = "load_diagnostics", table_count = selected_tables.len(), coverage_count = coverage.len(), "contextual decode replay diagnostics loaded"); + return std::result::Result::Ok(crate::DemoDecodeDiagnosticsPayload { + tables: selected_tables, + coverage: coverage.into_iter().map(crate::coverage_payload).collect(), + }); +} + +/// Loads a bounded journal of committed SPL Memo transaction annotations. +#[tauri::command] +async fn demo_decode_replay_annotations( + state: tauri::State<'_, crate::AppState>, + request: crate::DemoTransactionAnnotationRequest, +) -> std::result::Result, std::string::String> { + let filter_result = kb_store::MaterializedEventFilter::new( + std::option::Option::Some("transaction_annotations".to_string()), + std::option::Option::Some("transaction_annotation".to_string()), + request.signature_contains, + request.limit, + ); + let filter = match filter_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + tracing::debug!(target: crate::TRACING_TARGET, action = "load_transaction_annotations", signature_contains = ?filter.signature_contains, limit = filter.limit, "load bounded committed transaction annotation journal"); + let store_result = crate::connect_postgres_store(state.active_profile()).await; + let store = match store_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let rows_result = + kb_store::DecodePipelineStore::list_materialized_events(&store, &filter).await; + let rows = match rows_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), + }; + let mut output = std::vec::Vec::with_capacity(rows.len()); + for row in rows { + let mapped = crate::annotation_payload(row); + match mapped { + std::result::Result::Ok(value) => output.push(value), + std::result::Result::Err(error) => return std::result::Result::Err(error), + } + } + tracing::debug!(target: crate::TRACING_TARGET, action = "load_transaction_annotations", row_count = output.len(), "bounded committed transaction annotation journal loaded"); + return std::result::Result::Ok(output); +} #[tauri::command] fn open_demo_config_window( @@ -217,15 +657,11 @@ fn open_demo_config_window( } #[tauri::command] -fn load_demo_config( - state: tauri::State<'_, crate::AppState>, -) -> crate::DemoConfigPayload { +fn load_demo_config(state: tauri::State<'_, crate::AppState>) -> crate::DemoConfigPayload { return crate::demo_config_payload(state.inner()); } -fn into_ipc_result( - result: kb_core::Result, -) -> std::result::Result { +fn into_ipc_result(result: kb_core::Result) -> std::result::Result { return match result { std::result::Result::Ok(value) => std::result::Result::Ok(value), std::result::Result::Err(error) => std::result::Result::Err(error.to_string()), @@ -244,9 +680,7 @@ fn load_project_readme() -> std::result::Result) -> bool { - let running = state - .demo_backfill_running() - .load(std::sync::atomic::Ordering::Acquire); + let running = state.demo_backfill_running().load(std::sync::atomic::Ordering::Acquire); state .demo_backfill_cancel_requested() .store(true, std::sync::atomic::Ordering::Release); @@ -305,9 +739,7 @@ fn demo_backfill_options( default_max_pages: 20, default_max_concurrent_requests: 4, default_max_retries: 2, - running: state - .demo_backfill_running() - .load(std::sync::atomic::Ordering::Acquire), + running: state.demo_backfill_running().load(std::sync::atomic::Ordering::Acquire), }; } @@ -327,9 +759,7 @@ async fn demo_backfill_execute( if acquire_result.is_err() { return std::result::Result::Err("a backfill campaign is already running".to_string()); } - let _run_guard = crate::DemoBackfillRunGuard { - running: state.demo_backfill_running(), - }; + let _run_guard = crate::DemoBackfillRunGuard { running: state.demo_backfill_running() }; state .demo_backfill_cancel_requested() .store(false, std::sync::atomic::Ordering::Release); @@ -414,9 +844,7 @@ fn demo_http_list_pool_clients( } #[tauri::command] -fn demo_http_options( - state: tauri::State<'_, crate::AppState>, -) -> crate::DemoHttpOptionsPayload { +fn demo_http_options(state: tauri::State<'_, crate::AppState>) -> crate::DemoHttpOptionsPayload { return crate::DemoHttpOptionsPayload { roles: crate::build_http_role_options(state.http_pool().snapshot()), methods: crate::build_http_method_options(), @@ -480,7 +908,10 @@ fn open_demo_ws_window( #[tauri::command] fn demo_ws_list_pool_clients( state: tauri::State<'_, crate::AppState>, -) -> std::result::Result, std::string::String> { +) -> std::result::Result< + std::vec::Vec, + std::string::String, +> { let pool = match state.demo_ws_pool() { std::result::Result::Ok(value) => value, std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()), @@ -876,4 +1307,3 @@ async fn load_demo_sql_replay_entities( } return std::result::Result::Ok(output); } - diff --git a/kb-app-demo-desktop/vite.config.ts b/kb-app-demo-desktop/vite.config.ts index 23946e3..6b8c1e1 100644 --- a/kb-app-demo-desktop/vite.config.ts +++ b/kb-app-demo-desktop/vite.config.ts @@ -1,5 +1,5 @@ // file: kb-app-demo-desktop/vite.config.ts -// version: 4 +// version: 5 import { defineConfig, normalizePath } from "vite"; import { NodePackageImporter } from "sass-embedded"; @@ -31,6 +31,8 @@ export default defineConfig(() => ({ "demo_sql_pg_core": normalizePath(resolve(__dirname, 'frontend/demo_sql_pg_core.html')), "demo_sql_replay_candidates": normalizePath(resolve(__dirname, 'frontend/demo_sql_replay_candidates.html')), "demo_config": normalizePath(resolve(__dirname, 'frontend/demo_config.html')), + "demo_core_extraction": normalizePath(resolve(__dirname, 'frontend/demo_core_extraction.html')), + "demo_decode_replay": normalizePath(resolve(__dirname, 'frontend/demo_decode_replay.html')), }, output: { entryFileNames: 'js/[name]-[hash].js',