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..e0907e0 --- /dev/null +++ b/src/services/exportPipeline.ts @@ -0,0 +1,421 @@ +"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, + // 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" }, + }); + } + + /** 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..6aac58b --- /dev/null +++ b/tests/e2e/export.spec.ts @@ -0,0 +1,102 @@ +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; + +// 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", () => { + // 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 + // 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, + headers: { "content-type": "application/json", ...CORS_HEADERS }, + 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, + headers: { "content-type": "application/x-ndjson", ...CORS_HEADERS }, + 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 (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 exportButton.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", + ]); + }); +});