diff --git a/Aetheria.Rts.Web/Electron/aetheria-remote-publication-reader.ts b/Aetheria.Rts.Web/Electron/aetheria-remote-publication-reader.ts index ae9d725ad..9a0f07cc5 100644 --- a/Aetheria.Rts.Web/Electron/aetheria-remote-publication-reader.ts +++ b/Aetheria.Rts.Web/Electron/aetheria-remote-publication-reader.ts @@ -1,10 +1,5 @@ -import { decode } from "@msgpack/msgpack"; import { CultMesh } from "cultmesh-ts"; import type { CultNetPeer } from "cultnet-ts"; -import type { - CultNetErrorMessage, - CultNetSnapshotResponseRawMessage, -} from "cultnet-ts"; const connectionId = 0x43554c54; @@ -51,67 +46,17 @@ export class AetheriaRemotePublicationReader { } private async readSingleDocument(schemaId: string, recordKey: string): Promise { - const response = await this.requestSnapshot([recordKey]); - const document = response.documents.find(candidate => - candidate.recordKey === recordKey); - if (!document) { - throw new Error(`Aetheria remote publication ${this.endpoint} did not return ${schemaId} at ${recordKey}.`); - } - - return decode(toBytes(document.payload)); - } - - private async requestSnapshot( - recordKeys: readonly string[], - ): Promise { - const peer = await this.peer(); - const messageId = `${this.runtimeId}:snapshot:${Date.now()}:${Math.random().toString(16).slice(2)}`; - - return new Promise((resolve, reject) => { - const cleanup = (): void => { - clearTimeout(timer); - peer.off("message", onMessage); - peer.off("invalidMessage", onInvalidMessage); - peer.off("error", onError); - peer.off("close", onClose); - }; - const rejectWith = (error: Error): void => { - cleanup(); - reject(error); - }; - const onMessage = (message: unknown): void => { - const response = message as CultNetSnapshotResponseRawMessage | CultNetErrorMessage; - if (response.schemaVersion === "cultnet.error.v0") { - cleanup(); - reject(new Error(response.error)); - return; - } - - if (response.schemaVersion !== "cultnet.snapshot_response_raw.v0" || - response.messageId !== messageId) { - return; - } - - cleanup(); - resolve(response); - }; - const onInvalidMessage = (error: Error): void => rejectWith(error); - const onError = (error: Error): void => rejectWith(error); - const onClose = (): void => rejectWith(new Error("CultMesh peer closed before snapshot response.")); - const timer = setTimeout( - () => rejectWith(new Error(`Timed out waiting for Aetheria remote snapshot ${messageId}.`)), - this.timeoutMs); - - peer.on("message", onMessage); - peer.on("invalidMessage", onInvalidMessage); - peer.on("error", onError); - peer.on("close", onClose); - peer.sendSnapshotRequest({ - schemaVersion: "cultnet.snapshot_request.v0", - messageId, - recordKeys: [...recordKeys], - }); - }); + return CultMesh.documentFromPeerSnapshot( + () => this.peer(), + schemaId, + recordKey, + { + documentId: recordKey, + routeHint: CultMesh.routeHint("network", this.endpoint), + timeoutMs: this.timeoutMs, + messageIdPrefix: `${this.runtimeId}:snapshot`, + }, + ).latest(); } private async peer(): Promise { @@ -129,11 +74,3 @@ export class AetheriaRemotePublicationReader { return this.#peer; } } - -function toBytes(value: unknown): Uint8Array { - if (value instanceof Uint8Array) - return value; - if (Array.isArray(value)) - return Uint8Array.from(value); - throw new Error("CultNet raw document payload was not binary."); -} diff --git a/Aetheria.Rts.Web/scripts/verify-stage7b-rts-client.ps1 b/Aetheria.Rts.Web/scripts/verify-stage7b-rts-client.ps1 index 4bc8f1808..23d73e430 100644 --- a/Aetheria.Rts.Web/scripts/verify-stage7b-rts-client.ps1 +++ b/Aetheria.Rts.Web/scripts/verify-stage7b-rts-client.ps1 @@ -102,6 +102,24 @@ if ($LASTEXITCODE -gt 1) { Write-Error "Stage 7B verifier could not run rg for transport wrapper layout checks." } +$remotePublicationReaderPatterns = @( + '@msgpack/msgpack', + 'sendSnapshotRequest', + 'snapshot_response_raw', + 'CultNetSnapshotResponseRawMessage', + 'recordKeys', + 'decode\(' +) + +$remotePublicationReaderPattern = ($remotePublicationReaderPatterns -join '|') +$remotePublicationReaderHits = & rg -n $remotePublicationReaderPattern Electron/aetheria-remote-publication-reader.ts 2>$null +if ($LASTEXITCODE -eq 0) { + Write-Error "Stage 7B verifier failed: RTS remote publication reader still owns raw snapshot/codec details.`n$remotePublicationReaderHits" +} +if ($LASTEXITCODE -gt 1) { + Write-Error "Stage 7B verifier could not run rg for remote publication reader layout checks." +} + $requiredBindingSymbols = @( 'createAetheriaRuntimeRtsOperationHandles', 'createAetheriaRuntimeRtsQueryHandles',