316 lines
11 KiB
TypeScript
316 lines
11 KiB
TypeScript
import { SelfUpdatingETARepository } from "./SelfUpdatingETARepository";
|
|
import { BaseRedisETARepository } from "./BaseRedisETARepository";
|
|
import { RedisClientType } from "redis";
|
|
import { ShuttleGetterRepository, ShuttleRepositoryEvent, ShuttleStopArrival, ShuttleTravelTimeDataIdentifier, ShuttleTravelTimeDateFilterArguments, ShuttleWillArriveAtStopPayload, ShuttleWillLeaveStopPayload } from "../ShuttleGetterRepository";
|
|
import { IEta, IOrderedStop, IShuttle } from "../../../entities/ShuttleRepositoryEntities";
|
|
import { ETARepositoryEvent } from "./ETAGetterRepository";
|
|
import createRedisClientForRepository from "../../../helpers/createRedisClientForRepository";
|
|
|
|
export class RedisSelfUpdatingETARepository extends BaseRedisETARepository implements SelfUpdatingETARepository {
|
|
private isListening = false;
|
|
|
|
constructor(
|
|
readonly shuttleRepository: ShuttleGetterRepository,
|
|
redisClient: RedisClientType = createRedisClientForRepository(),
|
|
private referenceTime: Date | null = null,
|
|
) {
|
|
super(redisClient);
|
|
|
|
this.setReferenceTime = this.setReferenceTime.bind(this);
|
|
this.getAverageTravelTimeSeconds = this.getAverageTravelTimeSeconds.bind(this);
|
|
this.startListeningForUpdates = this.startListeningForUpdates.bind(this);
|
|
this.handleShuttleWillArriveAtStop = this.handleShuttleWillArriveAtStop.bind(this);
|
|
this.handleShuttleUpdate = this.handleShuttleUpdate.bind(this);
|
|
this.updateCascadingEta = this.updateCascadingEta.bind(this);
|
|
this.getAverageTravelTimeSecondsWithFallbacks = this.getAverageTravelTimeSecondsWithFallbacks.bind(this);
|
|
this.removeEtaIfExists = this.removeEtaIfExists.bind(this);
|
|
this.handleShuttleWillLeaveStop = this.handleShuttleWillLeaveStop.bind(this);
|
|
}
|
|
|
|
private createHistoricalEtaTimeSeriesKey = (routeId: string, fromStopId: string, toStopId: string) => {
|
|
return `shuttle:eta:historical:${routeId}:${fromStopId}:${toStopId}`;
|
|
}
|
|
|
|
setReferenceTime(referenceTime: Date) {
|
|
this.referenceTime = referenceTime;
|
|
}
|
|
|
|
public async getAverageTravelTimeSeconds(
|
|
{ routeId, fromStopId, toStopId }: ShuttleTravelTimeDataIdentifier,
|
|
{ from, to }: ShuttleTravelTimeDateFilterArguments
|
|
): Promise<number | undefined> {
|
|
const timeSeriesKey = this.createHistoricalEtaTimeSeriesKey(routeId, fromStopId, toStopId);
|
|
const fromTimestamp = from.getTime();
|
|
const toTimestamp = to.getTime();
|
|
const intervalMs = toTimestamp - fromTimestamp + 1;
|
|
|
|
try {
|
|
const aggregationResult = await this.redisClient.sendCommand([
|
|
'TS.RANGE',
|
|
timeSeriesKey,
|
|
fromTimestamp.toString(),
|
|
toTimestamp.toString(),
|
|
'AGGREGATION',
|
|
'AVG',
|
|
intervalMs.toString()
|
|
]) as [string, string][];
|
|
|
|
if (aggregationResult && aggregationResult.length > 0) {
|
|
const [, averageValue] = aggregationResult[0];
|
|
return parseFloat(averageValue);
|
|
}
|
|
|
|
return;
|
|
} catch (error) {
|
|
console.warn(`Failed to get average travel time for ${timeSeriesKey}: ${error instanceof Error ? error.message : String(error)}`);
|
|
return;
|
|
}
|
|
}
|
|
|
|
startListeningForUpdates(): void {
|
|
if (this.isListening) {
|
|
console.warn("Already listening to updates; did you call startListeningForUpdates twice?");
|
|
return;
|
|
}
|
|
this.shuttleRepository.addListener(ShuttleRepositoryEvent.SHUTTLE_UPDATED, this.handleShuttleUpdate);
|
|
this.shuttleRepository.addListener(ShuttleRepositoryEvent.SHUTTLE_WILL_ARRIVE_AT_STOP, this.handleShuttleWillArriveAtStop);
|
|
this.shuttleRepository.addListener(ShuttleRepositoryEvent.SHUTTLE_WILL_LEAVE_STOP, this.handleShuttleWillLeaveStop);
|
|
this.isListening = true;
|
|
}
|
|
|
|
stopListeningForUpdates(): void {
|
|
if (!this.isListening) {
|
|
return;
|
|
}
|
|
this.shuttleRepository.removeListener(ShuttleRepositoryEvent.SHUTTLE_UPDATED, this.handleShuttleUpdate);
|
|
this.shuttleRepository.removeListener(ShuttleRepositoryEvent.SHUTTLE_WILL_ARRIVE_AT_STOP, this.handleShuttleWillArriveAtStop);
|
|
this.shuttleRepository.removeListener(ShuttleRepositoryEvent.SHUTTLE_WILL_LEAVE_STOP, this.handleShuttleWillLeaveStop);
|
|
this.isListening = false;
|
|
}
|
|
|
|
private async getAverageTravelTimeSecondsWithFallbacks(
|
|
identifier: ShuttleTravelTimeDataIdentifier,
|
|
dateFilters: ShuttleTravelTimeDateFilterArguments[]
|
|
): Promise<number | undefined> {
|
|
for (const dateFilter of dateFilters) {
|
|
const result = await this.getAverageTravelTimeSeconds(identifier, dateFilter);
|
|
if (result !== undefined) {
|
|
return result;
|
|
}
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
private async handleShuttleUpdate(shuttle: IShuttle) {
|
|
const isAtStop = await this.shuttleRepository.checkIfShuttleIsAtStop(shuttle.id);
|
|
const lastStop = await this.shuttleRepository.getShuttleLastStopArrival(shuttle.id);
|
|
if (!lastStop) return;
|
|
|
|
if (isAtStop) {
|
|
// Update the ETA *to* the stop the shuttle is currently at,
|
|
// before starting from the current stop as normal.
|
|
// Account for cases where the shuttle arrived way earlier than
|
|
// expected based on the calculated ETA.
|
|
|
|
await this.addOrUpdateEta({
|
|
secondsRemaining: 1,
|
|
shuttleId: shuttle.id,
|
|
stopId: lastStop.stopId,
|
|
systemId: shuttle.systemId,
|
|
updatedTime: new Date(),
|
|
});
|
|
}
|
|
|
|
const lastOrderedStop = await this.shuttleRepository.getOrderedStopByRouteAndStopId(shuttle.routeId, lastStop.stopId);
|
|
|
|
await this.updateCascadingEta({
|
|
shuttle,
|
|
currentStop: lastOrderedStop,
|
|
originalStopArrival: lastStop,
|
|
});
|
|
}
|
|
|
|
private async updateCascadingEta({
|
|
shuttle,
|
|
currentStop,
|
|
originalStopArrival,
|
|
runningTravelTimeSeconds = 0
|
|
}: {
|
|
shuttle: IShuttle;
|
|
currentStop: IOrderedStop | null;
|
|
originalStopArrival: ShuttleStopArrival;
|
|
runningTravelTimeSeconds?: number;
|
|
}) {
|
|
if (!currentStop) return;
|
|
const nextStop = currentStop?.nextStop;
|
|
if (!nextStop) return;
|
|
// In case the system we have loops around
|
|
if (nextStop.stopId === originalStopArrival.stopId) return;
|
|
|
|
let referenceCurrentTime = new Date();
|
|
if (this.referenceTime != null) {
|
|
referenceCurrentTime = this.referenceTime;
|
|
}
|
|
|
|
const oneWeekAgo = new Date(referenceCurrentTime.getTime() - (60 * 60 * 24 * 7 * 1000));
|
|
const oneDayAgo = new Date(referenceCurrentTime.getTime() - (60 * 60 * 24 * 1000));
|
|
const twoHoursAgo = new Date(referenceCurrentTime.getTime() - (120 * 60 * 1000));
|
|
|
|
const twoHoursInMs = 120 * 60 * 1000;
|
|
const travelTimeSeconds = await this.getAverageTravelTimeSecondsWithFallbacks({
|
|
routeId: shuttle.routeId,
|
|
fromStopId: currentStop.stopId,
|
|
toStopId: nextStop.stopId,
|
|
}, [
|
|
{
|
|
from: twoHoursAgo,
|
|
to: new Date(),
|
|
},
|
|
{
|
|
from: new Date(oneDayAgo.getTime() - (twoHoursInMs)),
|
|
to: oneDayAgo,
|
|
},
|
|
{
|
|
from: new Date(oneWeekAgo.getTime() - (twoHoursInMs)),
|
|
to: oneWeekAgo,
|
|
},
|
|
]);
|
|
|
|
if (travelTimeSeconds == undefined) return;
|
|
|
|
const elapsedTimeMs = referenceCurrentTime.getTime() - originalStopArrival.timestamp.getTime();
|
|
const predictedTimeSeconds = travelTimeSeconds - (elapsedTimeMs / 1000) + runningTravelTimeSeconds;
|
|
|
|
await this.addOrUpdateEta({
|
|
secondsRemaining: predictedTimeSeconds,
|
|
shuttleId: shuttle.id,
|
|
stopId: nextStop.stopId,
|
|
systemId: nextStop.systemId,
|
|
updatedTime: new Date(),
|
|
});
|
|
|
|
const nextStopWithNextNextStop = await this.shuttleRepository.getOrderedStopByRouteAndStopId(shuttle.routeId, nextStop.stopId);
|
|
await this.updateCascadingEta(
|
|
{
|
|
shuttle,
|
|
currentStop: nextStopWithNextNextStop,
|
|
originalStopArrival,
|
|
runningTravelTimeSeconds: runningTravelTimeSeconds + travelTimeSeconds,
|
|
},
|
|
)
|
|
}
|
|
|
|
|
|
private async handleShuttleWillArriveAtStop({
|
|
lastStopArrival: lastArrival,
|
|
willArriveAt: currentArrival,
|
|
}: ShuttleWillArriveAtStopPayload) {
|
|
// only update time traveled if last arrival exists
|
|
if (lastArrival) {
|
|
// disallow cases where this gets triggered multiple times
|
|
if (lastArrival.stopId === currentArrival.stopId) return;
|
|
|
|
const shuttle = await this.shuttleRepository.getShuttleById(lastArrival.shuttleId);
|
|
if (!shuttle) return;
|
|
|
|
const routeId = shuttle.routeId;
|
|
const fromStopId = lastArrival.stopId;
|
|
const toStopId = currentArrival.stopId;
|
|
|
|
const travelTimeSeconds = (currentArrival.timestamp.getTime() - lastArrival.timestamp.getTime()) / 1000;
|
|
await this.addTravelTimeDataPoint({ routeId, fromStopId, toStopId, }, travelTimeSeconds, currentArrival.timestamp.getTime());
|
|
}
|
|
}
|
|
|
|
private async handleShuttleWillLeaveStop({
|
|
stopArrivalThatShuttleIsLeaving,
|
|
}: ShuttleWillLeaveStopPayload) {
|
|
await this.removeEtaIfExists(stopArrivalThatShuttleIsLeaving.shuttleId, stopArrivalThatShuttleIsLeaving.stopId);
|
|
}
|
|
|
|
|
|
public async addTravelTimeDataPoint(
|
|
{ routeId, fromStopId, toStopId }: ShuttleTravelTimeDataIdentifier,
|
|
travelTimeSeconds: number,
|
|
timestamp = Date.now(),
|
|
): Promise<void> {
|
|
const historicalEtaTimeSeriesKey = this.createHistoricalEtaTimeSeriesKey(routeId, fromStopId, toStopId);
|
|
|
|
try {
|
|
await this.redisClient.sendCommand([
|
|
'TS.ADD',
|
|
historicalEtaTimeSeriesKey,
|
|
timestamp.toString(),
|
|
travelTimeSeconds.toString(),
|
|
'LABELS',
|
|
'routeId',
|
|
routeId,
|
|
'fromStopId',
|
|
fromStopId,
|
|
'toStopId',
|
|
toStopId
|
|
]);
|
|
} catch (error) {
|
|
await this.createHistoricalEtaTimeSeriesAndAddDataPoint(
|
|
historicalEtaTimeSeriesKey,
|
|
timestamp,
|
|
travelTimeSeconds,
|
|
routeId,
|
|
fromStopId,
|
|
toStopId
|
|
);
|
|
}
|
|
}
|
|
|
|
|
|
private async createHistoricalEtaTimeSeriesAndAddDataPoint(
|
|
timeSeriesKey: string,
|
|
timestamp: number,
|
|
travelTimeSeconds: number,
|
|
routeId: string,
|
|
fromStopId: string,
|
|
toStopId: string,
|
|
): Promise<void> {
|
|
try {
|
|
await this.redisClient.sendCommand([
|
|
'TS.CREATE',
|
|
timeSeriesKey,
|
|
'RETENTION',
|
|
'2678400000', // one month in milliseconds
|
|
'LABELS',
|
|
'routeId',
|
|
routeId,
|
|
'fromStopId',
|
|
fromStopId,
|
|
'toStopId',
|
|
toStopId
|
|
]);
|
|
await this.redisClient.sendCommand([
|
|
'TS.ADD',
|
|
timeSeriesKey,
|
|
timestamp.toString(),
|
|
travelTimeSeconds.toString()
|
|
]);
|
|
} catch (createError) {
|
|
await this.redisClient.sendCommand([
|
|
'TS.ADD',
|
|
timeSeriesKey,
|
|
timestamp.toString(),
|
|
travelTimeSeconds.toString()
|
|
]);
|
|
}
|
|
}
|
|
|
|
private async removeEtaIfExists(shuttleId: string, stopId: string): Promise<IEta | null> {
|
|
const existingEta = await this.getEtaForShuttleAndStopId(shuttleId, stopId);
|
|
if (existingEta === null) {
|
|
return null;
|
|
}
|
|
|
|
const key = this.createEtaKey(shuttleId, stopId);
|
|
await this.redisClient.del(key);
|
|
this.emit(ETARepositoryEvent.ETA_REMOVED, existingEta);
|
|
return existingEta;
|
|
}
|
|
}
|