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

import { validTransitIdentifier } from '../../src/data/transitProxy.js';
export { validTransitIdentifier } from '../../src/data/transitProxy.js';
/** Bounded, process-local operator position history. No upstream or scene access. */
export const TRANSIT_HISTORY_LIMITS = Object.freeze({
  retentionMs: 900000,
  fixes: 128,
  feedBytes: 16 * 1024 * 1024,
  processBytes: 64 * 1024 * 1024,
  keys: 15000,
  identifierBytes: 256,
  responseBytes: 32768,
});
const FIX_BYTES = 32,
  CHUNK = 8,
  EPOCH_SLOTS = 8,
  EPOCH_BYTES = 544;
const META_BYTES = 64;
const encoder = new TextEncoder(),
  decoder = new TextDecoder();
// All stores in this process share the hard process allocation ceiling.
const processStorage = { bytes: 0, keys: 0, stores: new Set() };

function contextOf(record, mode) {
  const trip = typeof record.tripId === 'string' ? record.tripId : '';
  const route = typeof record.routeId === 'string' ? record.routeId : '';
  if (encoder.encode(trip).length > 256 || encoder.encode(route).length > 256)
    return null;
  return { trip, route, mode: mode || 'unknown' };
}
function contextAt(entry, slot) {
  const offset = 0,
    view = entry.contexts[slot];
  const bytes = new Uint8Array(view.buffer);
  const length = view.getUint16(offset + 4),
    routeLength = view.getUint16(offset + 6);
  return {
    id: view.getUint32(offset),
    trip: decoder.decode(bytes.subarray(offset + 16, offset + 16 + length)),
    route: decoder.decode(
      bytes.subarray(offset + 272, offset + 272 + routeLength),
    ),
    mode: decoder.decode(
      bytes.subarray(offset + 528, offset + 528 + view.getUint8(offset + 8)),
    ),
  };
}
function storeContext(entry, context, id) {
  const slot = (id - 1) % EPOCH_SLOTS,
    offset = 0;
  const trip = encoder.encode(context.trip),
    route = encoder.encode(context.route),
    mode = encoder.encode(context.mode).subarray(0, 16);
  const view = entry.contexts[slot],
    bytes = new Uint8Array(view.buffer);
  view.setUint32(offset, id);
  view.setUint16(offset + 4, trip.length);
  view.setUint16(offset + 6, route.length);
  view.setUint8(offset + 8, mode.length);
  bytes.set(trip, offset + 16);
  bytes.set(route, offset + 272);
  bytes.set(mode, offset + 528);
}
function offset(entry, index) {
  return ((entry.head + index) % entry.capacity) * FIX_BYTES;
}
function read(entry, index) {
  const at = offset(entry, index),
    v = entry.fixes;
  return [
    v.getFloat64(at),
    v.getFloat64(at + 8),
    v.getFloat64(at + 16),
    v.getUint32(at + 24),
    v.getUint32(at + 28),
  ];
}
function firstTime(entry) {
  return entry.count ? entry.fixes.getFloat64(offset(entry, 0)) : Infinity;
}
function drop(entry) {
  entry.head = (entry.head + 1) % entry.capacity;
  entry.count--;
  entry.truncated = true;
}

// Indexed eviction queues keep removal and receipt-order updates logarithmic.
function evictionQueue() {
  const items = [],
    positions = new Map();
  const less = (a, b) =>
    a.observedAt < b.observedAt ||
    (a.observedAt === b.observedAt && a.key < b.key);
  function swap(a, b) {
    [items[a], items[b]] = [items[b], items[a]];
    positions.set(items[a], a);
    positions.set(items[b], b);
  }
  function repair(index) {
    while (index > 0) {
      const parent = (index - 1) >>> 1;
      if (!less(items[index], items[parent])) break;
      swap(index, parent);
      index = parent;
    }
    for (;;) {
      let child = index * 2 + 1;
      if (child >= items.length) break;
      if (child + 1 < items.length && less(items[child + 1], items[child]))
        child++;
      if (!less(items[child], items[index])) break;
      swap(index, child);
      index = child;
    }
  }
  function remove(entry) {
    const index = positions.get(entry);
    if (index === undefined) return;
    positions.delete(entry);
    const last = items.pop();
    if (index < items.length) {
      items[index] = last;
      positions.set(last, index);
      repair(index);
    }
  }
  return {
    remove,
    add(entry) {
      remove(entry);
      positions.set(entry, items.length);
      items.push(entry);
      repair(items.length - 1);
    },
    oldest(except) {
      if (items[0] !== except) return items[0] || null;
      if (!items[1]) return null;
      return items[2] && less(items[2], items[1]) ? items[2] : items[1];
    },
    get size() {
      return items.length;
    },
  };
}

export function createTransitHistory({
  limits = {},
  storage = processStorage,
  sweep = true,
} = {}) {
  const policy = { ...TRANSIT_HISTORY_LIMITS };
  for (const [key, limit] of Object.entries(limits))
    if (key in policy && Number.isFinite(limit) && limit > 0)
      policy[key] = Math.min(policy[key], Math.floor(limit));
  const entries = new Map(),
    feedBytes = new Map(),
    eviction = evictionQueue(),
    feedEviction = new Map();
  storage.feedBytes ||= new Map();
  storage.lossUntil ||= new Map();
  let protectedEntries = null;
  function queue(entry) {
    if (!feedEviction.has(entry.feed))
      feedEviction.set(entry.feed, evictionQueue());
    eviction.add(entry);
    feedEviction.get(entry.feed).add(entry);
  }
  function unqueue(entry) {
    eviction.remove(entry);
    const feedQueue = feedEviction.get(entry.feed);
    feedQueue?.remove(entry);
    if (feedQueue?.size === 0) feedEviction.delete(entry.feed);
  }
  function recordLoss(feed, until) {
    storage.lossUntil.set(
      feed,
      Math.max(storage.lossUntil.get(feed) || 0, until),
    );
  }
  let prunedAt = -Infinity;
  let allocatedBytes = 0,
    timer = null,
    closed = false;
  function account(feed, delta) {
    allocatedBytes += delta;
    storage.bytes += delta;
    storage.feedBytes.set(feed, (storage.feedBytes.get(feed) || 0) + delta);
    if (storage.feedBytes.get(feed) === 0) storage.feedBytes.delete(feed);
    feedBytes.set(feed, (feedBytes.get(feed) || 0) + delta);
    if (feedBytes.get(feed) === 0) feedBytes.delete(feed);
  }
  function remove(entry) {
    unqueue(entry);
    entries.delete(entry.key);
    storage.keys--;
    account(entry.feed, -entry.bytes);
  }
  function canAllocate(feed, bytes) {
    return (
      storage.bytes + bytes <= policy.processBytes &&
      (storage.feedBytes.get(feed) || 0) + bytes <= policy.feedBytes
    );
  }
  function resize(entry, capacity) {
    const next = new Uint8Array(capacity * FIX_BYTES),
      before = new Uint8Array(entry.fixes.buffer);
    for (let i = 0; i < entry.count; i++)
      next.set(
        before.subarray(offset(entry, i), offset(entry, i) + FIX_BYTES),
        i * FIX_BYTES,
      );
    const delta = next.byteLength - entry.fixes.byteLength;
    entry.fixes = new DataView(next.buffer);
    entry.head = 0;
    entry.capacity = capacity;
    entry.bytes += delta;
    account(entry.feed, delta);
  }
  function prune(now) {
    if (now === prunedAt) return;
    prunedAt = now;
    for (const [feed, until] of storage.lossUntil)
      if (until <= now) storage.lossUntil.delete(feed);
    for (const entry of entries.values()) {
      while (entry.count && firstTime(entry) < now - policy.retentionMs)
        drop(entry);
      if (!entry.count || now - entry.observedAt >= policy.retentionMs)
        remove(entry);
      else if (
        entry.capacity > CHUNK &&
        entry.count <= entry.capacity / 2 &&
        canAllocate(
          entry.feed,
          Math.max(CHUNK, Math.ceil(entry.count / CHUNK) * CHUNK) * FIX_BYTES,
        )
      )
        resize(entry, Math.max(CHUNK, Math.ceil(entry.count / CHUNK) * CHUNK));
    }
  }
  function oldest(feed = null, except = null) {
    return (feed ? feedEviction.get(feed) : eviction)?.oldest(except) || null;
  }
  function evictOne(feed = null, except = null) {
    const entry = oldest(feed, except);
    if (!entry) return false;
    recordLoss(entry.feed, entry.observedAt + policy.retentionMs);
    if (entry.count > CHUNK) {
      const target = Math.max(CHUNK, entry.capacity - CHUNK);
      if (!canAllocate(entry.feed, target * FIX_BYTES)) {
        remove(entry);
        return true;
      }
      while (entry.count > target) drop(entry);
      resize(entry, target);
    } else remove(entry);
    return true;
  }
  const api = {
    prune,
    evictOne,
    oldest,
    get allocatedBytes() {
      return allocatedBytes;
    },
  };
  storage.stores.add(api);
  function reserve(feed, bytes, except = null, newKey = false) {
    while ((storage.feedBytes.get(feed) || 0) + bytes > policy.feedBytes) {
      let owner = null,
        candidate = null;
      for (const store of storage.stores) {
        const entry = store.oldest(feed, except);
        if (entry && (!candidate || entry.observedAt < candidate.observedAt)) {
          owner = store;
          candidate = entry;
        }
      }
      if (!owner || !owner.evictOne(feed, except)) return false;
    }
    while (
      storage.bytes + bytes > policy.processBytes ||
      (newKey && storage.keys >= policy.keys)
    ) {
      let owner = null,
        candidate = null;
      for (const store of storage.stores) {
        const entry = store.oldest(null, except);
        if (entry && (!candidate || entry.observedAt < candidate.observedAt)) {
          owner = store;
          candidate = entry;
        }
      }
      if (!owner || !owner.evictOne(null, except)) return false;
    }
    return true;
  }
  function ingest(feed, records, now) {
    if (
      closed ||
      feed.historyRetention !== true ||
      !validTransitIdentifier(feed.id)
    )
      return;
    prune(now);
    // Pin all existing snapshot members before admitting any new key. A later
    // row must not lose its retained path to an earlier row in the same feed.
    protectedEntries = new Set();
    for (const record of records) {
      const entry = entries.get(`${feed.id}\u0000${record.id}`);
      if (entry) {
        protectedEntries.add(entry);
        unqueue(entry);
      }
    }
    try {
      for (const record of records) {
        if (
          !validTransitIdentifier(record.id) ||
          !Number.isFinite(record.lat) ||
          Math.abs(record.lat) > 90 ||
          !Number.isFinite(record.lon) ||
          Math.abs(record.lon) > 180
        )
          continue;
        const t = record.timestamp * 1000;
        if (
          !Number.isFinite(t) ||
          t < now - policy.retentionMs ||
          t > now + 10000
        )
          continue;
        const mode =
          feed.routeMode?.(record.routeId) || feed.defaultMode || 'unknown';
        const context = contextOf(record, mode);
        if (!context) continue;
        const key = `${feed.id}\u0000${record.id}`;
        let entry = entries.get(key);
        if (entry) {
          const last = read(entry, entry.count - 1);
          if (t <= last[0]) continue;
        } else {
          const encodedKey = encoder.encode(key),
            bytes =
              META_BYTES +
              EPOCH_BYTES +
              encodedKey.byteLength +
              CHUNK * FIX_BYTES;
          if (!reserve(feed.id, bytes, null, true)) {
            recordLoss(feed.id, now + policy.retentionMs);
            continue;
          }
          entry = {
            key,
            encodedKey,
            feed: feed.id,
            vehicle: record.id,
            metadata: new DataView(new ArrayBuffer(META_BYTES)),
            contexts: [new DataView(new ArrayBuffer(EPOCH_BYTES))],
            fixes: new DataView(new ArrayBuffer(CHUNK * FIX_BYTES)),
            capacity: CHUNK,
            head: 0,
            count: 0,
            epoch: 0,
            observedAt: now,
            truncated: (storage.lossUntil.get(feed.id) || 0) > now,
            bytes,
          };
          entries.set(key, entry);
          protectedEntries.add(entry);
          storage.keys++;
          account(feed.id, bytes);
        }
        let flags =
          record.timestampSource === 'vehicle'
            ? 1
            : record.timestampSource === 'header'
              ? 2
              : 4;
        const previous = entry.epoch
          ? contextAt(entry, (entry.epoch - 1) % EPOCH_SLOTS)
          : null;
        if (
          !previous ||
          previous.trip !== context.trip ||
          previous.route !== context.route ||
          previous.mode !== context.mode
        ) {
          const slot = entry.epoch % EPOCH_SLOTS;
          if (!entry.contexts[slot]) {
            if (!reserve(feed.id, EPOCH_BYTES, entry)) {
              entry.truncated = true;
              recordLoss(feed.id, now + policy.retentionMs);
              continue;
            }
            entry.contexts[slot] = new DataView(new ArrayBuffer(EPOCH_BYTES));
            entry.bytes += EPOCH_BYTES;
            account(feed.id, EPOCH_BYTES);
          }
          entry.epoch++;
          // Expiring epoch metadata also expires its positions; never label a
          // retained point with a reused context slot.
          const expiredEpoch = entry.epoch - EPOCH_SLOTS;
          while (entry.count && read(entry, 0)[4] <= expiredEpoch) drop(entry);
          storeContext(entry, context, entry.epoch);
          flags |= 16;
        }
        if (entry.count === entry.capacity) {
          const next = Math.min(policy.fixes, entry.capacity + CHUNK);
          if (
            next > entry.capacity &&
            reserve(feed.id, next * FIX_BYTES, entry)
          )
            resize(entry, next);
          else drop(entry);
        }
        const at = offset(entry, entry.count++),
          v = entry.fixes;
        v.setFloat64(at, t);
        v.setFloat64(at + 8, record.lat);
        v.setFloat64(at + 16, record.lon);
        v.setUint32(at + 24, flags);
        v.setUint32(at + 28, entry.epoch);
        entry.observedAt = now;
      }
    } finally {
      for (const entry of protectedEntries) queue(entry);
      protectedEntries = null;
    }
  }
  function get(feedId, vehicleId, now) {
    prune(now);
    const entry = entries.get(`${feedId}\u0000${vehicleId}`);
    const fixes = [],
      epochs = [];
    if (entry) {
      const used = new Set();
      for (let i = 0; i < entry.count; i++) {
        const fix = read(entry, i);
        fixes.push(fix);
        used.add(fix[4]);
      }
      for (const id of used)
        epochs.push(contextAt(entry, (id - 1) % EPOCH_SLOTS));
    }
    const payload = {
      version: 1,
      feedId,
      vehicleId,
      oldestT: fixes[0]?.[0] ?? null,
      newestT: fixes.at(-1)?.[0] ?? null,
      truncated:
        entry?.truncated ||
        (!entry && (storage.lossUntil.get(feedId) || 0) > now),
      epochs,
      fixes,
    };
    // The normal bound is smaller; keep a byte guard for worst-case Unicode.
    while (
      encoder.encode(JSON.stringify(payload)).length > policy.responseBytes &&
      payload.fixes.length
    ) {
      payload.fixes.shift();
      payload.oldestT = payload.fixes[0]?.[0] ?? null;
      payload.truncated = true;
      const used = new Set(payload.fixes.map((f) => f[4]));
      payload.epochs = payload.epochs.filter((e) => used.has(e.id));
    }
    return payload;
  }
  function clear() {
    if (timer) clearInterval(timer);
    timer = null;
    closed = true;
    for (const entry of entries.values()) remove(entry);
    storage.stores.delete(api);
  }
  if (sweep) {
    timer = setInterval(() => prune(Date.now()), 60000);
    timer.unref?.();
  }
  return {
    ingest,
    get,
    prune,
    clear,
    diagnostics: () => ({
      allocatedBytes,
      processBytes: storage.bytes,
      keys: entries.size,
      feedBytes: Object.fromEntries(feedBytes),
      sweepActive: timer !== null,
    }),
  };
}