workinf_Blender_Wasm/web/app/src/volume/nanovdb-stream.ts

import {
  planNanoVDBRanges,
  validateNanoVDBBundleManifest,
  verifyNanoVDBChunk,
  type NanoVDBBundleManifestIR,
  type NanoVDBRangeIR,
} from "../../../protocol/volume-vdb";

export interface NanoVDBStreamProgressIR {
  completedChunks: number;
  totalChunks: number;
  completedBytes: number;
  totalBytes: number;
}

export interface NanoVDBStreamResultIR extends NanoVDBStreamProgressIR {
  declaredBundleSha256: string;
}

export type NanoVDBRangeSource = (range: NanoVDBRangeIR, signal: AbortSignal) => Promise<ArrayBuffer>;
export type NanoVDBChunkConsumer = (range: NanoVDBRangeIR, data: ArrayBuffer, signal: AbortSignal) => Promise<void> | void;

function cancelled(signal: AbortSignal): void {
  if (signal.aborted) throw new DOMException("NanoVDB stream cancelled", "AbortError");
}

export async function streamNanoVDBChunks(
  manifestValue: NanoVDBBundleManifestIR,
  source: NanoVDBRangeSource,
  consume: NanoVDBChunkConsumer,
  signal: AbortSignal,
  onProgress?: (progress: NanoVDBStreamProgressIR) => void,
): Promise<NanoVDBStreamResultIR> {
  const manifest = validateNanoVDBBundleManifest(manifestValue);
  const ranges = planNanoVDBRanges(manifest);
  let completedBytes = 0;
  for (const range of ranges) {
    cancelled(signal);
    const data = await source(range, signal);
    cancelled(signal);
    await verifyNanoVDBChunk(manifest.chunks[range.chunkIndex], data);
    cancelled(signal);
    await consume(range, data, signal);
    completedBytes += data.byteLength;
    onProgress?.({
      completedChunks: range.chunkIndex + 1,
      totalChunks: ranges.length,
      completedBytes,
      totalBytes: manifest.bundleByteLength,
    });
  }
  return {
    completedChunks: ranges.length,
    totalChunks: ranges.length,
    completedBytes,
    totalBytes: manifest.bundleByteLength,
    declaredBundleSha256: manifest.bundleSha256,
  };
}

function parseContentRange(value: string | null): { start: number; endInclusive: number; total: number } | undefined {
  const match = value?.match(/^bytes (\d+)-(\d+)\/(\d+)$/);
  if (!match) return undefined;
  const start = Number(match[1]);
  const endInclusive = Number(match[2]);
  const total = Number(match[3]);
  if (![start, endInclusive, total].every(Number.isSafeInteger)) return undefined;
  return { start, endInclusive, total };
}

export function createHttpNanoVDBRangeSource(
  url: string,
  expectedBundleBytes: number,
  fetcher: typeof fetch = fetch,
): NanoVDBRangeSource {
  return createResumableHttpNanoVDBRangeSource(url, expectedBundleBytes, { fetcher, retries: 0, requireStableEtag: false });
}

export interface NanoVDBHttpRangeOptions {
  fetcher?: typeof fetch;
  retries?: number;
  retryDelayMs?: number;
  requireStableEtag?: boolean;
}

async function waitForHttpRetry(delayMs: number, attempt: number, signal: AbortSignal): Promise<void> {
  if (delayMs === 0) return;
  await new Promise<void>((resolve, reject) => {
    const onAbort = (): void => {
      clearTimeout(timer);
      reject(new DOMException("NanoVDB HTTP range cancelled", "AbortError"));
    };
    const timer = setTimeout(() => {
      signal.removeEventListener("abort", onAbort);
      resolve();
    }, delayMs * (attempt + 1));
    signal.addEventListener("abort", onAbort, { once: true });
  });
}

function protocolFailure(error: unknown): boolean {
  return error instanceof Error && /^(?:NANOVDB_|VDB_)/.test(error.message);
}

export function createResumableHttpNanoVDBRangeSource(
  url: string,
  expectedBundleBytes: number,
  options: NanoVDBHttpRangeOptions = {},
): NanoVDBRangeSource {
  if (!url || !Number.isSafeInteger(expectedBundleBytes) || expectedBundleBytes <= 0) throw new Error("NANOVDB_MANIFEST_INVALID: HTTP range source is invalid");
  const fetcher = options.fetcher ?? fetch;
  const retries = options.retries ?? 2;
  const retryDelayMs = options.retryDelayMs ?? 25;
  if (!Number.isSafeInteger(retries) || retries < 0 || retries > 8 || !Number.isSafeInteger(retryDelayMs) || retryDelayMs < 0 || retryDelayMs > 10_000) {
    throw new Error("NANOVDB_MANIFEST_INVALID: HTTP retry policy is invalid");
  }
  let etag: string | undefined;
  return async (range, signal) => {
    if (!Number.isSafeInteger(range.start) || !Number.isSafeInteger(range.endExclusive) || range.start < 0 || range.endExclusive <= range.start || range.endExclusive > expectedBundleBytes) {
      throw new Error("NANOVDB_STREAM_INCOMPLETE: HTTP range is outside the NanoVDB bundle");
    }
    const output = new Uint8Array(range.endExclusive - range.start);
    let written = 0;
    let lastStatus = 0;
    let lastFailure = "network interruption";
    for (let attempt = 0; attempt <= retries; attempt++) {
      if (signal.aborted) throw new DOMException("NanoVDB HTTP range cancelled", "AbortError");
      const requestStart = range.start + written;
      const headers: Record<string, string> = { Range: `bytes=${requestStart}-${range.endExclusive - 1}` };
      if (etag) headers["If-Range"] = etag;
      let response: Response;
      try {
        response = await fetcher(url, { method: "GET", headers, signal, cache: "no-store" });
      }
      catch (error) {
        if (signal.aborted || error instanceof DOMException && error.name === "AbortError") throw new DOMException("NanoVDB HTTP range cancelled", "AbortError");
        lastFailure = error instanceof Error ? error.message : String(error);
        if (attempt === retries) break;
        await waitForHttpRetry(retryDelayMs, attempt, signal);
        continue;
      }
      lastStatus = response.status;
      if (response.status === 408 || response.status === 425 || response.status === 429 || response.status >= 500) {
        if (attempt === retries) break;
        await waitForHttpRetry(retryDelayMs, attempt, signal);
        continue;
      }
      if (response.status !== 206) throw new Error(`NANOVDB_STREAM_INCOMPLETE: HTTP range request returned ${response.status}, expected 206`);
      const responseEtag = response.headers.get("ETag") ?? undefined;
      if (etag && responseEtag !== etag) throw new Error("NANOVDB_HASH_MISMATCH: HTTP ETag changed during NanoVDB streaming");
      if (!etag && responseEtag) etag = responseEtag;
      if (options.requireStableEtag && !etag) throw new Error("NANOVDB_STREAM_INCOMPLETE: HTTP ETag is required for resumable streaming");
      const contentRange = parseContentRange(response.headers.get("Content-Range"));
      if (!contentRange || contentRange.start !== requestStart || contentRange.endInclusive !== range.endExclusive - 1 || contentRange.total !== expectedBundleBytes) {
        throw new Error("NANOVDB_STREAM_INCOMPLETE: HTTP Content-Range does not match the NanoVDB manifest");
      }
      if (!response.body) throw new Error("NANOVDB_STREAM_INCOMPLETE: HTTP range response has no body");
      const reader = response.body.getReader();
      try {
        while (true) {
          const next = await reader.read();
          if (next.done) break;
          const value = next.value;
          if (written + value.byteLength > output.byteLength) {
            throw new Error("NANOVDB_STREAM_INCOMPLETE: HTTP range response exceeds the requested byte length");
          }
          output.set(value, written);
          written += value.byteLength;
        }
      }
      catch (error) {
        if (signal.aborted || error instanceof DOMException && error.name === "AbortError") throw new DOMException("NanoVDB HTTP range cancelled", "AbortError");
        if (protocolFailure(error)) throw error;
        if (written === output.byteLength) return output.buffer;
        lastFailure = error instanceof Error ? error.message : String(error);
        if (attempt === retries) break;
        await waitForHttpRetry(retryDelayMs, attempt, signal);
        continue;
      }
      finally {
        reader.releaseLock();
      }
      if (written !== output.byteLength) {
        throw new Error("NANOVDB_STREAM_INCOMPLETE: HTTP range response has an unexpected byte length");
      }
      return output.buffer;
    }
    throw new Error(`NANOVDB_STREAM_INCOMPLETE: HTTP range retry budget exhausted after ${lastStatus ? `status ${lastStatus}` : lastFailure}`);
  };
}