From 3ceace05ff1e03ab515524e9de6ec1f8b200ae44 Mon Sep 17 00:00:00 2001 From: real-venus Date: Thu, 25 Jun 2026 04:56:00 -0700 Subject: [PATCH 1/3] feat: add streaming client-side CSV/GeoJSON/shapefile bulk export pipeline (#60) Streams filtered resource data from the REST API in chunked, gzipped NDJSON requests, transforms rows into the target format, and writes the result to disk via the File System Access API (Blob download fallback) without holding the whole dataset in memory. - types/export.ts: config, filters, progress, events and bounds (10k rows/chunk, 100-chunk cap, 50MB flush threshold, 6-decimal coord precision) - utils/ndjsonParser.ts: streaming NDJSON parser over DecompressionStream with a pako polyfill fallback - utils/formatTransformers.ts: RFC 4180 CSV, GeoJSON Point Feature (truncated coords), shapefile record extraction - utils/shapefileWriter.ts + utils/zip.ts: EPSG:4326 Point .shp/.shx/.dbf/.prj bundled into a STORED zip (with CRC32) - utils/fileWriter.ts: FS Access WritableStream writer + in-memory Blob fallback with memory-pressure warning - services/exportPipeline.ts: orchestrator with 2-chunk sliding prefetch window, memory guard, client-side filtering and progress events - hooks/useExportProgress.ts: reduces pipeline events into progress state - components/panels/ExportDialog.tsx: format selector, column picker, filter builder, progress bar + cancel - app/export: host route for the dialog - unit tests (transformers, ndjson, zip/shapefile, pipeline) + Playwright export spec verifying downloaded row count --- src/app/export/page.tsx | 31 ++ src/components/panels/ExportDialog.tsx | 368 ++++++++++++++++++++++ src/hooks/useExportProgress.ts | 134 ++++++++ src/services/exportPipeline.ts | 419 +++++++++++++++++++++++++ src/types/export.ts | 107 +++++++ src/utils/fileWriter.ts | 197 ++++++++++++ src/utils/formatTransformers.ts | 118 +++++++ src/utils/ndjsonParser.ts | 153 +++++++++ src/utils/shapefileWriter.ts | 225 +++++++++++++ src/utils/zip.ts | 113 +++++++ tests/e2e/export.spec.ts | 71 +++++ tests/unit/exportPipeline.test.ts | 183 +++++++++++ tests/unit/formatTransformers.test.ts | 93 ++++++ tests/unit/ndjsonParser.test.ts | 58 ++++ tests/unit/zipShapefile.test.ts | 75 +++++ 15 files changed, 2345 insertions(+) create mode 100644 src/app/export/page.tsx create mode 100644 src/components/panels/ExportDialog.tsx create mode 100644 src/hooks/useExportProgress.ts create mode 100644 src/services/exportPipeline.ts create mode 100644 src/types/export.ts create mode 100644 src/utils/fileWriter.ts create mode 100644 src/utils/formatTransformers.ts create mode 100644 src/utils/ndjsonParser.ts create mode 100644 src/utils/shapefileWriter.ts create mode 100644 src/utils/zip.ts create mode 100644 tests/e2e/export.spec.ts create mode 100644 tests/unit/exportPipeline.test.ts create mode 100644 tests/unit/formatTransformers.test.ts create mode 100644 tests/unit/ndjsonParser.test.ts create mode 100644 tests/unit/zipShapefile.test.ts diff --git a/src/app/export/page.tsx b/src/app/export/page.tsx new file mode 100644 index 0000000..b678f39 --- /dev/null +++ b/src/app/export/page.tsx @@ -0,0 +1,31 @@ +"use client"; + +import { useState } from "react"; +import { ExportDialog } from "@/components/panels/ExportDialog"; + +/** + * Standalone host page for the bulk export dialog. Kept separate from the main + * dashboard so the export flow can be driven and verified end to end. + */ +export default function ExportPage() { + const [open, setOpen] = useState(false); + + return ( +
+

Resource Data Export

+

+ Export filtered resource-consumption data as CSV, GeoJSON or a zipped + shapefile. Large queries are streamed in chunks and written straight to + disk. +

+ + setOpen(false)} /> +
+ ); +} diff --git a/src/components/panels/ExportDialog.tsx b/src/components/panels/ExportDialog.tsx new file mode 100644 index 0000000..b268c40 --- /dev/null +++ b/src/components/panels/ExportDialog.tsx @@ -0,0 +1,368 @@ +"use client"; + +import { useCallback, useEffect, useMemo, useState } from "react"; +import type { + ExportConfig, + ExportFilter, + ExportFormat, + FilterOperator, + SchemaColumn, +} from "@/types/export"; +import { MAX_ROWS } from "@/types/export"; +import { + useExportProgress, + type UseExportProgressReturn, +} from "@/hooks/useExportProgress"; +import type { ExportPipelineDeps } from "@/services/exportPipeline"; + +const FORMAT_OPTIONS: { value: ExportFormat; label: string }[] = [ + { value: "csv", label: "CSV" }, + { value: "geojson", label: "GeoJSON" }, + { value: "shapefile", label: "Shapefile (zipped)" }, +]; + +const OPERATORS: { value: FilterOperator; label: string }[] = [ + { value: "eq", label: "=" }, + { value: "neq", label: "≠" }, + { value: "gt", label: ">" }, + { value: "gte", label: "≥" }, + { value: "lt", label: "<" }, + { value: "lte", label: "≤" }, + { value: "contains", label: "contains" }, +]; + +export interface ExportDialogProps { + open: boolean; + onClose: () => void; + /** Override the schema fetch (defaults to GET /api/resources/schema). */ + fetchSchema?: () => Promise; + /** Injected pipeline dependencies (for testing / custom transport). */ + pipelineDeps?: ExportPipelineDeps; + /** Override the progress hook (for testing). */ + progress?: UseExportProgressReturn; + baseUrl?: string; +} + +async function defaultFetchSchema(baseUrl: string): Promise { + const url = `${baseUrl}/api/resources/schema`; + const res = await fetch(url, { headers: { Accept: "application/json" } }); + if (!res.ok) throw new Error(`Schema request failed: HTTP ${res.status}`); + return (await res.json()) as SchemaColumn[]; +} + +export function ExportDialog({ + open, + onClose, + fetchSchema, + pipelineDeps, + progress: progressOverride, + baseUrl, +}: ExportDialogProps) { + const resolvedBaseUrl = + baseUrl ?? process.env.NEXT_PUBLIC_API_URL ?? "http://localhost:4000"; + + const internal = useExportProgress(pipelineDeps); + const { progress, start, cancel, reset, isRunning } = + progressOverride ?? internal; + + const [schema, setSchema] = useState([]); + const [schemaError, setSchemaError] = useState(null); + const [format, setFormat] = useState("csv"); + const [selectedColumns, setSelectedColumns] = useState([]); + const [filters, setFilters] = useState([]); + + // Load the column schema when the dialog opens. + useEffect(() => { + if (!open) return; + let cancelled = false; + const load = fetchSchema ?? (() => defaultFetchSchema(resolvedBaseUrl)); + load() + .then((cols) => { + if (cancelled) return; + setSchema(cols); + setSelectedColumns(cols.map((c) => c.name)); + setSchemaError(null); + }) + .catch((err: unknown) => { + if (!cancelled) setSchemaError((err as Error).message); + }); + return () => { + cancelled = true; + }; + }, [open, fetchSchema, resolvedBaseUrl]); + + // Close on Escape (unless an export is running). + useEffect(() => { + if (!open) return; + const onKey = (e: KeyboardEvent) => { + if (e.key === "Escape" && !isRunning) onClose(); + }; + window.addEventListener("keydown", onKey); + return () => window.removeEventListener("keydown", onKey); + }, [open, isRunning, onClose]); + + const toggleColumn = useCallback((name: string) => { + setSelectedColumns((prev) => + prev.includes(name) ? prev.filter((c) => c !== name) : [...prev, name] + ); + }, []); + + const addFilter = useCallback(() => { + const first = schema[0]?.name ?? ""; + setFilters((prev) => [...prev, { field: first, operator: "eq", value: "" }]); + }, [schema]); + + const updateFilter = useCallback( + (index: number, patch: Partial) => { + setFilters((prev) => + prev.map((f, i) => (i === index ? { ...f, ...patch } : f)) + ); + }, + [] + ); + + const removeFilter = useCallback((index: number) => { + setFilters((prev) => prev.filter((_, i) => i !== index)); + }, []); + + const percent = useMemo(() => { + if (!progress.totalChunks) return 0; + return Math.min( + 100, + Math.round((progress.currentChunk / progress.totalChunks) * 100) + ); + }, [progress.currentChunk, progress.totalChunks]); + + const handleExport = useCallback(() => { + const config: ExportConfig = { + format, + columns: selectedColumns, + filters: filters.filter((f) => f.field), + maxRows: MAX_ROWS, + fileName: "resource-export", + }; + void start(config); + }, [format, selectedColumns, filters, start]); + + const handleClose = useCallback(() => { + if (isRunning) cancel(); + reset(); + onClose(); + }, [isRunning, cancel, reset, onClose]); + + if (!open) return null; + + return ( +
+
+
+

+ Bulk Export +

+ +
+ + {schemaError && ( +
+ Could not load column schema: {schemaError} +
+ )} + + {/* Format selector */} + + + {/* Column picker */} +
+ Columns +
+ {schema.length === 0 && ( + + Loading columns… + + )} + {schema.map((col) => ( + + ))} +
+
+ + {/* Filter builder */} +
+
+ Filters + +
+ {filters.map((filter, i) => ( +
+ + + updateFilter(i, { value: e.target.value })} + disabled={isRunning} + placeholder="value" + className="flex-1 rounded-lg border border-border bg-background px-2 py-1.5 text-sm focus:outline-none focus:ring-2 focus:ring-ring disabled:opacity-50" + /> + +
+ ))} +
+ + {/* Progress */} + {progress.status !== "idle" && ( +
+
+
+
+
+ + {progress.status === "complete" + ? `Done — ${progress.rowsWritten.toLocaleString()} rows` + : progress.status === "error" + ? progress.error ?? "Export failed." + : progress.status === "cancelled" + ? "Export cancelled." + : `Chunk ${progress.currentChunk}/${ + progress.totalChunks ?? "?" + } — ${progress.rowsWritten.toLocaleString()} rows`} + + + {(progress.bytesDownloaded / (1024 * 1024)).toFixed(1)} MB in + +
+ {progress.warning && ( +

{progress.warning}

+ )} + {progress.usedFallback && ( +

+ Using in-memory download (File System Access API unavailable). +

+ )} +
+ )} + + {/* Actions */} +
+ {isRunning ? ( + + ) : ( + <> + + + + )} +
+
+
+ ); +} + +export default ExportDialog; diff --git a/src/hooks/useExportProgress.ts b/src/hooks/useExportProgress.ts new file mode 100644 index 0000000..517c077 --- /dev/null +++ b/src/hooks/useExportProgress.ts @@ -0,0 +1,134 @@ +"use client"; + +import { useCallback, useEffect, useRef, useState } from "react"; +import type { ExportConfig, ExportProgress } from "@/types/export"; +import { + ExportPipeline, + type ExportPipelineDeps, +} from "@/services/exportPipeline"; + +const INITIAL_PROGRESS: ExportProgress = { + status: "idle", + currentChunk: 0, + totalChunks: null, + bytesDownloaded: 0, + rowsWritten: 0, + bytesWritten: 0, + usedFallback: false, + warning: null, + error: null, +}; + +export interface UseExportProgressReturn { + progress: ExportProgress; + /** Begin an export; resolves when finished, cancelled or errored. */ + start: (config: ExportConfig) => Promise; + cancel: () => void; + reset: () => void; + isRunning: boolean; +} + +/** + * React binding over {@link ExportPipeline}. Subscribes to pipeline events + * (chunkStart, chunkComplete, warning, complete, cancelled, error) and reduces + * them into a single {@link ExportProgress} object for the UI. + */ +export function useExportProgress( + deps?: ExportPipelineDeps +): UseExportProgressReturn { + const [progress, setProgress] = useState(INITIAL_PROGRESS); + const pipelineRef = useRef(null); + const mountedRef = useRef(true); + + useEffect(() => { + mountedRef.current = true; + return () => { + mountedRef.current = false; + pipelineRef.current?.cancel(); + }; + }, []); + + const update = useCallback((patch: Partial) => { + if (!mountedRef.current) return; + setProgress((prev) => ({ ...prev, ...patch })); + }, []); + + const start = useCallback( + async (config: ExportConfig) => { + if (pipelineRef.current) return; // One export at a time. + + const pipeline = new ExportPipeline(config, deps); + pipelineRef.current = pipeline; + + setProgress({ + ...INITIAL_PROGRESS, + status: "preparing", + totalChunks: pipeline.totalChunks, + }); + + const unsubscribe = pipeline.on((event) => { + switch (event.type) { + case "chunkStart": + update({ + status: "exporting", + currentChunk: event.chunk + 1, + totalChunks: event.totalChunks, + usedFallback: pipeline.usedFallback, + }); + break; + case "chunkComplete": + update({ + rowsWritten: event.rowsWritten, + bytesDownloaded: event.bytesDownloaded, + bytesWritten: event.bytesWritten, + usedFallback: pipeline.usedFallback, + }); + break; + case "warning": + update({ warning: event.message, usedFallback: pipeline.usedFallback }); + break; + case "complete": + update({ + status: "complete", + rowsWritten: event.rowsWritten, + bytesWritten: event.bytesWritten, + }); + break; + case "cancelled": + update({ status: "cancelled" }); + break; + case "error": + update({ status: "error", error: event.message }); + break; + } + }); + + try { + await pipeline.run(); + } catch { + // The "error" event already populated progress.error. + } finally { + unsubscribe(); + pipelineRef.current = null; + } + }, + [deps, update] + ); + + const cancel = useCallback(() => { + pipelineRef.current?.cancel(); + }, []); + + const reset = useCallback(() => { + pipelineRef.current?.cancel(); + pipelineRef.current = null; + setProgress(INITIAL_PROGRESS); + }, []); + + const isRunning = + progress.status === "preparing" || + progress.status === "exporting" || + progress.status === "finalizing"; + + return { progress, start, cancel, reset, isRunning }; +} diff --git a/src/services/exportPipeline.ts b/src/services/exportPipeline.ts new file mode 100644 index 0000000..612daff --- /dev/null +++ b/src/services/exportPipeline.ts @@ -0,0 +1,419 @@ +"use client"; + +import { + CHUNK_SIZE, + MAX_CHUNKS, + MAX_ROWS, + MEMORY_FLUSH_THRESHOLD, + PREFETCH_WINDOW, + type ExportConfig, + type ExportEvent, + type ExportEventListener, + type ExportFilter, + type ResourceRow, +} from "@/types/export"; +import { parseNdjsonStream } from "@/utils/ndjsonParser"; +import { + csvHeader, + csvTransformer, + geoJsonTransformerString, + shapefileTransformer, + type GeometryFieldConfig, +} from "@/utils/formatTransformers"; +import { + ShapefilePointWriter, + buildShapefileZipEntries, + type ShapefileField, +} from "@/utils/shapefileWriter"; +import { createStoredZip } from "@/utils/zip"; +import { + createFileWriter, + type CreateFileWriterOptions, + type FileWriter, +} from "@/utils/fileWriter"; + +/** Injectable dependencies, primarily for testing. */ +export interface ExportPipelineDeps { + fetchFn?: typeof fetch; + createWriter?: (options: CreateFileWriterOptions) => Promise; + baseUrl?: string; + /** Column → shapefile DBF type map (defaults all columns to "C"). */ + shapefileFieldTypes?: Record; +} + +const FORMAT_META: Record< + ExportConfig["format"], + { mimeType: string; extension: string; description: string } +> = { + csv: { mimeType: "text/csv", extension: ".csv", description: "CSV file" }, + geojson: { + mimeType: "application/geo+json", + extension: ".geojson", + description: "GeoJSON file", + }, + shapefile: { + mimeType: "application/zip", + extension: ".zip", + description: "Zipped shapefile", + }, +}; + +function compareFilter(rowValue: unknown, filter: ExportFilter): boolean { + const { operator, value } = filter; + if (operator === "contains") { + return String(rowValue ?? "") + .toLowerCase() + .includes(String(value).toLowerCase()); + } + // Numeric comparison when both sides are numbers, else lexical. + const a = rowValue as number | string; + const b = value as number | string; + switch (operator) { + case "eq": + return a === b || String(a) === String(b); + case "neq": + return !(a === b || String(a) === String(b)); + case "gt": + return a > b; + case "gte": + return a >= b; + case "lt": + return a < b; + case "lte": + return a <= b; + default: + return true; + } +} + +function matchesFilters(row: ResourceRow, filters: ExportFilter[]): boolean { + for (const filter of filters) { + if (!compareFilter(row[filter.field], filter)) return false; + } + return true; +} + +/** + * Streaming bulk-export orchestrator. Fetches chunked, gzipped NDJSON from the + * REST API with a sliding prefetch window, transforms rows into the target + * format, and writes the result through a {@link FileWriter} while keeping + * in-memory buffering under {@link MEMORY_FLUSH_THRESHOLD}. + */ +export class ExportPipeline { + private readonly listeners = new Set(); + private readonly controller = new AbortController(); + private readonly fetchFn: typeof fetch; + private readonly createWriter: ( + options: CreateFileWriterOptions + ) => Promise; + private readonly baseUrl: string; + private readonly geometry: GeometryFieldConfig; + private readonly maxRows: number; + + private writer: FileWriter | null = null; + private buffer: Uint8Array[] = []; + private bufferBytes = 0; + private bytesDownloaded = 0; + private bytesWritten = 0; + private rowsWritten = 0; + private cancelled = false; + + // Format-specific output state. + private csvHeaderWritten = false; + private geoJsonStarted = false; + private shapefileWriter: ShapefilePointWriter | null = null; + + constructor( + private readonly config: ExportConfig, + private readonly deps: ExportPipelineDeps = {} + ) { + this.fetchFn = deps.fetchFn ?? globalThis.fetch.bind(globalThis); + this.createWriter = deps.createWriter ?? createFileWriter; + this.baseUrl = + deps.baseUrl ?? + process.env.NEXT_PUBLIC_API_URL ?? + "http://localhost:4000"; + this.geometry = { + lonField: config.lonField ?? "longitude", + latField: config.latField ?? "latitude", + }; + this.maxRows = Math.min(config.maxRows ?? MAX_ROWS, MAX_ROWS); + } + + on(listener: ExportEventListener): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + private emit(event: ExportEvent): void { + for (const listener of this.listeners) listener(event); + } + + cancel(): void { + this.cancelled = true; + this.controller.abort(); + } + + get totalChunks(): number { + return Math.min(MAX_CHUNKS, Math.ceil(this.maxRows / CHUNK_SIZE)); + } + + /** Whether the Blob download fallback is in use (no FS Access API). */ + get usedFallback(): boolean { + return this.writer?.usedFallback ?? false; + } + + private buildUrl(offset: number, limit: number): string { + const params = new URLSearchParams({ + offset: String(offset), + limit: String(limit), + }); + if (this.config.columns.length) { + params.set("columns", this.config.columns.join(",")); + } + for (const f of this.config.filters ?? []) { + params.append("filter", `${f.field}|${f.operator}|${String(f.value)}`); + } + return `${this.baseUrl}/api/resources/export?${params.toString()}`; + } + + private fetchChunk(chunkIndex: number): Promise { + const offset = chunkIndex * CHUNK_SIZE; + const limit = Math.min(CHUNK_SIZE, this.maxRows - offset); + return this.fetchFn(this.buildUrl(offset, limit), { + signal: this.controller.signal, + headers: { Accept: "application/x-ndjson", "Accept-Encoding": "gzip" }, + }); + } + + /** Encode and buffer output, flushing to the writer past the memory guard. */ + private async enqueue(text: string): Promise { + const bytes = new TextEncoder().encode(text); + this.buffer.push(bytes); + this.bufferBytes += bytes.byteLength; + if (this.bufferBytes >= MEMORY_FLUSH_THRESHOLD) { + await this.flush(); + } + } + + private async flush(): Promise { + if (!this.buffer.length || !this.writer) return; + const merged = new Uint8Array(this.bufferBytes); + let offset = 0; + for (const part of this.buffer) { + merged.set(part, offset); + offset += part.byteLength; + } + this.buffer = []; + this.bufferBytes = 0; + await this.writer.write(merged); + this.bytesWritten = this.writer.bytesWritten; + } + + private async writeRow(row: ResourceRow): Promise { + const cols = this.config.columns; + switch (this.config.format) { + case "csv": { + const columns = cols.length ? cols : Object.keys(row); + if (!this.csvHeaderWritten) { + await this.enqueue(csvHeader(columns)); + this.csvHeaderWritten = true; + } + await this.enqueue(csvTransformer(row, columns)); + break; + } + case "geojson": { + const prefix = this.geoJsonStarted ? "," : ""; + this.geoJsonStarted = true; + await this.enqueue( + prefix + geoJsonTransformerString(row, cols, this.geometry) + ); + break; + } + case "shapefile": { + // Shapefiles must be serialised at the end (header carries lengths); + // accumulate point records now and emit the zip on finalize. + if (!this.shapefileWriter) { + this.shapefileWriter = new ShapefilePointWriter( + this.shapefileFields(row) + ); + } + const rec = shapefileTransformer(row, cols, this.geometry); + this.shapefileWriter.addPoint(rec.lon, rec.lat, rec.attributes); + break; + } + } + this.rowsWritten += 1; + } + + private shapefileFields(sampleRow: ResourceRow): ShapefileField[] { + const cols = this.config.columns.length + ? this.config.columns + : Object.keys(sampleRow); + const types = this.deps.shapefileFieldTypes ?? {}; + return cols + .filter( + (c) => c !== this.geometry.lonField && c !== this.geometry.latField + ) + // DBF field names are capped at 10 chars; keep the projection order. + .map((c) => ({ name: c.slice(0, 10), type: types[c] ?? "C" })); + } + + /** Run the export end to end. Resolves when the file is fully written. */ + async run(): Promise { + const meta = FORMAT_META[this.config.format]; + const baseName = this.config.fileName ?? "resource-export"; + + this.writer = await this.createWriter({ + fileName: `${baseName}${meta.extension}`, + mimeType: meta.mimeType, + description: meta.description, + extensions: [meta.extension], + onWarning: (message) => this.emit({ type: "warning", message }), + }); + + if (this.config.format === "geojson") { + await this.enqueue('{"type":"FeatureCollection","features":['); + } + + const filters = this.config.filters ?? []; + const inflight = new Map>(); + let nextToFetch = 0; + let ended = false; + + const prime = () => { + while ( + inflight.size < PREFETCH_WINDOW && + nextToFetch < this.totalChunks && + !ended + ) { + inflight.set(nextToFetch, this.fetchChunk(nextToFetch)); + nextToFetch += 1; + } + }; + + try { + prime(); + + for (let chunk = 0; chunk < this.totalChunks && !ended; chunk++) { + if (this.cancelled) break; + const responsePromise = inflight.get(chunk); + inflight.delete(chunk); + if (!responsePromise) break; + + this.emit({ + type: "chunkStart", + chunk, + totalChunks: this.totalChunks, + }); + + const response = await responsePromise; + if (!response.ok) { + throw new Error(`Export chunk ${chunk} failed: HTTP ${response.status}`); + } + if (!response.body) { + ended = true; + break; + } + + let rowsInChunk = 0; + const gzip = (response.headers.get("content-encoding") ?? "").includes( + "gzip" + ); + for await (const row of parseNdjsonStream(response.body, { + gzip, + signal: this.controller.signal, + onBytes: (n) => { + this.bytesDownloaded += n; + }, + })) { + rowsInChunk += 1; + if (filters.length && !matchesFilters(row, filters)) continue; + if (this.rowsWritten >= this.maxRows) { + ended = true; + break; + } + await this.writeRow(row); + } + + this.emit({ + type: "chunkComplete", + chunk, + rowsWritten: this.rowsWritten, + bytesDownloaded: this.bytesDownloaded, + bytesWritten: this.bytesWritten + this.bufferBytes, + }); + + // A short chunk means the dataset is exhausted. + if (rowsInChunk < CHUNK_SIZE) ended = true; + if (!ended) prime(); + } + + if (this.cancelled) { + await this.abortWriter(inflight); + this.emit({ type: "cancelled" }); + return; + } + + await this.finalize(baseName); + await this.drain(inflight); + + this.emit({ + type: "complete", + rowsWritten: this.rowsWritten, + bytesWritten: this.bytesWritten, + }); + } catch (err) { + await this.abortWriter(inflight); + if ((err as Error)?.name === "AbortError" || this.cancelled) { + this.emit({ type: "cancelled" }); + return; + } + this.emit({ type: "error", message: (err as Error).message }); + throw err; + } + } + + private async finalize(baseName: string): Promise { + if (this.config.format === "geojson") { + await this.enqueue("]}"); + } + if (this.config.format === "shapefile" && this.shapefileWriter) { + const components = this.shapefileWriter.build(); + const zip = createStoredZip(buildShapefileZipEntries(components, baseName)); + this.buffer.push(zip); + this.bufferBytes += zip.byteLength; + } + await this.flush(); + if (this.writer) { + await this.writer.close(); + this.bytesWritten = this.writer.bytesWritten; + } + } + + private async abortWriter(inflight: Map>): Promise { + await this.writer?.abort().catch(() => {}); + await this.drain(inflight); + } + + /** Discard any prefetched-but-unused response bodies. */ + private async drain(inflight: Map>): Promise { + for (const promise of inflight.values()) { + try { + const res = await promise; + await res.body?.cancel().catch(() => {}); + } catch { + // Aborted/failed prefetches are expected here. + } + } + inflight.clear(); + } +} + +/** Convenience factory mirroring the class constructor. */ +export function createExportPipeline( + config: ExportConfig, + deps?: ExportPipelineDeps +): ExportPipeline { + return new ExportPipeline(config, deps); +} diff --git a/src/types/export.ts b/src/types/export.ts new file mode 100644 index 0000000..5e22000 --- /dev/null +++ b/src/types/export.ts @@ -0,0 +1,107 @@ +/** + * Types and bounds for the client-side bulk export pipeline. + * + * Filtered resource-consumption data is streamed from the REST API in chunked, + * gzipped NDJSON requests, transformed into CSV / GeoJSON / shapefile, and + * written to disk via the File System Access API (or a Blob download fallback) + * without ever holding the whole dataset in memory. + */ + +export type ExportFormat = "csv" | "geojson" | "shapefile"; + +/** Rows per REST request. */ +export const CHUNK_SIZE = 10_000; +/** Hard cap on chunks → 1,000,000 rows total export limit. */ +export const MAX_CHUNKS = 100; +export const MAX_ROWS = CHUNK_SIZE * MAX_CHUNKS; + +/** Flush buffered output to disk once it exceeds this many bytes. */ +export const MEMORY_FLUSH_THRESHOLD = 50 * 1024 * 1024; +/** Max bytes to accumulate before the Blob fallback warns about memory. */ +export const FALLBACK_MEMORY_LIMIT = 100 * 1024 * 1024; + +/** Decimal places GeoJSON/shapefile coordinates are truncated to (~0.11 m). */ +export const COORD_PRECISION = 6; + +/** Number of chunks fetched ahead of processing (sliding window). */ +export const PREFETCH_WINDOW = 2; + +export type FilterOperator = + | "eq" + | "neq" + | "gt" + | "gte" + | "lt" + | "lte" + | "contains"; + +export interface ExportFilter { + field: string; + operator: FilterOperator; + value: string | number | boolean; +} + +export interface ExportConfig { + format: ExportFormat; + /** Projected columns, in output order. Empty = all columns from the schema. */ + columns: string[]; + filters?: ExportFilter[]; + /** Upper bound on exported rows; clamped to {@link MAX_ROWS}. */ + maxRows?: number; + /** Longitude property name for geometry formats. @default "longitude" */ + lonField?: string; + /** Latitude property name for geometry formats. @default "latitude" */ + latField?: string; + /** Suggested download file name (without extension). */ + fileName?: string; +} + +/** A single parsed NDJSON row from the export endpoint. */ +export type ResourceRow = Record; + +/** A column definition from GET /api/resources/schema. */ +export interface SchemaColumn { + name: string; + type: "string" | "number" | "boolean" | "date"; + label?: string; +} + +export type ExportStatus = + | "idle" + | "preparing" + | "exporting" + | "finalizing" + | "complete" + | "cancelled" + | "error"; + +export interface ExportProgress { + status: ExportStatus; + currentChunk: number; + totalChunks: number | null; + bytesDownloaded: number; + rowsWritten: number; + bytesWritten: number; + /** True when the Blob download fallback is in use (no FS Access API). */ + usedFallback: boolean; + /** Non-fatal warning, e.g. memory pressure on the fallback path. */ + warning: string | null; + error: string | null; +} + +/** Events emitted by the pipeline and surfaced through {@link ExportProgress}. */ +export type ExportEvent = + | { type: "chunkStart"; chunk: number; totalChunks: number | null } + | { + type: "chunkComplete"; + chunk: number; + rowsWritten: number; + bytesDownloaded: number; + bytesWritten: number; + } + | { type: "warning"; message: string } + | { type: "complete"; rowsWritten: number; bytesWritten: number } + | { type: "cancelled" } + | { type: "error"; message: string }; + +export type ExportEventListener = (event: ExportEvent) => void; diff --git a/src/utils/fileWriter.ts b/src/utils/fileWriter.ts new file mode 100644 index 0000000..c0abaf8 --- /dev/null +++ b/src/utils/fileWriter.ts @@ -0,0 +1,197 @@ +/** + * Output abstraction for the export pipeline. + * + * Implementation 1 ({@link FsAccessFileWriter}) streams bytes straight to disk + * through the File System Access API's `FileSystemWritableFileStream`, so the + * full export never resides in memory. + * + * Implementation 2 ({@link BlobFileWriter}) is the fallback for browsers without + * the API: it accumulates bytes and triggers a Blob download on close, warning + * once accumulation crosses {@link FALLBACK_MEMORY_LIMIT}. + */ + +import { FALLBACK_MEMORY_LIMIT } from "@/types/export"; + +export interface FileWriter { + write(chunk: Uint8Array): Promise; + /** Finalise the file (close stream / trigger download). */ + close(): Promise; + /** Abort without finalising; discards buffered/streamed data where possible. */ + abort(reason?: unknown): Promise; + readonly bytesWritten: number; + readonly usedFallback: boolean; +} + +// --- Minimal File System Access API typings (not in the default DOM lib) --- + +interface FileSystemWritableFileStreamLike { + write(data: BufferSource): Promise; + close(): Promise; + abort?(reason?: unknown): Promise; +} + +interface FileSystemFileHandleLike { + createWritable(options?: { + keepExistingData?: boolean; + }): Promise; +} + +interface SaveFilePickerOptions { + suggestedName?: string; + types?: { description?: string; accept: Record }[]; +} + +type ShowSaveFilePicker = ( + options?: SaveFilePickerOptions +) => Promise; + +export function isFileSystemAccessSupported(): boolean { + return ( + typeof window !== "undefined" && + typeof (window as unknown as { showSaveFilePicker?: ShowSaveFilePicker }) + .showSaveFilePicker === "function" + ); +} + +class FsAccessFileWriter implements FileWriter { + readonly usedFallback = false; + private _bytesWritten = 0; + + constructor(private readonly stream: FileSystemWritableFileStreamLike) {} + + get bytesWritten(): number { + return this._bytesWritten; + } + + async write(chunk: Uint8Array): Promise { + await this.stream.write(chunk as unknown as BufferSource); + this._bytesWritten += chunk.byteLength; + } + + async close(): Promise { + await this.stream.close(); + } + + async abort(reason?: unknown): Promise { + if (this.stream.abort) { + await this.stream.abort(reason); + } else { + await this.stream.close().catch(() => {}); + } + } +} + +export interface BlobFileWriterOptions { + fileName: string; + mimeType: string; + /** Emitted once when accumulation crosses the memory limit. */ + onWarning?: (message: string) => void; + /** Override the download trigger (used in tests). */ + onDownload?: (blob: Blob, fileName: string) => void; +} + +class BlobFileWriter implements FileWriter { + readonly usedFallback = true; + private chunks: Uint8Array[] = []; + private _bytesWritten = 0; + private warned = false; + private aborted = false; + + constructor(private readonly options: BlobFileWriterOptions) {} + + get bytesWritten(): number { + return this._bytesWritten; + } + + async write(chunk: Uint8Array): Promise { + if (this.aborted) return; + this.chunks.push(chunk); + this._bytesWritten += chunk.byteLength; + if ( + !this.warned && + this._bytesWritten > FALLBACK_MEMORY_LIMIT && + this.options.onWarning + ) { + this.warned = true; + this.options.onWarning( + `Export exceeds ${Math.round( + FALLBACK_MEMORY_LIMIT / (1024 * 1024) + )} MB and your browser lacks the File System Access API; holding the file in memory may be slow.` + ); + } + } + + async close(): Promise { + if (this.aborted) return; + const parts = this.chunks as unknown as BlobPart[]; + const blob = new Blob(parts, { type: this.options.mimeType }); + this.chunks = []; + + if (this.options.onDownload) { + this.options.onDownload(blob, this.options.fileName); + return; + } + triggerBlobDownload(blob, this.options.fileName); + } + + async abort(): Promise { + this.aborted = true; + this.chunks = []; + } +} + +function triggerBlobDownload(blob: Blob, fileName: string): void { + const url = URL.createObjectURL(blob); + const anchor = document.createElement("a"); + anchor.href = url; + anchor.download = fileName; + document.body.appendChild(anchor); + anchor.click(); + anchor.remove(); + // Revoke on the next tick so the download has a chance to start. + setTimeout(() => URL.revokeObjectURL(url), 0); +} + +export interface CreateFileWriterOptions { + fileName: string; + mimeType: string; + /** Picker description, e.g. "CSV file". */ + description?: string; + /** File extensions for the picker, e.g. [".csv"]. */ + extensions?: string[]; + onWarning?: (message: string) => void; + /** Force the Blob fallback regardless of API support (used in tests). */ + forceFallback?: boolean; + onDownload?: (blob: Blob, fileName: string) => void; +} + +/** + * Resolve the best available {@link FileWriter}. Prefers the File System Access + * API (true streaming to disk); falls back to an in-memory Blob download. A + * cancelled save picker propagates as an `AbortError`. + */ +export async function createFileWriter( + options: CreateFileWriterOptions +): Promise { + if (!options.forceFallback && isFileSystemAccessSupported()) { + const picker = ( + window as unknown as { showSaveFilePicker: ShowSaveFilePicker } + ).showSaveFilePicker; + const accept: Record = { + [options.mimeType]: options.extensions ?? [], + }; + const handle = await picker({ + suggestedName: options.fileName, + types: [{ description: options.description, accept }], + }); + const stream = await handle.createWritable(); + return new FsAccessFileWriter(stream); + } + + return new BlobFileWriter({ + fileName: options.fileName, + mimeType: options.mimeType, + onWarning: options.onWarning, + onDownload: options.onDownload, + }); +} diff --git a/src/utils/formatTransformers.ts b/src/utils/formatTransformers.ts new file mode 100644 index 0000000..7d12ba3 --- /dev/null +++ b/src/utils/formatTransformers.ts @@ -0,0 +1,118 @@ +/** + * Row-to-format conversion helpers for the bulk export pipeline. + * + * - CSV: RFC 4180 compliant escaping and quoting. + * - GeoJSON: a `Feature` with `Point` geometry; coordinates truncated to + * {@link COORD_PRECISION} decimal places. + * - Shapefile: extraction of `(lon, lat, attributes)` consumed by + * {@link ShapefilePointWriter}. + */ + +import { COORD_PRECISION, type ResourceRow } from "@/types/export"; + +const FACTOR = 10 ** COORD_PRECISION; + +/** Truncate (not round) a coordinate to {@link COORD_PRECISION} decimals. */ +export function truncateCoord(value: number): number { + if (!Number.isFinite(value)) return 0; + return Math.trunc(value * FACTOR) / FACTOR; +} + +/** RFC 4180 field escaping: quote when the value contains , " CR or LF. */ +export function csvEscape(value: unknown): string { + if (value === null || value === undefined) return ""; + const str = typeof value === "string" ? value : String(value); + if (/[",\r\n]/.test(str)) { + return `"${str.replace(/"/g, '""')}"`; + } + return str; +} + +/** Header line for the given column projection (CRLF terminated). */ +export function csvHeader(columns: string[]): string { + return columns.map(csvEscape).join(",") + "\r\n"; +} + +/** A single CSV data line for the projected columns (CRLF terminated). */ +export function csvTransformer(row: ResourceRow, columns: string[]): string { + return columns.map((col) => csvEscape(row[col])).join(",") + "\r\n"; +} + +export interface GeoJsonFeature { + type: "Feature"; + geometry: { type: "Point"; coordinates: [number, number] }; + properties: Record; +} + +export interface GeometryFieldConfig { + lonField: string; + latField: string; +} + +function coerceNumber(value: unknown): number { + return typeof value === "number" ? value : Number(value); +} + +/** + * Convert a row into a GeoJSON `Feature`. The geometry is built from the + * configured lon/lat fields; `properties` carries the projected columns (the + * geometry fields are excluded since they are represented by the geometry). + */ +export function geoJsonTransformer( + row: ResourceRow, + columns: string[], + geometry: GeometryFieldConfig +): GeoJsonFeature { + const lon = truncateCoord(coerceNumber(row[geometry.lonField])); + const lat = truncateCoord(coerceNumber(row[geometry.latField])); + + const projected = columns.length ? columns : Object.keys(row); + const properties: Record = {}; + for (const col of projected) { + if (col === geometry.lonField || col === geometry.latField) continue; + properties[col] = row[col]; + } + + return { + type: "Feature", + geometry: { type: "Point", coordinates: [lon, lat] }, + properties, + }; +} + +/** Serialised GeoJSON feature string (no surrounding whitespace). */ +export function geoJsonTransformerString( + row: ResourceRow, + columns: string[], + geometry: GeometryFieldConfig +): string { + return JSON.stringify(geoJsonTransformer(row, columns, geometry)); +} + +export interface ShapefileRecordInput { + lon: number; + lat: number; + attributes: Record; +} + +/** + * Extract the geometry and attribute payload for a shapefile point record. + * Coordinates are truncated to {@link COORD_PRECISION}; attributes exclude the + * geometry fields. + */ +export function shapefileTransformer( + row: ResourceRow, + columns: string[], + geometry: GeometryFieldConfig +): ShapefileRecordInput { + const lon = truncateCoord(coerceNumber(row[geometry.lonField])); + const lat = truncateCoord(coerceNumber(row[geometry.latField])); + + const projected = columns.length ? columns : Object.keys(row); + const attributes: Record = {}; + for (const col of projected) { + if (col === geometry.lonField || col === geometry.latField) continue; + attributes[col] = row[col]; + } + return { lon, lat, attributes }; +} diff --git a/src/utils/ndjsonParser.ts b/src/utils/ndjsonParser.ts new file mode 100644 index 0000000..bcce261 --- /dev/null +++ b/src/utils/ndjsonParser.ts @@ -0,0 +1,153 @@ +/** + * Streaming NDJSON parser. Pipes a gzipped response body through the browser's + * native `DecompressionStream` (falling back to the `pako` polyfill), splits the + * decoded text on newlines, and yields parsed JSON objects as soon as each + * complete line is available — without buffering the whole response. + */ + +import type { ResourceRow } from "@/types/export"; + +export interface NdjsonParseOptions { + /** Response body is gzip-compressed. @default true */ + gzip?: boolean; + /** Invoked with the number of compressed bytes read, for progress. */ + onBytes?: (bytes: number) => void; + signal?: AbortSignal; +} + +type PakoModule = { + inflate: (data: Uint8Array) => Uint8Array; + ungzip?: (data: Uint8Array) => Uint8Array; +}; + +let pakoPromise: Promise | null = null; + +async function loadPako(): Promise { + if (pakoPromise) return pakoPromise; + pakoPromise = (async () => { + // `pako` is an optional polyfill loaded only on browsers without + // DecompressionStream. The specifier is held in a variable so bundlers do + // not attempt to resolve this optional dependency at build time. + const specifier = "pako"; + const mod = (await import(/* @vite-ignore */ /* webpackIgnore: true */ specifier)) as { + default?: PakoModule; + } & PakoModule; + return (mod.ungzip ? mod : mod.default) as PakoModule; + })(); + return pakoPromise; +} + +function hasDecompressionStream(): boolean { + return typeof globalThis.DecompressionStream === "function"; +} + +/** Count raw bytes flowing through the stream (before decompression). */ +function byteCountingStream( + onBytes?: (bytes: number) => void +): TransformStream { + return new TransformStream({ + transform(chunk, controller) { + onBytes?.(chunk.byteLength); + controller.enqueue(chunk); + }, + }); +} + +/** + * Yield parsed rows from a (gzipped) NDJSON response body. Blank lines are + * skipped; a malformed final line without a trailing newline is still parsed. + */ +export async function* parseNdjsonStream( + body: ReadableStream, + options: NdjsonParseOptions = {} +): AsyncGenerator { + const { gzip = true, onBytes, signal } = options; + + // Native decompression keeps everything as a stream. The pako fallback must + // buffer the (gzipped) body first, then inflate — used only on old browsers. + if (gzip && !hasDecompressionStream()) { + yield* parseWithPako(body, onBytes, signal); + return; + } + + let stream: ReadableStream = body.pipeThrough( + byteCountingStream(onBytes) + ); + if (gzip) { + // Cast: the DOM lib types DecompressionStream's writable side as + // BufferSource, which TS will not unify with ReadableStream. + stream = stream.pipeThrough( + new DecompressionStream("gzip") as unknown as ReadableWritablePair< + Uint8Array, + Uint8Array + > + ); + } + + const reader = stream + .pipeThrough( + new TextDecoderStream() as unknown as ReadableWritablePair< + string, + Uint8Array + > + ) + .getReader(); + let buffer = ""; + + try { + for (;;) { + if (signal?.aborted) throw new DOMException("Aborted", "AbortError"); + const { done, value } = await reader.read(); + if (done) break; + buffer += value; + + let newlineIndex: number; + while ((newlineIndex = buffer.indexOf("\n")) >= 0) { + const line = buffer.slice(0, newlineIndex).trim(); + buffer = buffer.slice(newlineIndex + 1); + if (line) yield JSON.parse(line) as ResourceRow; + } + } + const last = buffer.trim(); + if (last) yield JSON.parse(last) as ResourceRow; + } finally { + reader.releaseLock(); + } +} + +async function* parseWithPako( + body: ReadableStream, + onBytes?: (bytes: number) => void, + signal?: AbortSignal +): AsyncGenerator { + const reader = body.getReader(); + const chunks: Uint8Array[] = []; + let total = 0; + try { + for (;;) { + if (signal?.aborted) throw new DOMException("Aborted", "AbortError"); + const { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + total += value.byteLength; + onBytes?.(value.byteLength); + } + } finally { + reader.releaseLock(); + } + + const compressed = new Uint8Array(total); + let offset = 0; + for (const c of chunks) { + compressed.set(c, offset); + offset += c.byteLength; + } + + const pako = await loadPako(); + const inflated = (pako.ungzip ?? pako.inflate)(compressed); + const text = new TextDecoder().decode(inflated); + for (const rawLine of text.split("\n")) { + const line = rawLine.trim(); + if (line) yield JSON.parse(line) as ResourceRow; + } +} diff --git a/src/utils/shapefileWriter.ts b/src/utils/shapefileWriter.ts new file mode 100644 index 0000000..0c0e465 --- /dev/null +++ b/src/utils/shapefileWriter.ts @@ -0,0 +1,225 @@ +/** + * Minimal shapefile writer for Point geometries in EPSG:4326 (WGS84). + * + * Produces the four standard components — `.shp` (geometry), `.shx` (index), + * `.dbf` (attributes) and `.prj` (projection) — which {@link createStoredZip} + * bundles into a single download. Only the Point shape type is supported, which + * matches the resource-consumption export (a coordinate per row). + * + * Shapefiles cannot be written in a single streaming pass: the `.shp`/`.shx` + * headers carry the total file length and the `.dbf` header carries field + * widths derived from the data. Records are therefore buffered and serialised in + * {@link ShapefilePointWriter.build}; the buffer is bounded by the export row + * cap (1,000,000 points ≈ tens of MB). + */ + +const SHAPE_TYPE_POINT = 1; +const HEADER_SIZE = 100; +const POINT_CONTENT_BYTES = 20; // 4 (type) + 8 (x) + 8 (y) +const POINT_CONTENT_WORDS = POINT_CONTENT_BYTES / 2; + +/** WKT for the WGS84 geographic coordinate system. */ +const WGS84_WKT = + 'GEOGCS["GCS_WGS_1984",DATUM["D_WGS_1984",' + + 'SPHEROID["WGS_1984",6378137.0,298.257223563]],' + + 'PRIMEM["Greenwich",0.0],UNIT["Degree",0.0174532925199433]]'; + +export type DbfFieldType = "C" | "N"; + +export interface ShapefileField { + name: string; + type: DbfFieldType; +} + +interface PointRecord { + x: number; + y: number; + attributes: Record; +} + +export interface ShapefileComponents { + shp: Uint8Array; + shx: Uint8Array; + dbf: Uint8Array; + prj: Uint8Array; +} + +function writeShpHeader(view: DataView, fileLengthBytes: number, bbox: BBox): void { + view.setInt32(0, 9994, false); // file code (big-endian) + view.setInt32(24, fileLengthBytes / 2, false); // length in 16-bit words (BE) + view.setInt32(28, 1000, true); // version (LE) + view.setInt32(32, SHAPE_TYPE_POINT, true); // shape type (LE) + view.setFloat64(36, bbox.xmin, true); + view.setFloat64(44, bbox.ymin, true); + view.setFloat64(52, bbox.xmax, true); + view.setFloat64(60, bbox.ymax, true); + // Z and M ranges remain 0 (bytes 68..99). +} + +interface BBox { + xmin: number; + ymin: number; + xmax: number; + ymax: number; +} + +export class ShapefilePointWriter { + private readonly records: PointRecord[] = []; + private readonly bbox: BBox = { + xmin: Infinity, + ymin: Infinity, + xmax: -Infinity, + ymax: -Infinity, + }; + + constructor(private readonly fields: ShapefileField[]) {} + + get count(): number { + return this.records.length; + } + + addPoint(x: number, y: number, attributes: Record): void { + this.records.push({ x, y, attributes }); + if (x < this.bbox.xmin) this.bbox.xmin = x; + if (x > this.bbox.xmax) this.bbox.xmax = x; + if (y < this.bbox.ymin) this.bbox.ymin = y; + if (y > this.bbox.ymax) this.bbox.ymax = y; + } + + build(): ShapefileComponents { + const n = this.records.length; + const bbox: BBox = n + ? this.bbox + : { xmin: 0, ymin: 0, xmax: 0, ymax: 0 }; + + // --- .shp --- + const recordBytes = n * (8 + POINT_CONTENT_BYTES); // 8-byte record header each + const shp = new Uint8Array(HEADER_SIZE + recordBytes); + const shpView = new DataView(shp.buffer); + writeShpHeader(shpView, shp.length, bbox); + + // --- .shx --- + const shx = new Uint8Array(HEADER_SIZE + n * 8); + const shxView = new DataView(shx.buffer); + writeShpHeader(shxView, shx.length, bbox); + + let shpOffsetWords = HEADER_SIZE / 2; + for (let i = 0; i < n; i++) { + const rec = this.records[i]; + const recOffsetBytes = shpOffsetWords * 2; + + // .shp record header (big-endian). + shpView.setInt32(recOffsetBytes, i + 1, false); // record number (1-based) + shpView.setInt32(recOffsetBytes + 4, POINT_CONTENT_WORDS, false); + // .shp record content (little-endian). + shpView.setInt32(recOffsetBytes + 8, SHAPE_TYPE_POINT, true); + shpView.setFloat64(recOffsetBytes + 12, rec.x, true); + shpView.setFloat64(recOffsetBytes + 20, rec.y, true); + + // .shx index entry (big-endian): offset + content length, both in words. + shxView.setInt32(HEADER_SIZE + i * 8, shpOffsetWords, false); + shxView.setInt32(HEADER_SIZE + i * 8 + 4, POINT_CONTENT_WORDS, false); + + shpOffsetWords += 4 + POINT_CONTENT_WORDS; // 8-byte header = 4 words + } + + return { + shp, + shx, + dbf: this.buildDbf(), + prj: new TextEncoder().encode(WGS84_WKT), + }; + } + + private buildDbf(): Uint8Array { + const n = this.records.length; + const encoder = new TextEncoder(); + + // Determine per-field width (and decimals for numeric fields) from data. + const specs = this.fields.map((field) => { + let length = 1; + let decimals = 0; + for (const rec of this.records) { + const formatted = formatDbfValue(rec.attributes[field.name], field.type); + if (field.type === "N") { + const dot = formatted.indexOf("."); + if (dot >= 0) decimals = Math.max(decimals, formatted.length - dot - 1); + } + length = Math.max(length, formatted.length); + } + const cap = field.type === "C" ? 254 : 19; + return { field, length: Math.min(length, cap), decimals: Math.min(decimals, 15) }; + }); + + const recordSize = + 1 + specs.reduce((sum, s) => sum + s.length, 0); // 1 = deletion flag + const headerSize = 32 + specs.length * 32 + 1; + + const dbf = new Uint8Array(headerSize + n * recordSize + 1); // +1 EOF marker + const view = new DataView(dbf.buffer); + + view.setUint8(0, 0x03); // dBASE III, no memo + view.setUint8(1, 95); // year since 1900 (fixed → deterministic) + view.setUint8(2, 1); // month + view.setUint8(3, 1); // day + view.setUint32(4, n, true); // record count + view.setUint16(8, headerSize, true); + view.setUint16(10, recordSize, true); + + // Field descriptors. + let pos = 32; + for (const spec of specs) { + const nameBytes = encoder.encode(spec.field.name).slice(0, 10); + dbf.set(nameBytes, pos); // null-padded by zero-initialised buffer + view.setUint8(pos + 11, spec.field.type.charCodeAt(0)); + view.setUint8(pos + 16, spec.length); + view.setUint8(pos + 17, spec.decimals); + pos += 32; + } + view.setUint8(pos, 0x0d); // header terminator + pos += 1; + + // Records. + for (const rec of this.records) { + dbf[pos] = 0x20; // not deleted + pos += 1; + for (const spec of specs) { + const raw = formatDbfValue(rec.attributes[spec.field.name], spec.field.type); + const field = padDbfField(raw, spec.length, spec.field.type); + dbf.set(encoder.encode(field), pos); + pos += spec.length; + } + } + dbf[pos] = 0x1a; // EOF marker + + return dbf; + } +} + +function formatDbfValue(value: unknown, type: DbfFieldType): string { + if (value === null || value === undefined) return ""; + if (type === "N") { + const n = typeof value === "number" ? value : Number(value); + return Number.isFinite(n) ? String(n) : ""; + } + return String(value); +} + +/** C fields are left-justified, N fields right-justified; both space-padded. */ +function padDbfField(value: string, length: number, type: DbfFieldType): string { + let v = value; + if (v.length > length) v = v.slice(0, length); + return type === "N" ? v.padStart(length, " ") : v.padEnd(length, " "); +} + +export function buildShapefileZipEntries( + components: ShapefileComponents, + baseName: string +): { name: string; data: Uint8Array }[] { + return [ + { name: `${baseName}.shp`, data: components.shp }, + { name: `${baseName}.shx`, data: components.shx }, + { name: `${baseName}.dbf`, data: components.dbf }, + { name: `${baseName}.prj`, data: components.prj }, + ]; +} diff --git a/src/utils/zip.ts b/src/utils/zip.ts new file mode 100644 index 0000000..c829fd9 --- /dev/null +++ b/src/utils/zip.ts @@ -0,0 +1,113 @@ +/** + * Minimal ZIP writer using the STORE method (no compression). Sufficient for + * bundling the components of a shapefile (`.shp`, `.shx`, `.dbf`, `.prj`) into a + * single `.zip` download. Shapefile components are already binary and small + * relative to the dataset, so deflate would add complexity for little gain. + */ + +export interface ZipEntry { + name: string; + data: Uint8Array; +} + +const CRC32_TABLE = (() => { + const table = new Uint32Array(256); + for (let n = 0; n < 256; n++) { + let c = n; + for (let k = 0; k < 8; k++) { + c = c & 1 ? 0xedb88320 ^ (c >>> 1) : c >>> 1; + } + table[n] = c >>> 0; + } + return table; +})(); + +export function crc32(bytes: Uint8Array): number { + let crc = 0xffffffff; + for (let i = 0; i < bytes.length; i++) { + crc = CRC32_TABLE[(crc ^ bytes[i]) & 0xff] ^ (crc >>> 8); + } + return (crc ^ 0xffffffff) >>> 0; +} + +/** DOS date/time. We use a fixed timestamp so output is deterministic. */ +const DOS_TIME = 0; // 00:00:00 +const DOS_DATE = 0x21; // 1980-01-01 + +/** Build a STORED (uncompressed) ZIP archive from the given entries. */ +export function createStoredZip(entries: ZipEntry[]): Uint8Array { + const encoder = new TextEncoder(); + const localParts: Uint8Array[] = []; + const centralParts: Uint8Array[] = []; + let offset = 0; + + for (const entry of entries) { + const nameBytes = encoder.encode(entry.name); + const crc = crc32(entry.data); + const size = entry.data.length; + + // Local file header (30 bytes + name). + const local = new DataView(new ArrayBuffer(30)); + local.setUint32(0, 0x04034b50, true); // signature + local.setUint16(4, 20, true); // version needed + local.setUint16(6, 0, true); // flags + local.setUint16(8, 0, true); // method = store + local.setUint16(10, DOS_TIME, true); + local.setUint16(12, DOS_DATE, true); + local.setUint32(14, crc, true); + local.setUint32(18, size, true); // compressed size + local.setUint32(22, size, true); // uncompressed size + local.setUint16(26, nameBytes.length, true); + local.setUint16(28, 0, true); // extra length + + const localHeader = new Uint8Array(local.buffer); + localParts.push(localHeader, nameBytes, entry.data); + + // Central directory header (46 bytes + name). + const central = new DataView(new ArrayBuffer(46)); + central.setUint32(0, 0x02014b50, true); // signature + central.setUint16(4, 20, true); // version made by + central.setUint16(6, 20, true); // version needed + central.setUint16(8, 0, true); // flags + central.setUint16(10, 0, true); // method + central.setUint16(12, DOS_TIME, true); + central.setUint16(14, DOS_DATE, true); + central.setUint32(16, crc, true); + central.setUint32(20, size, true); + central.setUint32(24, size, true); + central.setUint16(28, nameBytes.length, true); + central.setUint16(30, 0, true); // extra length + central.setUint16(32, 0, true); // comment length + central.setUint16(34, 0, true); // disk number + central.setUint16(36, 0, true); // internal attrs + central.setUint32(38, 0, true); // external attrs + central.setUint32(42, offset, true); // local header offset + + centralParts.push(new Uint8Array(central.buffer), nameBytes); + + offset += localHeader.length + nameBytes.length + entry.data.length; + } + + const centralSize = centralParts.reduce((n, p) => n + p.length, 0); + + // End of central directory record (22 bytes). + const eocd = new DataView(new ArrayBuffer(22)); + eocd.setUint32(0, 0x06054b50, true); + eocd.setUint16(4, 0, true); // disk number + eocd.setUint16(6, 0, true); // central dir start disk + eocd.setUint16(8, entries.length, true); + eocd.setUint16(10, entries.length, true); + eocd.setUint32(12, centralSize, true); + eocd.setUint32(16, offset, true); // central dir offset + eocd.setUint16(20, 0, true); // comment length + + const allParts = [...localParts, ...centralParts, new Uint8Array(eocd.buffer)]; + const total = allParts.reduce((n, p) => n + p.length, 0); + const out = new Uint8Array(total); + let pos = 0; + for (const part of allParts) { + out.set(part, pos); + pos += part.length; + } + return out; +} diff --git a/tests/e2e/export.spec.ts b/tests/e2e/export.spec.ts new file mode 100644 index 0000000..6acc942 --- /dev/null +++ b/tests/e2e/export.spec.ts @@ -0,0 +1,71 @@ +import { test, expect } from "@playwright/test"; +import { readFileSync } from "node:fs"; + +/** + * Drives the bulk export dialog end to end against mocked REST endpoints and + * verifies the downloaded CSV has the expected row count. The browser lacks the + * File System Access API in headless mode, so the Blob download fallback fires + * and Playwright captures it as a download event. + */ + +const ROW_COUNT = 250; + +test.describe("Bulk export", () => { + test.beforeEach(async ({ page }) => { + // Column schema. + await page.route("**/api/resources/schema", async (route) => { + await route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify([ + { name: "id", type: "number" }, + { name: "kwh", type: "number" }, + ]), + }); + }); + + // Chunked NDJSON export. Returns ROW_COUNT rows for the first chunk and an + // empty body afterwards (signals end of data). + await page.route("**/api/resources/export*", async (route) => { + const url = new URL(route.request().url()); + const offset = Number(url.searchParams.get("offset") ?? "0"); + let body = ""; + if (offset === 0) { + body = + Array.from({ length: ROW_COUNT }, (_, i) => + JSON.stringify({ id: i, kwh: i * 2 }) + ).join("\n") + "\n"; + } + await route.fulfill({ + status: 200, + contentType: "application/x-ndjson", + body, + }); + }); + }); + + test("exports a CSV with the expected row count", async ({ page }) => { + await page.goto("/export"); + await page.getByTestId("open-export").click(); + + // Wait for the schema-driven column checkboxes to render. + await expect(page.getByRole("checkbox").first()).toBeVisible(); + + const downloadPromise = page.waitForEvent("download"); + await page.getByRole("button", { name: "Export", exact: true }).click(); + const download = await downloadPromise; + + expect(download.suggestedFilename()).toBe("resource-export.csv"); + + const path = await download.path(); + expect(path).toBeTruthy(); + const content = readFileSync(path!, "utf-8"); + const lines = content.trimEnd().split("\r\n"); + + // Header + ROW_COUNT data rows. + expect(lines.length).toBe(ROW_COUNT + 1); + expect(lines[0]).toBe("id,kwh"); + expect(lines[1]).toBe("0,0"); + expect(lines[ROW_COUNT]).toBe(`${ROW_COUNT - 1},${(ROW_COUNT - 1) * 2}`); + }); +}); diff --git a/tests/unit/exportPipeline.test.ts b/tests/unit/exportPipeline.test.ts new file mode 100644 index 0000000..5a57841 --- /dev/null +++ b/tests/unit/exportPipeline.test.ts @@ -0,0 +1,183 @@ +import { describe, it, expect } from "vitest"; +import { ExportPipeline } from "@/services/exportPipeline"; +import type { FileWriter } from "@/utils/fileWriter"; +import type { ExportConfig, ExportEvent, ResourceRow } from "@/types/export"; + +/** In-memory FileWriter that records everything written. */ +function memoryWriter(): FileWriter & { bytes(): Uint8Array; text(): string } { + const chunks: Uint8Array[] = []; + let total = 0; + return { + usedFallback: true, + get bytesWritten() { + return total; + }, + async write(chunk) { + chunks.push(chunk); + total += chunk.byteLength; + }, + async close() {}, + async abort() {}, + bytes() { + const out = new Uint8Array(total); + let o = 0; + for (const c of chunks) { + out.set(c, o); + o += c.byteLength; + } + return out; + }, + text() { + return new TextDecoder().decode(this.bytes()); + }, + }; +} + +/** Build a fetch stub that serves NDJSON pages keyed by the `offset` param. */ +function ndjsonFetch(pages: ResourceRow[][]): typeof fetch { + return (async (input: RequestInfo | URL) => { + const url = new URL(String(input)); + const offset = Number(url.searchParams.get("offset") ?? "0"); + const limit = Number(url.searchParams.get("limit") ?? "10000"); + const pageIndex = Math.floor(offset / 10000); + const rows = pages[pageIndex] ?? []; + const body = rows.map((r) => JSON.stringify(r)).join("\n") + (rows.length ? "\n" : ""); + void limit; + return new Response(body, { + status: 200, + headers: { "content-type": "application/x-ndjson" }, + }); + }) as typeof fetch; +} + +async function runPipeline( + config: ExportConfig, + pages: ResourceRow[][] +): Promise<{ writer: ReturnType; events: ExportEvent[] }> { + const writer = memoryWriter(); + const events: ExportEvent[] = []; + const pipeline = new ExportPipeline(config, { + fetchFn: ndjsonFetch(pages), + createWriter: async () => writer, + baseUrl: "http://test.local", + }); + pipeline.on((e) => events.push(e)); + await pipeline.run(); + return { writer, events }; +} + +describe("ExportPipeline — CSV", () => { + it("writes a header and one line per row, stopping on a short chunk", async () => { + const { writer, events } = await runPipeline( + { format: "csv", columns: ["id", "kwh"], maxRows: 100 }, + [[{ id: 1, kwh: 10 }, { id: 2, kwh: 20 }, { id: 3, kwh: 30 }]] + ); + + expect(writer.text()).toBe("id,kwh\r\n1,10\r\n2,20\r\n3,30\r\n"); + const complete = events.find((e) => e.type === "complete"); + expect(complete).toMatchObject({ type: "complete", rowsWritten: 3 }); + }); + + it("applies client-side filters", async () => { + const { writer } = await runPipeline( + { + format: "csv", + columns: ["id", "kwh"], + maxRows: 100, + filters: [{ field: "kwh", operator: "gte", value: 20 }], + }, + [[{ id: 1, kwh: 10 }, { id: 2, kwh: 20 }, { id: 3, kwh: 30 }]] + ); + expect(writer.text()).toBe("id,kwh\r\n2,20\r\n3,30\r\n"); + }); + + it("spans multiple chunks via the sliding prefetch window", async () => { + const fullPage = Array.from({ length: 10000 }, (_, i) => ({ id: i })); + const lastPage = [{ id: 10000 }, { id: 10001 }]; + const { writer, events } = await runPipeline( + { format: "csv", columns: ["id"], maxRows: 1_000_000 }, + [fullPage, lastPage] + ); + const lines = writer.text().trimEnd().split("\r\n"); + // header + 10002 rows + expect(lines.length).toBe(1 + 10002); + const chunkCompletes = events.filter((e) => e.type === "chunkComplete"); + expect(chunkCompletes.length).toBe(2); + }); +}); + +describe("ExportPipeline — GeoJSON", () => { + it("wraps features in a FeatureCollection with truncated coordinates", async () => { + const { writer } = await runPipeline( + { + format: "geojson", + columns: ["id"], + maxRows: 100, + lonField: "lon", + latField: "lat", + }, + [[{ id: "a", lon: 1.23456789, lat: 2.3456789 }]] + ); + const parsed = JSON.parse(writer.text()); + expect(parsed.type).toBe("FeatureCollection"); + expect(parsed.features).toHaveLength(1); + expect(parsed.features[0].geometry.coordinates).toEqual([1.234567, 2.345678]); + expect(parsed.features[0].properties).toEqual({ id: "a" }); + }); + + it("emits a valid empty FeatureCollection when there are no rows", async () => { + const { writer } = await runPipeline( + { format: "geojson", columns: ["id"], maxRows: 100 }, + [[]] + ); + expect(JSON.parse(writer.text())).toEqual({ + type: "FeatureCollection", + features: [], + }); + }); +}); + +describe("ExportPipeline — shapefile", () => { + it("produces a zip archive containing the four shapefile components", async () => { + const { writer } = await runPipeline( + { + format: "shapefile", + columns: ["meter"], + maxRows: 100, + lonField: "lon", + latField: "lat", + }, + [[{ meter: "A", lon: 1, lat: 2 }, { meter: "B", lon: 3, lat: 4 }]] + ); + const bytes = writer.bytes(); + const view = new DataView(bytes.buffer); + // ZIP local file header signature. + expect(view.getUint32(0, true)).toBe(0x04034b50); + // Four entries in the central directory (EOCD is the last 22 bytes). + expect(view.getUint16(bytes.length - 22 + 10, true)).toBe(4); + }); +}); + +describe("ExportPipeline — cancellation", () => { + it("emits a cancelled event and stops when cancel() is called", async () => { + const writer = memoryWriter(); + const events: ExportEvent[] = []; + const pipeline = new ExportPipeline( + { format: "csv", columns: ["id"], maxRows: 1_000_000 }, + { + fetchFn: ndjsonFetch([ + Array.from({ length: 10000 }, (_, i) => ({ id: i })), + ]), + createWriter: async () => writer, + baseUrl: "http://test.local", + } + ); + pipeline.on((e) => { + events.push(e); + if (e.type === "chunkStart") pipeline.cancel(); + }); + await pipeline.run(); + expect(events.some((e) => e.type === "cancelled")).toBe(true); + expect(events.some((e) => e.type === "complete")).toBe(false); + }); +}); diff --git a/tests/unit/formatTransformers.test.ts b/tests/unit/formatTransformers.test.ts new file mode 100644 index 0000000..0fe54d2 --- /dev/null +++ b/tests/unit/formatTransformers.test.ts @@ -0,0 +1,93 @@ +import { describe, it, expect } from "vitest"; +import { + csvEscape, + csvHeader, + csvTransformer, + truncateCoord, + geoJsonTransformer, + shapefileTransformer, +} from "@/utils/formatTransformers"; + +describe("csvEscape (RFC 4180)", () => { + it("leaves simple values unquoted", () => { + expect(csvEscape("plain")).toBe("plain"); + expect(csvEscape(42)).toBe("42"); + }); + + it("quotes values containing a comma", () => { + expect(csvEscape("a,b")).toBe('"a,b"'); + }); + + it("escapes embedded double quotes by doubling them", () => { + expect(csvEscape('she said "hi"')).toBe('"she said ""hi"""'); + }); + + it("quotes values containing newlines or carriage returns", () => { + expect(csvEscape("line1\nline2")).toBe('"line1\nline2"'); + expect(csvEscape("a\r\nb")).toBe('"a\r\nb"'); + }); + + it("renders null/undefined as empty", () => { + expect(csvEscape(null)).toBe(""); + expect(csvEscape(undefined)).toBe(""); + }); +}); + +describe("csvHeader / csvTransformer", () => { + it("emits a CRLF-terminated header in column order", () => { + expect(csvHeader(["id", "name"])).toBe("id,name\r\n"); + }); + + it("projects only the requested columns in order", () => { + const row = { id: 1, name: "Meter, A", extra: "ignored" }; + expect(csvTransformer(row, ["name", "id"])).toBe('"Meter, A",1\r\n'); + }); + + it("renders missing columns as empty fields", () => { + expect(csvTransformer({ id: 1 }, ["id", "missing"])).toBe("1,\r\n"); + }); +}); + +describe("truncateCoord", () => { + it("truncates (not rounds) to 6 decimal places", () => { + expect(truncateCoord(12.34567891)).toBe(12.345678); + expect(truncateCoord(-0.1234569)).toBe(-0.123456); + }); + + it("handles non-finite input", () => { + expect(truncateCoord(NaN)).toBe(0); + expect(truncateCoord(Infinity)).toBe(0); + }); +}); + +describe("geoJsonTransformer", () => { + const geometry = { lonField: "lon", latField: "lat" }; + + it("builds a Point Feature with truncated coordinates", () => { + const row = { lon: 1.23456789, lat: -2.3456789, usage: 100, id: "m1" }; + const feature = geoJsonTransformer(row, ["id", "usage"], geometry); + expect(feature.type).toBe("Feature"); + expect(feature.geometry).toEqual({ + type: "Point", + coordinates: [1.234567, -2.345678], + }); + expect(feature.properties).toEqual({ id: "m1", usage: 100 }); + }); + + it("excludes geometry fields from properties", () => { + const row = { lon: 1, lat: 2, usage: 5 }; + const feature = geoJsonTransformer(row, ["lon", "lat", "usage"], geometry); + expect(feature.properties).toEqual({ usage: 5 }); + }); +}); + +describe("shapefileTransformer", () => { + it("extracts truncated lon/lat and attribute payload", () => { + const geometry = { lonField: "lon", latField: "lat" }; + const row = { lon: 10.1234567, lat: 20.7654321, meter: "X", kwh: 12 }; + const rec = shapefileTransformer(row, ["meter", "kwh"], geometry); + expect(rec.lon).toBe(10.123456); + expect(rec.lat).toBe(20.765432); + expect(rec.attributes).toEqual({ meter: "X", kwh: 12 }); + }); +}); diff --git a/tests/unit/ndjsonParser.test.ts b/tests/unit/ndjsonParser.test.ts new file mode 100644 index 0000000..73bcd6d --- /dev/null +++ b/tests/unit/ndjsonParser.test.ts @@ -0,0 +1,58 @@ +import { describe, it, expect } from "vitest"; +import { parseNdjsonStream } from "@/utils/ndjsonParser"; + +/** Build a ReadableStream that emits the given string pieces as UTF-8 bytes. */ +function streamFrom(pieces: string[]): ReadableStream { + const encoder = new TextEncoder(); + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i < pieces.length) { + controller.enqueue(encoder.encode(pieces[i++])); + } else { + controller.close(); + } + }, + }); +} + +async function collect( + body: ReadableStream, + opts?: Parameters[1] +): Promise { + const rows: unknown[] = []; + for await (const row of parseNdjsonStream(body, { gzip: false, ...opts })) { + rows.push(row); + } + return rows; +} + +describe("parseNdjsonStream (uncompressed)", () => { + it("yields one parsed object per line", async () => { + const rows = await collect( + streamFrom(['{"a":1}\n', '{"a":2}\n', '{"a":3}\n']) + ); + expect(rows).toEqual([{ a: 1 }, { a: 2 }, { a: 3 }]); + }); + + it("handles lines split across stream chunks", async () => { + const rows = await collect(streamFrom(['{"id":', '10}\n{"id":', "20}\n"])); + expect(rows).toEqual([{ id: 10 }, { id: 20 }]); + }); + + it("parses a trailing line without a final newline", async () => { + const rows = await collect(streamFrom(['{"x":1}\n{"x":2}'])); + expect(rows).toEqual([{ x: 1 }, { x: 2 }]); + }); + + it("skips blank lines", async () => { + const rows = await collect(streamFrom(['{"a":1}\n', "\n", '{"a":2}\n'])); + expect(rows).toEqual([{ a: 1 }, { a: 2 }]); + }); + + it("reports bytes read via onBytes", async () => { + let bytes = 0; + await collect(streamFrom(['{"a":1}\n']), { onBytes: (n) => (bytes += n) }); + expect(bytes).toBe(8); // '{"a":1}\n' is 8 bytes + }); +}); diff --git a/tests/unit/zipShapefile.test.ts b/tests/unit/zipShapefile.test.ts new file mode 100644 index 0000000..6e18bcd --- /dev/null +++ b/tests/unit/zipShapefile.test.ts @@ -0,0 +1,75 @@ +import { describe, it, expect } from "vitest"; +import { createStoredZip, crc32 } from "@/utils/zip"; +import { + ShapefilePointWriter, + buildShapefileZipEntries, +} from "@/utils/shapefileWriter"; + +describe("crc32", () => { + it("matches the known CRC-32 of 'The quick brown fox jumps over the lazy dog'", () => { + const bytes = new TextEncoder().encode( + "The quick brown fox jumps over the lazy dog" + ); + expect(crc32(bytes)).toBe(0x414fa339); + }); +}); + +describe("createStoredZip", () => { + it("writes a valid local-file and EOCD signature with the right entry count", () => { + const zip = createStoredZip([ + { name: "a.txt", data: new TextEncoder().encode("hello") }, + { name: "b.txt", data: new TextEncoder().encode("world") }, + ]); + const view = new DataView(zip.buffer); + // Local file header signature. + expect(view.getUint32(0, true)).toBe(0x04034b50); + // End-of-central-directory signature appears at the tail (22-byte record). + const eocdOffset = zip.length - 22; + expect(view.getUint32(eocdOffset, true)).toBe(0x06054b50); + // Total entries in the central directory. + expect(view.getUint16(eocdOffset + 10, true)).toBe(2); + }); +}); + +describe("ShapefilePointWriter", () => { + it("produces .shp/.shx with the shapefile magic and matching record count", () => { + const writer = new ShapefilePointWriter([ + { name: "meter", type: "C" }, + { name: "kwh", type: "N" }, + ]); + writer.addPoint(1.5, 2.5, { meter: "A", kwh: 100 }); + writer.addPoint(-3.25, 4.75, { meter: "BB", kwh: 9999 }); + + const { shp, shx, dbf, prj } = writer.build(); + + const shpView = new DataView(shp.buffer); + // File code 9994 (big-endian) and shape type 1 = Point (little-endian). + expect(shpView.getInt32(0, false)).toBe(9994); + expect(shpView.getInt32(32, true)).toBe(1); + // Header (100) + 2 records * (8 header + 20 content) = 156 bytes. + expect(shp.length).toBe(100 + 2 * 28); + + // .shx header length field equals the file length in 16-bit words. + const shxView = new DataView(shx.buffer); + expect(shxView.getInt32(24, false)).toBe(shx.length / 2); + + // DBF record count. + const dbfView = new DataView(dbf.buffer); + expect(dbfView.getUint32(4, true)).toBe(2); + + // PRJ is WGS84 WKT. + expect(new TextDecoder().decode(prj)).toContain("GCS_WGS_1984"); + }); + + it("bundles four components into the zip entries", () => { + const writer = new ShapefilePointWriter([{ name: "id", type: "C" }]); + writer.addPoint(0, 0, { id: "x" }); + const entries = buildShapefileZipEntries(writer.build(), "export"); + expect(entries.map((e) => e.name)).toEqual([ + "export.shp", + "export.shx", + "export.dbf", + "export.prj", + ]); + }); +}); From b72d76cac52938ce7ec37e4089c8d84b90a7f2ea Mon Sep 17 00:00:00 2001 From: real-venus Date: Thu, 25 Jun 2026 05:07:05 -0700 Subject: [PATCH 2/3] test: fix E2E export spec for cross-origin mocks and headless downloads (#60) The export dialog fetches from NEXT_PUBLIC_API_URL (cross-origin from the dev server), so the Playwright route mocks now return permissive CORS headers; without them the browser blocked the responses and the schema never loaded, leaving the Export button disabled until the job timed out. Also force the Blob download fallback in the spec (headless Chromium may expose showSaveFilePicker, whose native dialog can never appear and would hang the export), and drop the forbidden Accept-Encoding header from chunk requests since the browser manages it automatically. --- src/services/exportPipeline.ts | 4 +++- tests/e2e/export.spec.ts | 34 ++++++++++++++++++++++++++++++---- 2 files changed, 33 insertions(+), 5 deletions(-) diff --git a/src/services/exportPipeline.ts b/src/services/exportPipeline.ts index 612daff..e0907e0 100644 --- a/src/services/exportPipeline.ts +++ b/src/services/exportPipeline.ts @@ -182,7 +182,9 @@ export class ExportPipeline { const limit = Math.min(CHUNK_SIZE, this.maxRows - offset); return this.fetchFn(this.buildUrl(offset, limit), { signal: this.controller.signal, - headers: { Accept: "application/x-ndjson", "Accept-Encoding": "gzip" }, + // The browser manages Accept-Encoding (gzip) itself; setting it here is a + // no-op forbidden header, so we only advertise the body type we want. + headers: { Accept: "application/x-ndjson" }, }); } diff --git a/tests/e2e/export.spec.ts b/tests/e2e/export.spec.ts index 6acc942..7fd1611 100644 --- a/tests/e2e/export.spec.ts +++ b/tests/e2e/export.spec.ts @@ -10,13 +10,36 @@ import { readFileSync } from "node:fs"; const ROW_COUNT = 250; +// The export dialog talks to NEXT_PUBLIC_API_URL (default http://localhost:4000), +// which is cross-origin from the dev server. Mocked responses therefore need +// permissive CORS headers so the browser lets the page read them. +const CORS_HEADERS = { + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Headers": "*", + "Access-Control-Allow-Methods": "GET,OPTIONS", +}; + test.describe("Bulk export", () => { test.beforeEach(async ({ page }) => { + // Force the Blob download fallback: headless Chromium may expose + // showSaveFilePicker, whose native dialog can never appear and would hang + // the export. Removing it makes the pipeline emit a capturable download. + await page.addInitScript(() => { + try { + Object.defineProperty(window, "showSaveFilePicker", { + value: undefined, + configurable: true, + }); + } catch { + // ignore if the property is locked + } + }); + // Column schema. await page.route("**/api/resources/schema", async (route) => { await route.fulfill({ status: 200, - contentType: "application/json", + headers: { "content-type": "application/json", ...CORS_HEADERS }, body: JSON.stringify([ { name: "id", type: "number" }, { name: "kwh", type: "number" }, @@ -38,7 +61,7 @@ test.describe("Bulk export", () => { } await route.fulfill({ status: 200, - contentType: "application/x-ndjson", + headers: { "content-type": "application/x-ndjson", ...CORS_HEADERS }, body, }); }); @@ -48,11 +71,14 @@ test.describe("Bulk export", () => { await page.goto("/export"); await page.getByTestId("open-export").click(); - // Wait for the schema-driven column checkboxes to render. + // Wait for the schema-driven column checkboxes to render (proves the schema + // fetch resolved) and the Export button to become enabled. await expect(page.getByRole("checkbox").first()).toBeVisible(); + const exportButton = page.getByRole("button", { name: "Export", exact: true }); + await expect(exportButton).toBeEnabled(); const downloadPromise = page.waitForEvent("download"); - await page.getByRole("button", { name: "Export", exact: true }).click(); + await exportButton.click(); const download = await downloadPromise; expect(download.suggestedFilename()).toBe("resource-export.csv"); From d0b83cfe0bea4500a64d6b12d38cda72de8bdd2d Mon Sep 17 00:00:00 2001 From: real-venus Date: Thu, 25 Jun 2026 05:40:40 -0700 Subject: [PATCH 3/3] test: block service worker in E2E export spec so route mocks apply (#60) The app's service worker intercepts fetches once it controls the page, and Playwright's page.route does not see service-worker-originated requests. The mocked /api/resources endpoints were therefore bypassed and hit the real (absent) network, failing with net::ERR_FAILED so the schema never loaded and the export hung until timeout. Blocking the service worker for this spec lets the route mocks intercept every request. Verified: full E2E suite 22/22 green. --- tests/e2e/export.spec.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/e2e/export.spec.ts b/tests/e2e/export.spec.ts index 7fd1611..6aac58b 100644 --- a/tests/e2e/export.spec.ts +++ b/tests/e2e/export.spec.ts @@ -20,6 +20,11 @@ const CORS_HEADERS = { }; test.describe("Bulk export", () => { + // The app registers a service worker that intercepts fetches. Playwright's + // page.route does NOT see service-worker-originated requests, so without this + // the mocked endpoints would be bypassed and hit the real (absent) network. + test.use({ serviceWorkers: "block" }); + test.beforeEach(async ({ page }) => { // Force the Blob download fallback: headless Chromium may expose // showSaveFilePicker, whose native dialog can never appear and would hang