Skip to content

feat: stream detour results per-station via NDJSON #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
GeiserX merged 4 commits into from
Apr 6, 2026
Merged

feat: stream detour results per-station via NDJSON #21

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
366cde0
feat: stream detour results per-station via NDJSON
GeiserX Apr 6, 2026
39d0533
fix: lift 50-station cap, cancel stale flush, preserve streamed results
GeiserX Apr 6, 2026
f6b7167
fix: cap stations at 500, validate bounds, observe request.signal
GeiserX Apr 6, 2026
8674ed0
fix: backfill unseen stations, chunk <=500, propagate abort to Valhalla
GeiserX Apr 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions package-lock.json
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "pumperly",
"version": "1.2.0",
"version": "1.3.0",
"description": "Open-source energy route planner for fuel and electric vehicles. Self-hostable.",
"private": true,
"scripts": {
Expand Down
221 changes: 115 additions & 106 deletions src/app/api/route-detour/route.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,23 +4,29 @@ import { getRouteDuration } from "@/lib/valhalla";

const CONCURRENCY = 8;

const detourResultSchema = z.object({
id: z.string(),
detourMin: z.number(),
});

const stationSchema = z.object({
id: z.string(),
lon: z.number(),
lat: z.number(),
routeFraction: z.number(),
lon: z.number().min(-180).max(180),
lat: z.number().min(-90).max(90),
routeFraction: z.number().min(0).max(1),
});

const coordSchema = z.tuple([
z.number().min(-180).max(180),
z.number().min(-90).max(90),
]);

const bodySchema = z.object({
stations: z.array(stationSchema).min(1).max(50),
routeCoordinates: z.array(z.tuple([z.number(), z.number()])).min(2).max(3000),
stations: z.array(stationSchema).min(1).max(500),
routeCoordinates: z.array(coordSchema).min(2).max(3000),
});

interface DetourResult {
id: string;
detourMin: number;
}

/** Stream per-station detour times as NDJSON. Each line: `{"id":"…","detourMin":…}` */
export async function POST(request: NextRequest) {
let body: unknown;
try {
Expand All @@ -40,114 +46,117 @@ export async function POST(request: NextRequest) {
const { stations, routeCoordinates } = parseResult.data;
const numCoords = routeCoordinates.length;

try {
// Pre-compute cumulative segment lengths for consistent length-based fractions.
// routeFraction from PostGIS (ST_LineLocatePoint) is a fraction of total line
// length, so we need length-based — not vertex-index-based — exit/rejoin windows.
const cumLen: number[] = [0];
for (let i = 1; i < numCoords; i++) {
const dx = routeCoordinates[i][0] - routeCoordinates[i - 1][0];
const dy = routeCoordinates[i][1] - routeCoordinates[i - 1][1];
cumLen.push(cumLen[i - 1] + Math.sqrt(dx * dx + dy * dy));
}
const totalLen = cumLen[numCoords - 1];

// Binary-search: find the vertex index where cumLen >= targetLen
function distToIndex(targetLen: number): number {
let lo = 0, hi = numCoords - 1;
while (lo < hi) {
const mid = (lo + hi) >> 1;
if (cumLen[mid] < targetLen) lo = mid + 1;
else hi = mid;
}
return lo;
// Pre-compute cumulative segment lengths for consistent length-based fractions.
// routeFraction from PostGIS (ST_LineLocatePoint) is a fraction of total line
// length, so we need length-based — not vertex-index-based — exit/rejoin windows.
const cumLen: number[] = [0];
for (let i = 1; i < numCoords; i++) {
const dx = routeCoordinates[i][0] - routeCoordinates[i - 1][0];
const dy = routeCoordinates[i][1] - routeCoordinates[i - 1][1];
cumLen.push(cumLen[i - 1] + Math.sqrt(dx * dx + dy * dy));
}
const totalLen = cumLen[numCoords - 1];

// Binary-search: find the vertex index where cumLen >= targetLen
function distToIndex(targetLen: number): number {
let lo = 0, hi = numCoords - 1;
while (lo < hi) {
const mid = (lo + hi) >> 1;
if (cumLen[mid] < targetLen) lo = mid + 1;
else hi = mid;
}
return lo;
}

// Process stations with controlled concurrency
const results: DetourResult[] = [];
const queue = [...stations];
async function processStation(
s: z.infer<typeof stationSchema>,
): Promise<{ id: string; detourMin: number }> {
try {
if (totalLen === 0) return { id: s.id, detourMin: 0 };

async function processStation(
s: z.infer<typeof stationSchema>,
): Promise<DetourResult> {
try {
if (totalLen === 0) return { id: s.id, detourMin: 0 };
const stationDist = s.routeFraction * totalLen;
const windowDist = totalLen * 0.03;
const exitDist = Math.max(0, stationDist - windowDist);
const rejoinDist = Math.min(totalLen, stationDist + windowDist);

// Symmetric window: 3% of route length each side.
const stationDist = s.routeFraction * totalLen;
const windowDist = totalLen * 0.03;
const exitDist = Math.max(0, stationDist - windowDist);
const rejoinDist = Math.min(totalLen, stationDist + windowDist);
let exitIdx = distToIndex(exitDist);
let rejoinIdx = Math.min(numCoords - 1, distToIndex(rejoinDist));

let exitIdx = distToIndex(exitDist);
let rejoinIdx = Math.min(numCoords - 1, distToIndex(rejoinDist));

// If the window collapsed to a single vertex (sparse/downsampled geometry),
// widen to guarantee at least two distinct vertices for Valhalla routing.
if (exitIdx === rejoinIdx) {
exitIdx = Math.max(0, exitIdx - 1);
rejoinIdx = Math.min(numCoords - 1, rejoinIdx + 1);
if (exitIdx === rejoinIdx) {
exitIdx = Math.max(0, exitIdx - 1);
rejoinIdx = Math.min(numCoords - 1, rejoinIdx + 1);
if (exitIdx === rejoinIdx) {
return { id: s.id, detourMin: -1 };
}
}

const exitCoord = routeCoordinates[exitIdx];
const rejoinCoord = routeCoordinates[rejoinIdx];

// Two parallel Valhalla calls: detour leg and direct baseline.
// The old linear-interpolation baseline (routeDuration * fraction) assumed
// uniform speed, which is wildly wrong on long mixed highway/town routes.
const [detourDuration, baselineDuration] = await Promise.all([
// exit → station → rejoin
getRouteDuration([
{ lat: exitCoord[1], lon: exitCoord[0] },
{ lat: s.lat, lon: s.lon },
{ lat: rejoinCoord[1], lon: rejoinCoord[0] },
]),
// exit → rejoin (actual road time for this segment)
getRouteDuration([
{ lat: exitCoord[1], lon: exitCoord[0] },
{ lat: rejoinCoord[1], lon: rejoinCoord[0] },
]),
]);

if (detourDuration == null || baselineDuration == null) {
return { id: s.id, detourMin: -1 };
}
}

// Detour = via-station time minus direct time for same segment
const detourSec = Math.max(0, detourDuration - baselineDuration);
const detourMin = Math.round(detourSec / 6) / 10; // 1 decimal place

return { id: s.id, detourMin };
} catch (err) {
console.warn(`[route-detour] station ${s.id} failed:`, err);
const exitCoord = routeCoordinates[exitIdx];
const rejoinCoord = routeCoordinates[rejoinIdx];

const [detourDuration, baselineDuration] = await Promise.all([
getRouteDuration([
{ lat: exitCoord[1], lon: exitCoord[0] },
{ lat: s.lat, lon: s.lon },
{ lat: rejoinCoord[1], lon: rejoinCoord[0] },
], "auto", signal),
getRouteDuration([
{ lat: exitCoord[1], lon: exitCoord[0] },
{ lat: rejoinCoord[1], lon: rejoinCoord[0] },
], "auto", signal),
]);
Comment on lines +96 to +106
Copy link
Copy Markdown

@coderabbitai coderabbitai bot Apr 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

wc -l src/app/api/route-detour/route.ts

Repository: GeiserX/Pumperly

Length of output: 97

🏁 Script executed:

cat -n src/app/api/route-detour/route.ts | head -150

Repository: GeiserX/Pumperly

Length of output: 5600

Abort queued detours when the request is canceled.

The handler ignores request.signal, so canceled requests continue draining the queue and triggering upstream Valhalla work. With up to 50 stations and 2 calls per station, a canceled request can spawn ~100 unnecessary calls. Check request.signal.aborted before dequeuing (line 121), and pass the signal to getRouteDuration() if it accepts abort handling.

Additionally, the streamed response objects must be validated with Zod before encoding (line 124), per the API validation requirement.

Also applies to: 115–129

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@src/app/api/route-detour/route.ts` around lines 85 - 95, The handler
currently ignores request.signal and continues dequeuing/processing detours;
update the detour loop to check request.signal.aborted before calling dequeue
and before each expensive call, and pass that signal into getRouteDuration
(e.g., getRouteDuration(..., { signal: request.signal })) so in-flight work can
be aborted; additionally validate any streamed response objects with your Zod
schema (the same schema used for API responses) before encoding/queueing them to
the stream to ensure only validated objects are emitted (apply these changes
around the detour processing loop and calls to getRouteDuration and the stream
encoder).


if (detourDuration == null || baselineDuration == null) {
return { id: s.id, detourMin: -1 };
}
}

// Run with concurrency limit
const workers: Promise<void>[] = [];
for (let i = 0; i < CONCURRENCY; i++) {
workers.push(
(async () => {
while (queue.length > 0) {
const station = queue.shift()!;
const result = await processStation(station);
results.push(result);
}
})(),
);
}
await Promise.all(workers);
const detourSec = Math.max(0, detourDuration - baselineDuration);
const detourMin = Math.round(detourSec / 6) / 10;

return NextResponse.json({ detours: results });
} catch (err) {
console.error("[route-detour] failed:", err);
return NextResponse.json(
{ error: "Internal server error" },
{ status: 500 },
);
return { id: s.id, detourMin };
} catch (err) {
console.warn(`[route-detour] station ${s.id} failed:`, err);
return { id: s.id, detourMin: -1 };
}
}

const encoder = new TextEncoder();
const queue = [...stations];
const { signal } = request;

const stream = new ReadableStream({
async start(controller) {
// Drain queue on client disconnect so workers stop picking up new stations
const onAbort = () => { queue.length = 0; };
signal.addEventListener("abort", onAbort);
try {
const workers: Promise<void>[] = [];
for (let i = 0; i < CONCURRENCY; i++) {
workers.push(
(async () => {
while (queue.length > 0) {
const station = queue.shift()!;
const result = detourResultSchema.parse(await processStation(station));
if (signal.aborted) return;
controller.enqueue(encoder.encode(JSON.stringify(result) + "\n"));
}
})(),
);
}
await Promise.all(workers);
} catch (err) {
console.error("[route-detour] stream failed:", err);
} finally {
signal.removeEventListener("abort", onAbort);
controller.close();
}
},
});

return new Response(stream, {
headers: {
"Content-Type": "application/x-ndjson",
"Transfer-Encoding": "chunked",
"Cache-Control": "no-cache",
},
});
}
Loading
Loading