gods-eye-view/server/providers/wind.js

import { fetchGfsWind } from './wind/gfs.js';
import { fetchIfsWind } from './wind/ifs.js';
import { decodeWindGribMessage } from './wind/decode.js';

/** Serve bounded, single-flight, per-model forecast snapshots from fixed providers. */
export function windProxy({
  fetchImpl = fetch,
  now = () => Date.now(),
  decodeImpl = decodeWindGribMessage,
  targetDx = 1,
  ttlMs = 3600_000,
  timeoutMs = 40_000,
  models = { gfs: fetchGfsWind, ifs: fetchIfsWind },
} = {}) {
  const caches = new Map();
  const loadings = new Map();
  const grids = new Map();
  const attempts = new Map();
  const unavailable = (model) => ({
    manifest: {
      model,
      schemaVersion: 1,
      unavailable: true,
      stale: true,
      reason: 'Wind upstream unavailable',
    },
  });
  const sendJson = (res, value, status = 200) => {
    res.writeHead(status, {
      'Content-Type': 'application/json',
      'Cache-Control': 'no-store',
    });
    res.end(JSON.stringify(value));
  };
  function refresh(model) {
    const controller = new AbortController();
    const operation = { controller, waiters: 0, promise: null };
    loadings.set(model, operation);
    attempts.set(model, now());
    operation.promise = (async () => {
      const timer = setTimeout(() => controller.abort(), timeoutMs);
      try {
        const value = await models[model]({
          fetchImpl,
          now,
          decodeImpl,
          targetDx,
          signal: controller.signal,
        });
        controller.signal.throwIfAborted();
        const { grid, cycle } = value;
        if (!grid.u.every(Number.isFinite) || !grid.v.every(Number.isFinite))
          throw new Error('Invalid wind grid');
        const id = `${model}-${cycle.date}-${cycle.hour}-f${cycle.forecastHour || 0}-${targetDx}`;
        const manifest = {
          schemaVersion: 1,
          model,
          cycle,
          fetchedAt: now(),
          level: value.level,
          units: value.units,
          grid: {
            nx: grid.nx,
            ny: grid.ny,
            lo1: grid.lo1,
            la1: grid.la1,
            dx: grid.dx,
            dy: grid.dy,
          },
          stale: false,
          unavailable: false,
          reason: null,
          gridUrl: `/api/wind/grid/${id}.bin?model=${model}`,
        };
        const state = { id, grid, manifest, fetchedAt: now() };
        caches.set(model, state);
        // Retain the previous issued grid as well to cover a manifest/grid rollover.
        const history = grids.get(model) || new Map();
        history.set(id, state);
        while (history.size > 2) history.delete(history.keys().next().value);
        grids.set(model, history);
        return state;
      } catch {
        if (controller.signal.aborted && operation.waiters === 0) {
          attempts.delete(model);
          return unavailable(model);
        }
        const old = caches.get(model);
        if (old) {
          old.manifest = {
            ...old.manifest,
            stale: true,
            reason: 'Wind upstream unavailable',
          };
          return old;
        }
        return unavailable(model);
      } finally {
        clearTimeout(timer);
        controller.abort(); // Cancel a sibling request if the other component failed.
        if (loadings.get(model) === operation) loadings.delete(model);
      }
    })();
    return operation;
  }
  const handler = async (req, res) => {
    const url = new URL(req.url, 'http://localhost');
    const model = url.searchParams.get('model') || 'gfs';
    if (req.method !== 'GET')
      return sendJson(res, { error: 'method_not_allowed' }, 405);
    if (!['gfs', 'ifs'].includes(model) || !Object.hasOwn(models, model))
      return sendJson(res, { error: 'unknown_model' }, 400);
    if (url.pathname.startsWith('/grid/')) {
      const id = url.pathname.slice(6).replace(/\.bin$/, '');
      const state =
        url.pathname === `/grid/${id}.bin` && grids.get(model)?.get(id);
      if (!state) return sendJson(res, { error: 'unknown_grid' }, 404);
      const bytes = Buffer.concat([
        Buffer.from(
          state.grid.u.buffer,
          state.grid.u.byteOffset,
          state.grid.u.byteLength,
        ),
        Buffer.from(
          state.grid.v.buffer,
          state.grid.v.byteOffset,
          state.grid.v.byteLength,
        ),
      ]);
      res.writeHead(200, {
        'Content-Type': 'application/octet-stream',
        'Cache-Control': 'public, max-age=3600, immutable',
      });
      return res.end(bytes);
    }
    if (!['/', '/manifest', '/status'].includes(url.pathname))
      return sendJson(res, { error: 'not_found' }, 404);
    let state = caches.get(model);
    if (!state || now() - state.fetchedAt >= ttlMs) {
      let operation = loadings.get(model);
      if (!operation && now() - (attempts.get(model) ?? -Infinity) >= 60_000)
        operation = refresh(model);
      if (operation) {
        let disconnected = false;
        operation.waiters += 1;
        const close = () => {
          disconnected = true;
          if (--operation.waiters === 0) operation.controller.abort();
        };
        res.once?.('close', close);
        try {
          state = await operation.promise;
        } finally {
          res.removeListener?.('close', close);
          if (!disconnected) operation.waiters -= 1;
        }
        if (disconnected) return;
      }
    }
    const manifest = state?.manifest || unavailable(model).manifest;
    if (url.pathname === '/status') {
      const { gridUrl, ...status } = manifest;
      return sendJson(res, status);
    }
    return sendJson(res, manifest);
  };
  return {
    name: 'wind',
    configureServer({ middlewares }) {
      middlewares.use('/api/wind', handler);
    },
    configurePreviewServer({ middlewares }) {
      middlewares.use('/api/wind', handler);
    },
  };
}