From 9d4af1bb9f235fb7c8ea4c38cf21e30eff69a515 Mon Sep 17 00:00:00 2001 From: Gary Tokman Date: Sun, 24 May 2026 18:43:34 -0400 Subject: [PATCH] Add serverless Turso runtime --- README.md | 21 ++ package.json | 3 +- serverless.test.ts | 144 ++++++++++++ serverless.ts | 479 ++++++++++++++++++++++++++++++++++++++ src/libsql-static-gtfs.ts | 377 ++++++++++++++++++++++++++++++ 5 files changed, 1023 insertions(+), 1 deletion(-) create mode 100644 serverless.test.ts create mode 100644 serverless.ts create mode 100644 src/libsql-static-gtfs.ts diff --git a/README.md b/README.md index 14346a3..91e2394 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,27 @@ await mta.ready(); Most Turso databases require an auth token. You can omit `databaseAuthToken` only for databases configured to allow anonymous reads. +## Serverless Turso Runtime + +The default `mta-js` entrypoint uses `bun:sqlite` for local GTFS reads. In +Next.js/Vercel serverless routes, import `mta-js/serverless` to read and write +static GTFS data directly through Turso/libSQL without loading `bun:sqlite`. + +```ts +import { MTA } from "mta-js/serverless"; + +const mta = new MTA({ + databaseUrl: process.env.TURSO_DATABASE_URL!, + databaseAuthToken: process.env.TURSO_AUTH_TOKEN, + busTimeKey: process.env.MTA_BUS_KEY, +}); + +const arrivals = await mta.subway.arrivals({ + stopId: "A27", + route: "A", +}); +``` + The name is intentionally broader than a filesystem path so the public API can grow into hosted database adapters later without changing constructor shape. You can inspect whether each transit mode has static GTFS ready before serving traffic: diff --git a/package.json b/package.json index 7cbc4ae..63e910b 100644 --- a/package.json +++ b/package.json @@ -16,7 +16,8 @@ "module": "index.ts", "type": "module", "exports": { - ".": "./index.ts" + ".": "./index.ts", + "./serverless": "./serverless.ts" }, "types": "./index.ts", "bin": { diff --git a/serverless.test.ts b/serverless.test.ts new file mode 100644 index 0000000..8083ed2 --- /dev/null +++ b/serverless.test.ts @@ -0,0 +1,144 @@ +import { afterEach, expect, test } from "bun:test"; +import { MTA, StaticDataMissingError, UnknownStopError, encodeFeedMessage } from "./serverless"; + +const staticData = { + stops: [ + { + stop_id: "A27", + stop_name: "Jay St-MetroTech", + stop_lat: 40.692338, + stop_lon: -73.987342, + }, + { + stop_id: "A27N", + stop_name: "Jay St-MetroTech", + stop_lat: 40.692338, + stop_lon: -73.987342, + parent_station: "A27", + }, + ], + routes: [ + { + route_id: "A", + route_short_name: "A", + route_long_name: "8 Avenue Express", + route_type: 1, + route_color: "0039A6", + }, + ], + trips: [ + { + route_id: "A", + service_id: "weekday", + trip_id: "A-trip-1", + trip_headsign: "Inwood-207 St", + direction_id: 0, + }, + ], + stopTimes: [ + { + trip_id: "A-trip-1", + arrival_time: "12:00:00", + departure_time: "12:00:00", + stop_id: "A27N", + stop_sequence: 1, + }, + ], +}; + +const openClients: MTA[] = []; + +afterEach(() => { + while (openClients.length) openClients.pop()?.close(); +}); + +function tempDatabaseUrl(name: string) { + const tmp = process.env.TMPDIR ?? "/tmp"; + const id = `${Date.now()}-${Math.random().toString(36).slice(2)}`; + return `file:${tmp.replace(/\/$/, "")}/mta-js-serverless-${name}-${id}.sqlite`; +} + +test("serverless MTA reads and writes static GTFS through libSQL", async () => { + const mta = new MTA({ databaseUrl: tempDatabaseUrl("static") }); + openClients.push(mta); + + await mta.database.importStaticData({ + mode: "subway", + seed: staticData, + strategy: "schedule", + sourceUrl: "test://static-data", + }); + + expect((await mta.database.status()).subway).toMatchObject({ + mode: "subway", + ready: true, + sourceUrl: "test://static-data", + stopCount: 2, + routeCount: 1, + tripCount: 1, + stopTimeCount: 1, + }); + + await expect( + mta.stops.near({ + lat: 40.6923, + lon: -73.9873, + modes: ["subway"], + radiusMeters: 100, + }), + ).resolves.toContainEqual(expect.objectContaining({ id: "A27" })); +}); + +test("serverless subway arrivals join realtime feed to Turso static data", async () => { + const feed = encodeFeedMessage({ + header: { gtfsRealtimeVersion: "2.0", timestamp: 1_700_000_000 }, + entity: [ + { + id: "arrival-1", + tripUpdate: { + trip: { tripId: "A-trip-1", routeId: "A" }, + stopTimeUpdate: [{ stopId: "A27N", arrival: { time: 1_700_000_300 } }], + }, + }, + ], + }); + const mta = new MTA({ + databaseUrl: tempDatabaseUrl("arrivals"), + now: () => new Date("2023-11-14T22:13:20.000Z"), + fetch: (async () => new Response(feed)) as unknown as typeof fetch, + endpoints: { + subwayFeeds: { A: "feed://ace" }, + }, + }); + openClients.push(mta); + await mta.database.importStaticData({ mode: "subway", seed: staticData, strategy: "schedule" }); + + await expect(mta.subway.arrivals({ stopId: "A27", route: "A" })).resolves.toEqual([ + expect.objectContaining({ + mode: "subway", + direction: "north", + headsign: "Inwood-207 St", + arrivalTime: "2023-11-14T22:18:20.000Z", + minutes: 5, + tripId: "A-trip-1", + }), + ]); +}); + +test("serverless typed errors match the default entrypoint", async () => { + const mta = new MTA({ + databaseUrl: tempDatabaseUrl("errors"), + fetch: (async () => new Response(encodeFeedMessage({ header: { gtfsRealtimeVersion: "2.0" }, entity: [] }))) as unknown as typeof fetch, + endpoints: { + subwayFeeds: { A: "feed://ace" }, + }, + }); + openClients.push(mta); + + await expect(mta.stops.near({ lat: 40.6923, lon: -73.9873, modes: ["subway"] })).rejects.toThrow( + StaticDataMissingError, + ); + + await mta.database.importStaticData({ mode: "subway", seed: staticData }); + await expect(mta.subway.arrivals({ stopId: "NOPE", route: "A" })).rejects.toThrow(UnknownStopError); +}); diff --git a/serverless.ts b/serverless.ts new file mode 100644 index 0000000..b769133 --- /dev/null +++ b/serverless.ts @@ -0,0 +1,479 @@ +import { defaultEndpoints, subwayRouteColors } from "./src/defaults"; +import { MissingBusTimeKeyError, StaticDataMissingError, UnknownRouteError, UnknownStopError } from "./src/errors"; +import { decodeFeedMessage, type GtfsRealtimeFeed, type TranslatedString } from "./src/gtfs-realtime"; +import { fetchArrayBuffer, fetchJson, urlWithParams } from "./src/http"; +import { LibsqlStaticStore } from "./src/libsql-static-gtfs"; +import type { + Alert, + AlertQuery, + Arrival, + BusArrivalQuery, + BusVehicleQuery, + DatabaseStatus, + Direction, + GtfsImportSummary, + MTAEndpoints, + Route, + StaticGtfsImportStrategy, + StaticGtfsSeed, + Stop, + StopsNearQuery, + TransitMode, + Vehicle, +} from "./src/types"; + +export interface ServerlessMTAOptions { + databaseUrl: string; + databaseAuthToken?: string; + busTimeKey?: string; + realtimeCacheTtlMs?: number; + fetch?: typeof fetch; + now?: () => Date; + endpoints?: Partial; +} + +export class MTA { + readonly static: LibsqlStaticStore; + readonly database: DatabaseClient; + readonly subway: SubwayClient; + readonly bus: BusClient; + readonly alerts: AlertsClient; + readonly stops: StopsClient; + + readonly fetch: typeof fetch; + readonly now: () => Date; + readonly busTimeKey?: string; + readonly endpoints: MTAEndpoints; + private readonly realtimeCache = new Map(); + private readonly realtimeCacheTtlMs: number; + + constructor(readonly options: ServerlessMTAOptions) { + this.fetch = options.fetch ?? fetch; + this.now = options.now ?? (() => new Date()); + this.busTimeKey = options.busTimeKey; + this.realtimeCacheTtlMs = options.realtimeCacheTtlMs ?? 15_000; + this.endpoints = { + ...defaultEndpoints, + ...options.endpoints, + subwayFeeds: { + ...defaultEndpoints.subwayFeeds, + ...options.endpoints?.subwayFeeds, + }, + }; + this.static = new LibsqlStaticStore({ + databaseUrl: options.databaseUrl, + databaseAuthToken: options.databaseAuthToken, + }); + + this.database = new DatabaseClient(this); + this.subway = new SubwayClient(this); + this.bus = new BusClient(this); + this.alerts = new AlertsClient(this); + this.stops = new StopsClient(this); + } + + async ready() { + return this; + } + + close() { + this.static.close(); + } + + async realtimeFeed(url: string) { + const now = this.now().getTime(); + const cached = this.realtimeCache.get(url); + if (cached && cached.expiresAt > now) return cached.feed; + + const feed = decodeFeedMessage(await fetchArrayBuffer(this.fetch, url)); + if (this.realtimeCacheTtlMs > 0) { + this.realtimeCache.set(url, { feed, expiresAt: now + this.realtimeCacheTtlMs }); + } + return feed; + } +} + +class DatabaseClient { + constructor(private readonly mta: MTA) {} + + push() { + return this.mta.static.pushSchema(); + } + + hasStaticData(mode: TransitMode) { + return this.mta.static.hasStaticData(mode); + } + + status(): Promise { + return this.mta.static.status(); + } + + importStaticData(input: { + mode: TransitMode; + seed?: StaticGtfsSeed; + sourceUrl?: string; + strategy?: StaticGtfsImportStrategy; + }): Promise { + return this.mta.static.importStaticData({ + ...input, + fetch: this.mta.fetch, + }); + } + + async ensureStaticData(input: { + mode: TransitMode; + seed?: StaticGtfsSeed; + sourceUrl?: string; + strategy?: StaticGtfsImportStrategy; + }): Promise { + if (await this.mta.static.hasStaticData(input.mode)) { + return this.mta.static.importSummary(input.mode); + } + return this.importStaticData(input); + } +} + +class SubwayClient { + constructor(private readonly mta: MTA) {} + + async arrivals(query: { + stopId: string; + route?: string; + direction?: Direction | "uptown" | "downtown"; + limit?: number; + includeRaw?: boolean; + }): Promise { + const routeIds = query.route ? [normalizeRouteId(query.route)] : Object.keys(this.mta.endpoints.subwayFeeds); + const feeds = [...new Set(routeIds.map((route) => this.feedForRoute(route)))]; + const stopIds = await this.mta.static.getStopIdsForQuery(query.stopId); + if ((await this.mta.static.hasStaticData("subway")) && !(await this.mta.static.getStopOrParent(query.stopId))) { + throw new UnknownStopError(query.stopId); + } + const arrivals: Arrival[] = []; + + for (const feedUrl of feeds) { + const feed = await this.mta.realtimeFeed(feedUrl); + arrivals.push(...(await this.arrivalsFromFeed(feed, stopIds, query))); + } + + return arrivals + .sort((a, b) => Date.parse(a.arrivalTime) - Date.parse(b.arrivalTime)) + .slice(0, query.limit ?? 20); + } + + private feedForRoute(route: string) { + const feed = this.mta.endpoints.subwayFeeds[route]; + if (!feed) throw new UnknownRouteError(route); + return feed; + } + + private async arrivalsFromFeed( + feed: GtfsRealtimeFeed, + stopIds: Set, + query: { stopId: string; route?: string; direction?: Direction | "uptown" | "downtown"; includeRaw?: boolean }, + ) { + const arrivals: Arrival[] = []; + const wantedDirection = normalizeDirection(query.direction); + const now = this.mta.now().getTime(); + + for (const entity of feed.entity) { + const tripUpdate = entity.tripUpdate; + if (!tripUpdate) continue; + + const trip = tripUpdate.trip; + const routeId = normalizeRouteId(trip?.routeId ?? query.route ?? ""); + if (query.route && routeId !== normalizeRouteId(query.route)) continue; + + const staticTrip = trip?.tripId ? await this.mta.static.getTrip(trip.tripId) : undefined; + const route = routeWithFallback(await this.mta.static.getRoute(routeId), routeId); + + for (const update of tripUpdate.stopTimeUpdate ?? []) { + const stopId = update.stopId; + if (!stopId || !stopIds.has(stopId)) continue; + if (update.scheduleRelationship === "SKIPPED" || update.scheduleRelationship === "NO_DATA") continue; + + const direction = directionFromStopId(stopId); + if (wantedDirection && direction !== wantedDirection) continue; + + const event = update.arrival ?? update.departure; + if (!event?.time) continue; + + const stop = (await this.mta.static.getStopOrParent(stopId)) ?? fallbackStop(query.stopId); + arrivals.push({ + mode: "subway", + route, + stop, + direction, + headsign: staticTrip?.headsign ?? undefined, + arrivalTime: new Date(event.time * 1000).toISOString(), + departureTime: update.departure?.time ? new Date(update.departure.time * 1000).toISOString() : undefined, + minutes: Math.max(0, Math.round((event.time * 1000 - now) / 60_000)), + tripId: trip?.tripId, + realtime: true, + source: "mta-gtfs-rt", + raw: query.includeRaw ? entity : undefined, + }); + } + } + + return arrivals; + } +} + +class BusClient { + constructor(private readonly mta: MTA) {} + + async arrivals(query: BusArrivalQuery): Promise { + const key = this.requireKey(); + const body = await fetchJson( + this.mta.fetch, + urlWithParams(this.mta.endpoints.busStopMonitoring, { + key, + version: "2", + OperatorRef: "MTA", + MonitoringRef: query.stopId, + LineRef: query.route ? busLineRef(query.route) : undefined, + }), + ); + const journeys = monitoredStopVisits(body); + const now = this.mta.now().getTime(); + + const arrivals = await Promise.all( + journeys.map(async (journey): Promise => { + const mvj = journey.MonitoredVehicleJourney; + if (!mvj) return undefined; + const routeId = routeFromLineRef(mvj.LineRef ?? query.route ?? ""); + const call = mvj.MonitoredCall ?? {}; + const expected = call.ExpectedArrivalTime ?? call.AimedArrivalTime; + if (!expected) return undefined; + const stop = (await this.mta.static.getStop(String(call.StopPointRef ?? query.stopId))) ?? fallbackStop(query.stopId); + return { + mode: "bus", + route: routeWithFallback(await this.mta.static.getRoute(routeId), routeId), + stop, + direction: "unknown", + headsign: stringOrUndefined(mvj.DestinationName), + arrivalTime: new Date(expected).toISOString(), + minutes: Math.max(0, Math.round((Date.parse(expected) - now) / 60_000)), + tripId: stringOrUndefined(mvj.FramedVehicleJourneyRef?.DatedVehicleJourneyRef), + realtime: true, + source: "mta-bustime", + raw: query.includeRaw ? journey : undefined, + }; + }), + ); + + return arrivals + .filter((arrival): arrival is Arrival => Boolean(arrival)) + .sort((a, b) => Date.parse(a.arrivalTime) - Date.parse(b.arrivalTime)) + .slice(0, query.limit ?? 20); + } + + async vehicles(query: BusVehicleQuery = {}): Promise { + const key = this.requireKey(); + const body = await fetchJson( + this.mta.fetch, + urlWithParams(this.mta.endpoints.busVehicleMonitoring, { + key, + version: "2", + OperatorRef: "MTA", + LineRef: query.route ? busLineRef(query.route) : undefined, + VehicleRef: query.vehicleId, + }), + ); + + const vehicles = await Promise.all( + monitoredVehicleJourneys(body).map(async (mvj): Promise => { + const routeId = routeFromLineRef(mvj.LineRef ?? query.route ?? ""); + const location = mvj.VehicleLocation ?? {}; + const stopId = stringOrUndefined(mvj.MonitoredCall?.StopPointRef); + return { + mode: "bus", + route: routeWithFallback(await this.mta.static.getRoute(routeId), routeId), + vehicleId: stringOrUndefined(mvj.VehicleRef), + tripId: stringOrUndefined(mvj.FramedVehicleJourneyRef?.DatedVehicleJourneyRef), + stop: stopId ? (await this.mta.static.getStop(stopId)) ?? fallbackStop(stopId) : undefined, + lat: numberOrUndefined(location.Latitude), + lon: numberOrUndefined(location.Longitude), + bearing: numberOrUndefined(mvj.Bearing), + destinationName: stringOrUndefined(mvj.DestinationName), + recordedAt: mvj.RecordedAtTime ? new Date(mvj.RecordedAtTime).toISOString() : undefined, + source: "mta-bustime", + raw: query.includeRaw ? mvj : undefined, + }; + }), + ); + + return vehicles.slice(0, query.limit ?? 50); + } + + private requireKey() { + if (!this.mta.busTimeKey) throw new MissingBusTimeKeyError(); + return this.mta.busTimeKey; + } +} + +class AlertsClient { + constructor(private readonly mta: MTA) {} + + async current(query: AlertQuery = {}): Promise { + const feed = await this.mta.realtimeFeed(this.mta.endpoints.alerts); + const alerts: Alert[] = []; + + for (const entity of feed.entity) { + if (!entity.alert) continue; + const informed = entity.alert.informedEntity ?? []; + const routeIds = [...new Set(informed.map((item) => item.routeId).filter((id): id is string => Boolean(id)))]; + const stopIds = [...new Set(informed.map((item) => item.stopId).filter((id): id is string => Boolean(id)))]; + const routes = await Promise.all(routeIds.map(async (id) => routeWithFallback(await this.mta.static.getRoute(id), id))); + const stops = await Promise.all(stopIds.map(async (id) => (await this.mta.static.getStopOrParent(id)) ?? fallbackStop(id))); + + if (query.route && !routeIds.some((id) => normalizeRouteId(id) === normalizeRouteId(query.route!))) continue; + if (query.stopId && !stopIds.includes(query.stopId)) continue; + if (query.mode && !alertMatchesMode(query.mode, routes, stops, informed)) continue; + + alerts.push({ + id: entity.id, + mode: inferAlertMode(routes, stops, informed), + routes, + stops, + header: translatedText(entity.alert.headerText), + description: translatedText(entity.alert.descriptionText), + url: translatedText(entity.alert.url), + effect: entity.alert.effect, + activePeriods: (entity.alert.activePeriod ?? []).map((period) => ({ + start: period.start ? new Date(period.start * 1000).toISOString() : undefined, + end: period.end ? new Date(period.end * 1000).toISOString() : undefined, + })), + source: "mta-gtfs-rt", + raw: query.includeRaw ? entity : undefined, + }); + } + + return alerts; + } +} + +class StopsClient { + constructor(private readonly mta: MTA) {} + + async near(query: StopsNearQuery): Promise { + for (const mode of query.modes ?? []) { + if (!(await this.mta.static.hasStaticData(mode))) throw new StaticDataMissingError(mode); + } + return this.mta.static.stopsNear(query); + } +} + +function normalizeRouteId(route: string) { + return route.toUpperCase().trim(); +} + +function normalizeDirection(direction: Direction | "uptown" | "downtown" | undefined): Direction | undefined { + if (!direction) return undefined; + if (direction === "uptown") return "north"; + if (direction === "downtown") return "south"; + return direction; +} + +function directionFromStopId(stopId: string): Direction { + if (stopId.endsWith("N")) return "north"; + if (stopId.endsWith("S")) return "south"; + return "unknown"; +} + +function routeWithFallback(route: Route | undefined, routeId: string): Route { + return ( + route ?? { + id: routeId, + shortName: routeId, + color: subwayRouteColors[routeId] ? `#${subwayRouteColors[routeId]}` : undefined, + } + ); +} + +function fallbackStop(stopId: string): Stop { + return { id: stopId, name: stopId }; +} + +function busLineRef(route: string) { + const normalized = normalizeBusRouteId(route); + return normalized.includes("_") ? normalized : `MTA NYCT_${normalized}`; +} + +function routeFromLineRef(lineRef: string) { + return String(lineRef).split("_").at(-1)?.toUpperCase() ?? String(lineRef).toUpperCase(); +} + +function normalizeBusRouteId(route: string) { + const normalized = route.toUpperCase().trim(); + const aliases: Record = { + M14A: "M14A-SBS", + M14D: "M14D-SBS", + M15: "M15-SBS", + M23: "M23-SBS", + M34: "M34-SBS", + M34A: "M34A-SBS", + M60: "M60-SBS", + M79: "M79-SBS", + M86: "M86-SBS", + }; + return aliases[normalized] ?? normalized; +} + +function stringOrUndefined(value: unknown) { + return typeof value === "string" && value.length > 0 ? value : undefined; +} + +function numberOrUndefined(value: unknown) { + if (value === undefined || value === null || value === "") return undefined; + const number = Number(value); + return Number.isFinite(number) ? number : undefined; +} + +function translatedText(value: TranslatedString | undefined) { + return value?.translation?.find((translation) => !translation.language || translation.language === "en")?.text ?? + value?.translation?.[0]?.text; +} + +function monitoredStopVisits(body: unknown): any[] { + return ( + (body as any)?.Siri?.ServiceDelivery?.StopMonitoringDelivery?.[0]?.MonitoredStopVisit ?? + (body as any)?.Siri?.ServiceDelivery?.StopMonitoringDelivery?.MonitoredStopVisit ?? + [] + ); +} + +function monitoredVehicleJourneys(body: unknown): any[] { + const visits = + (body as any)?.Siri?.ServiceDelivery?.VehicleMonitoringDelivery?.[0]?.VehicleActivity ?? + (body as any)?.Siri?.ServiceDelivery?.VehicleMonitoringDelivery?.VehicleActivity ?? + []; + return visits.map((visit: any) => visit.MonitoredVehicleJourney).filter(Boolean); +} + +function inferAlertMode( + routes: Route[], + stops: Stop[], + informed: { routeType?: number }[], +): TransitMode | undefined { + if (stops.some((stop) => stop.mode)) return stops.find((stop) => stop.mode)?.mode; + if (informed.some((item) => item.routeType === 3)) return "bus"; + if (informed.some((item) => item.routeType === 1)) return "subway"; + if (routes.some((route) => route.type === 3)) return "bus"; + if (routes.some((route) => route.type === 1 || route.id.length <= 2)) return "subway"; + return undefined; +} + +function alertMatchesMode( + mode: TransitMode, + routes: Route[], + stops: Stop[], + informed: { routeType?: number }[], +) { + return inferAlertMode(routes, stops, informed) === mode; +} + +export { decodeFeedMessage, encodeFeedMessage } from "./src/gtfs-realtime"; +export * from "./src/errors"; +export type * from "./src/types"; diff --git a/src/libsql-static-gtfs.ts b/src/libsql-static-gtfs.ts new file mode 100644 index 0000000..4255ce5 --- /dev/null +++ b/src/libsql-static-gtfs.ts @@ -0,0 +1,377 @@ +import { createClient, type Client, type InArgs } from "@libsql/client"; +import { parse } from "csv-parse/sync"; +import { unzipSync } from "fflate"; +import { gtfsSchemaStatements } from "./schema"; +import type { + DatabaseStatus, + GtfsImportSummary, + Route, + StaticDataStatus, + StaticGtfsSeed, + StaticGtfsImportStrategy, + Stop, + StopsNearQuery, + TransitMode, +} from "./types"; + +type LibsqlStaticStoreOptions = { + databaseUrl: string; + databaseAuthToken?: string; +}; + +export class LibsqlStaticStore { + readonly client: Client; + + constructor(options: LibsqlStaticStoreOptions) { + this.client = createClient({ + url: options.databaseUrl, + authToken: options.databaseAuthToken, + }); + } + + close() { + this.client.close(); + } + + async pushSchema() { + for (const sql of gtfsSchemaStatements) { + await this.client.execute(sql); + } + return { remote: true, statements: gtfsSchemaStatements.length }; + } + + async hasStaticData(mode: TransitMode) { + try { + const result = await this.client.execute({ + sql: "select count(*) as count from gtfs_imports where mode = ?", + args: [mode], + }); + return Number(result.rows[0]?.count ?? 0) > 0; + } catch (error) { + if (isMissingTableError(error)) return false; + throw error; + } + } + + async status(): Promise { + return { + subway: await this.statusForMode("subway"), + bus: await this.statusForMode("bus"), + lirr: await this.statusForMode("lirr"), + "metro-north": await this.statusForMode("metro-north"), + }; + } + + async importStaticData(input: { + mode: TransitMode; + seed?: StaticGtfsSeed; + sourceUrl?: string; + strategy?: StaticGtfsImportStrategy; + fetch?: typeof fetch; + }): Promise { + const parsedSeed = + input.seed ?? + (input.sourceUrl + ? parseGtfsZip(await fetchArrayBuffer(input.fetch ?? fetch, input.sourceUrl)) + : undefined); + if (!parsedSeed) throw new Error("importStaticData requires either seed or sourceUrl."); + + const seed = applyImportStrategy(parsedSeed, input.strategy ?? "core"); + + await this.pushSchema(); + await batchChunks( + this.client, + (seed.stops ?? []).map((stop) => ({ + sql: `insert or replace into stops + (id, name, lat, lon, parent_station, location_type, mode) + values (?, ?, ?, ?, ?, ?, ?)`, + args: [ + stop.stop_id, + stop.stop_name, + numberOrNull(stop.stop_lat), + numberOrNull(stop.stop_lon), + stop.parent_station ?? null, + numberOrNull(stop.location_type), + input.mode, + ], + })), + ); + await batchChunks( + this.client, + (seed.routes ?? []).map((route) => ({ + sql: `insert or replace into routes + (id, short_name, long_name, type, color, text_color) + values (?, ?, ?, ?, ?, ?)`, + args: [ + route.route_id, + route.route_short_name ?? route.route_id, + route.route_long_name ?? null, + numberOrNull(route.route_type), + normalizeColor(route.route_color), + normalizeColor(route.route_text_color), + ], + })), + ); + await batchChunks( + this.client, + (seed.trips ?? []).map((trip) => ({ + sql: `insert or replace into trips + (id, route_id, service_id, headsign, direction_id) + values (?, ?, ?, ?, ?)`, + args: [ + trip.trip_id, + trip.route_id, + trip.service_id ?? null, + trip.trip_headsign ?? null, + numberOrNull(trip.direction_id), + ], + })), + ); + await batchChunks( + this.client, + (seed.stopTimes ?? []).map((stopTime) => ({ + sql: `insert or replace into stop_times + (trip_id, arrival_time, departure_time, stop_id, stop_sequence) + values (?, ?, ?, ?, ?)`, + args: [ + stopTime.trip_id, + stopTime.arrival_time ?? null, + stopTime.departure_time ?? null, + stopTime.stop_id, + numberOrNull(stopTime.stop_sequence) ?? 0, + ], + })), + ); + await this.client.execute({ + sql: `insert or replace into gtfs_imports + (mode, imported_at, source_url, stop_count, route_count, trip_count, stop_time_count) + values (?, ?, ?, ?, ?, ?, ?)`, + args: [ + input.mode, + new Date().toISOString(), + input.sourceUrl ?? null, + seed.stops?.length ?? 0, + seed.routes?.length ?? 0, + seed.trips?.length ?? 0, + seed.stopTimes?.length ?? 0, + ], + }); + + return this.importSummary(input.mode); + } + + async importSummary(mode: TransitMode): Promise { + const result = await this.client + .execute({ + sql: "select * from gtfs_imports where mode = ?", + args: [mode], + }) + .catch((error) => { + if (isMissingTableError(error)) return undefined; + throw error; + }); + if (!result) return undefined; + const row = result.rows[0]; + if (!row) return undefined; + return { + mode, + importedAt: String(row.imported_at), + sourceUrl: row.source_url ? String(row.source_url) : undefined, + stopCount: Number(row.stop_count ?? 0), + routeCount: Number(row.route_count ?? 0), + tripCount: Number(row.trip_count ?? 0), + stopTimeCount: Number(row.stop_time_count ?? 0), + }; + } + + async getStop(id: string): Promise { + const result = await this.client.execute({ sql: "select * from stops where id = ?", args: [id] }); + return result.rows[0] ? stopFromRow(result.rows[0]) : undefined; + } + + async getStopOrParent(id: string): Promise { + const direct = await this.getStop(id); + if (direct?.parentStation) return (await this.getStop(direct.parentStation)) ?? direct; + if (direct) return direct; + return this.getStop(stripDirectionSuffix(id)); + } + + async getRoute(idOrShortName: string): Promise { + const normalized = idOrShortName.toUpperCase(); + const result = await this.client.execute({ + sql: "select * from routes where upper(id) = ? or upper(short_name) = ? limit 1", + args: [normalized, normalized], + }); + return result.rows[0] ? routeFromRow(result.rows[0]) : undefined; + } + + async getTrip(id: string) { + const result = await this.client.execute({ sql: "select * from trips where id = ?", args: [id] }); + const row = result.rows[0]; + if (!row) return undefined; + return { + id: String(row.id), + route_id: String(row.route_id), + service_id: row.service_id ? String(row.service_id) : undefined, + headsign: row.headsign ? String(row.headsign) : undefined, + direction_id: numberOrUndefined(row.direction_id), + }; + } + + async getStopIdsForQuery(stopId: string) { + const ids = new Set([stopId]); + const parent = stripDirectionSuffix(stopId); + ids.add(parent); + ids.add(`${parent}N`); + ids.add(`${parent}S`); + + const result = await this.client.execute({ + sql: "select id from stops where parent_station = ?", + args: [parent], + }); + for (const row of result.rows) ids.add(String(row.id)); + return ids; + } + + async stopsNear(query: StopsNearQuery): Promise { + const radiusMeters = query.radiusMeters ?? 500; + const limit = query.limit ?? 20; + const latSpan = radiusMeters / 111_320; + const lonSpan = radiusMeters / (111_320 * Math.cos((query.lat * Math.PI) / 180)); + const modes = query.modes?.length ? query.modes : undefined; + + const result = await this.client.execute({ + sql: `select * from stops + where lat between ? and ? + and lon between ? and ? + and lat is not null + and lon is not null`, + args: [query.lat - latSpan, query.lat + latSpan, query.lon - lonSpan, query.lon + lonSpan], + }); + + return result.rows + .map((row) => ({ + stop: stopFromRow(row), + distance: distanceMeters(query.lat, query.lon, Number(row.lat), Number(row.lon)), + })) + .filter((row) => row.distance <= radiusMeters) + .filter((row) => !modes || !row.stop.mode || modes.includes(row.stop.mode)) + .sort((a, b) => a.distance - b.distance) + .slice(0, limit) + .map((row) => row.stop); + } + + private async statusForMode(mode: TransitMode): Promise { + const summary = await this.importSummary(mode); + return { + mode, + ready: Boolean(summary), + importedAt: summary?.importedAt, + sourceUrl: summary?.sourceUrl, + stopCount: summary?.stopCount ?? 0, + routeCount: summary?.routeCount ?? 0, + tripCount: summary?.tripCount ?? 0, + stopTimeCount: summary?.stopTimeCount ?? 0, + }; + } +} + +async function fetchArrayBuffer(fetchImpl: typeof fetch, sourceUrl: string) { + const response = await fetchImpl(sourceUrl); + if (!response.ok) throw new Error(`Failed to fetch GTFS zip from ${sourceUrl}: ${response.status}`); + return response.arrayBuffer(); +} + +function parseGtfsZip(zipBytes: ArrayBuffer | Uint8Array): Required { + const files = unzipSync(new Uint8Array(zipBytes)); + const text = (name: string) => { + const bytes = files[name]; + if (!bytes) return []; + return parse(new TextDecoder().decode(bytes), { + columns: true, + bom: true, + skip_empty_lines: true, + }) as Record[]; + }; + + return { + stops: text("stops.txt") as unknown as Required["stops"], + routes: text("routes.txt") as unknown as Required["routes"], + trips: text("trips.txt") as unknown as Required["trips"], + stopTimes: text("stop_times.txt") as unknown as Required["stopTimes"], + }; +} + +function applyImportStrategy(seed: StaticGtfsSeed, strategy: StaticGtfsImportStrategy): StaticGtfsSeed { + if (strategy === "schedule") return seed; + return { + stops: seed.stops, + routes: seed.routes, + trips: [], + stopTimes: [], + }; +} + +async function batchChunks(client: Client, statements: { sql: string; args: InArgs }[], size = 500) { + for (let index = 0; index < statements.length; index += size) { + const chunk = statements.slice(index, index + size); + if (chunk.length) await client.batch(chunk, "write"); + } +} + +function stopFromRow(row: Record): Stop { + return { + id: String(row.id), + name: String(row.name), + lat: numberOrUndefined(row.lat), + lon: numberOrUndefined(row.lon), + parentStation: row.parent_station ? String(row.parent_station) : undefined, + mode: row.mode ? (String(row.mode) as TransitMode) : undefined, + }; +} + +function routeFromRow(row: Record): Route { + return { + id: String(row.id), + shortName: row.short_name ? String(row.short_name) : undefined, + longName: row.long_name ? String(row.long_name) : undefined, + type: numberOrUndefined(row.type), + color: row.color ? String(row.color) : undefined, + textColor: row.text_color ? String(row.text_color) : undefined, + }; +} + +function stripDirectionSuffix(stopId: string) { + return stopId.replace(/[NS]$/, ""); +} + +function numberOrUndefined(value: unknown) { + if (value === undefined || value === null || value === "") return undefined; + const number = Number(value); + return Number.isFinite(number) ? number : undefined; +} + +function numberOrNull(value: unknown) { + return numberOrUndefined(value) ?? null; +} + +function normalizeColor(value: unknown) { + if (typeof value !== "string" || !value) return null; + return value.startsWith("#") ? value : `#${value}`; +} + +function distanceMeters(lat1: number, lon1: number, lat2: number, lon2: number) { + const radius = 6_371_000; + const phi1 = (lat1 * Math.PI) / 180; + const phi2 = (lat2 * Math.PI) / 180; + const deltaPhi = ((lat2 - lat1) * Math.PI) / 180; + const deltaLambda = ((lon2 - lon1) * Math.PI) / 180; + const a = + Math.sin(deltaPhi / 2) ** 2 + + Math.cos(phi1) * Math.cos(phi2) * Math.sin(deltaLambda / 2) ** 2; + return radius * 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a)); +} + +function isMissingTableError(error: unknown) { + return error instanceof Error && error.message.includes("no such table"); +}