gods-eye-view/server/providers/mapillary/features.js

import { readRequestBody } from '../common/request.js';
import {
  FEATURE_TILE_ZOOM,
  QUERY_MAX_BODY_BYTES,
  QUERY_MAX_FEATURES,
  QUERY_MAX_TILES,
  QUERY_TILE_CONCURRENCY,
  mapillaryToken,
} from './constants.js';
import { fetchTile, TileRequestError } from './tiles.js';
import { decodeFeatureRows } from './decode.js';
import {
  normalizeBbox,
  tilesForBbox,
} from '../../../src/layers/mapillary/tileMath.js';
import {
  createValueMatcher,
  normalizeValuePattern,
} from '../../../src/layers/mapillary/values.js';

function sendJson(res, status, payload) {
  res.statusCode = status;
  res.setHeader('Content-Type', 'application/json; charset=utf-8');
  res.setHeader('Cache-Control', 'no-store');
  res.end(JSON.stringify(payload));
}

/** Parse an ISO date (or epoch ms) into epoch ms, or null when absent/invalid. */
function parseEpochMs(value) {
  if (value === null || value === undefined || value === '') return null;
  if (typeof value === 'number') return Number.isFinite(value) ? value : null;
  const parsed = Date.parse(String(value));
  return Number.isFinite(parsed) ? parsed : null;
}

/**
 * Validate a feature query body into an executable request.
 * @returns {{ok:true, request:object}|{ok:false, status:number, error:string, detail?:object}}
 */
export function normalizeFeatureQuery(body) {
  const bbox = normalizeBbox(body?.bbox);
  if (!bbox)
    return {
      ok: false,
      status: 400,
      error: 'A bbox [west,south,east,north] is required',
    };
  const layer = body?.layer === 'signs' ? 'signs' : 'points';
  const values = Array.isArray(body?.values)
    ? body.values.map(normalizeValuePattern).filter(Boolean).slice(0, 64)
    : [];
  const since = parseEpochMs(body?.seenAfter);
  const until = parseEpochMs(body?.seenBefore);
  const { tiles, truncated, total } = tilesForBbox(bbox, FEATURE_TILE_ZOOM, {
    limit: QUERY_MAX_TILES,
  });
  if (truncated)
    return {
      ok: false,
      status: 413,
      error: 'area_too_large',
      detail: { tiles: total, limit: QUERY_MAX_TILES },
    };
  return { ok: true, request: { bbox, layer, values, since, until, tiles } };
}

/**
 * POST /api/mapillary/features — fan a bbox out over z14 map-feature tiles
 * and stream the matching features back as NDJSON, one line per tile, so
 * the globe fills in progressively while the rest of the city downloads.
 */
export async function handleFeatureQuery(req, res) {
  if (req.method !== 'POST')
    return sendJson(res, 405, { error: 'Method not allowed' });
  if (!mapillaryToken())
    return sendJson(res, 503, { error: 'no_key', keyRequired: true });
  let body;
  try {
    body = JSON.parse(
      (await readRequestBody(req, QUERY_MAX_BODY_BYTES)) || '{}',
    );
  } catch (error) {
    return sendJson(res, 400, { error: error?.message || 'Invalid JSON body' });
  }
  const normalized = normalizeFeatureQuery(body);
  if (!normalized.ok)
    return sendJson(res, normalized.status, {
      error: normalized.error,
      ...normalized.detail,
    });
  const { layer, values, since, until, tiles, bbox } = normalized.request;
  const matcher = createValueMatcher(values);

  const abandoned = new AbortController();
  req.on?.('aborted', () => abandoned.abort());
  res.on?.('close', () => abandoned.abort());

  res.statusCode = 200;
  res.setHeader('Content-Type', 'application/x-ndjson; charset=utf-8');
  res.setHeader('Cache-Control', 'no-store');
  res.setHeader('X-Accel-Buffering', 'no');
  const write = (line) => {
    if (res.writableEnded || abandoned.signal.aborted) return false;
    res.write(`${JSON.stringify(line)}\n`);
    return true;
  };
  write({
    type: 'start',
    tiles: tiles.length,
    zoom: FEATURE_TILE_ZOOM,
    layer,
    values,
    bbox,
  });

  let total = 0;
  let failed = 0;
  let truncated = false;
  let cursor = 0;
  const worker = async () => {
    while (cursor < tiles.length && !abandoned.signal.aborted) {
      if (total >= QUERY_MAX_FEATURES) {
        truncated = true;
        return;
      }
      const index = cursor++;
      const tile = tiles[index];
      try {
        const { bytes, source } = await fetchTile(
          { layer, z: tile.z, x: tile.x, y: tile.y },
          { signal: abandoned.signal },
        );
        const rows = decodeFeatureRows(
          bytes,
          { ...tile, layer },
          { matcher, since, until },
        );
        total += rows.length;
        write({ type: 'tile', i: index, x: tile.x, y: tile.y, source, rows });
      } catch (error) {
        if (abandoned.signal.aborted) return;
        failed++;
        write({
          type: 'tile',
          i: index,
          x: tile.x,
          y: tile.y,
          rows: [],
          error:
            error instanceof TileRequestError
              ? error.message
              : `tile unavailable (${error?.status || error?.message || 'error'})`,
        });
      }
    }
  };
  await Promise.all(
    Array.from(
      { length: Math.min(QUERY_TILE_CONCURRENCY, tiles.length) },
      worker,
    ),
  );
  write({ type: 'done', total, failed, truncated, tiles: tiles.length });
  if (!res.writableEnded) res.end();
}