Move repositories into folders.

src/repositories/notifications/InMemoryNotificationRepository.ts

@@ -0,0 +1,144 @@
+import {
+  Listener,
+  NotificationEvent,
+  NotificationLookupArguments,
+  NotificationRepository,
+  ScheduledNotification
+} from "./NotificationRepository";
+import { TupleKey } from "../../types/TupleKey";
+
+type DeviceIdSecondsThresholdAssociation = { [key: string]: number };
+
+export class InMemoryNotificationRepository implements NotificationRepository {
+  /**
+   * An object of device ID arrays to deliver notifications to.
+   * The key should be a combination of the shuttle ID and
+   * stop ID, which can be generated using `TupleKey`.
+   * The value is a dictionary of the device ID to the stored seconds threshold.
+   * @private
+   */
+  private deviceIdsToDeliverTo: { [key: string]: DeviceIdSecondsThresholdAssociation } = {}
+
+  private listeners: Listener[] = [];
+
+  constructor() {
+    this.getAllNotificationsForShuttleAndStopId = this.getAllNotificationsForShuttleAndStopId.bind(this);
+    this.getSecondsThresholdForNotificationIfExists = this.getSecondsThresholdForNotificationIfExists.bind(this);
+    this.deleteNotificationIfExists = this.deleteNotificationIfExists.bind(this);
+    this.addOrUpdateNotification = this.addOrUpdateNotification.bind(this);
+    this.isNotificationScheduled = this.isNotificationScheduled.bind(this);
+    this.subscribeToNotificationChanges = this.subscribeToNotificationChanges.bind(this);
+    this.unsubscribeFromNotificationChanges = this.unsubscribeFromNotificationChanges.bind(this);
+  }
+
+  async getAllNotificationsForShuttleAndStopId(shuttleId: string, stopId: string) {
+    const tuple = new TupleKey(shuttleId, stopId);
+    if (this.deviceIdsToDeliverTo[tuple.toString()] === undefined) {
+      return [];
+    }
+
+    return Object.keys(this.deviceIdsToDeliverTo[tuple.toString()])
+      .map((deviceId) => {
+        return {
+          shuttleId,
+          stopId,
+          deviceId,
+          secondsThreshold: this.deviceIdsToDeliverTo[tuple.toString()][deviceId]
+        }
+      });
+  }
+
+  async getSecondsThresholdForNotificationIfExists({
+    shuttleId,
+    stopId,
+    deviceId
+  }: NotificationLookupArguments) {
+    const tuple = new TupleKey(shuttleId, stopId);
+    if (this.deviceIdsToDeliverTo[tuple.toString()] === undefined) {
+      return null;
+    }
+    return this.deviceIdsToDeliverTo[tuple.toString()][deviceId];
+  }
+
+  async isNotificationScheduled(lookupArguments: NotificationLookupArguments): Promise<boolean> {
+    const threshold = await this.getSecondsThresholdForNotificationIfExists(lookupArguments);
+    return threshold !== null;
+  }
+
+  async addOrUpdateNotification({
+    shuttleId,
+    stopId,
+    deviceId,
+    secondsThreshold
+  }: ScheduledNotification) {
+    const tuple = new TupleKey(shuttleId, stopId);
+    if (this.deviceIdsToDeliverTo[tuple.toString()] === undefined) {
+      this.deviceIdsToDeliverTo[tuple.toString()] = {};
+    }
+
+    this.deviceIdsToDeliverTo[tuple.toString()][deviceId] = secondsThreshold;
+    this.listeners.forEach((listener: Listener) => {
+      const event: NotificationEvent = {
+        event: 'addOrUpdate',
+        notification: {
+          shuttleId,
+          stopId,
+          deviceId,
+          secondsThreshold
+        },
+      }
+
+      listener(event);
+    })
+  }
+
+  async deleteNotificationIfExists({
+    deviceId,
+    shuttleId,
+    stopId
+  }: NotificationLookupArguments) {
+    const tupleKey = new TupleKey(shuttleId, stopId);
+    if (
+      this.deviceIdsToDeliverTo[tupleKey.toString()] === undefined
+      || !(deviceId in this.deviceIdsToDeliverTo[tupleKey.toString()])
+    ) {
+      return;
+    }
+
+    const secondsThreshold = this.deviceIdsToDeliverTo[tupleKey.toString()][deviceId];
+    delete this.deviceIdsToDeliverTo[tupleKey.toString()][deviceId];
+
+    if (Object.keys(this.deviceIdsToDeliverTo[tupleKey.toString()]).length === 0) {
+      // no more device IDs remaining for this key combination
+      delete this.deviceIdsToDeliverTo[tupleKey.toString()];
+    }
+
+    this.listeners.forEach((listener) => {
+      const event: NotificationEvent = {
+        event: 'delete',
+        notification: {
+          deviceId,
+          shuttleId,
+          stopId,
+          secondsThreshold
+        }
+      }
+
+      listener(event);
+    })
+  }
+
+  public subscribeToNotificationChanges(listener: Listener): void {
+    const index = this.listeners.findIndex((existingListener) => existingListener == listener);
+    if (index < 0) {
+      this.listeners.push(listener);
+    }
+  }
+
+  public unsubscribeFromNotificationChanges(listener: Listener): void {
+    const index = this.listeners.findIndex((existingListener) => existingListener == listener);
+    if (index >= 0) {
+      this.listeners.splice(index, 1);
+    }
+  }
+}

src/repositories/notifications/NotificationRepository.ts

@@ -0,0 +1,33 @@
+export interface NotificationLookupArguments {
+  deviceId: string;
+  shuttleId: string;
+  stopId: string;
+}
+
+export interface ScheduledNotification extends NotificationLookupArguments {
+  /**
+   * Value which specifies the ETA of the shuttle for when
+   * the notification should fire.
+   * For example, a secondsThreshold of 180 would mean that the notification
+   * fires when the ETA drops below 3 minutes.
+   */
+  secondsThreshold: number;
+}
+
+export type Listener = ((event: NotificationEvent) => any);
+
+export interface NotificationEvent {
+  notification: ScheduledNotification,
+  event: 'delete' | 'addOrUpdate'
+}
+
+export interface NotificationRepository {
+  getAllNotificationsForShuttleAndStopId(shuttleId: string, stopId: string): Promise<ScheduledNotification[]>;
+  getSecondsThresholdForNotificationIfExists(lookupArguments: NotificationLookupArguments): Promise<number | null>;
+  isNotificationScheduled(lookupArguments: NotificationLookupArguments): Promise<boolean>;
+  addOrUpdateNotification(notification: ScheduledNotification): Promise<void>;
+  deleteNotificationIfExists(lookupArguments: NotificationLookupArguments): Promise<void>;
+
+  subscribeToNotificationChanges(listener: Listener): void;
+  unsubscribeFromNotificationChanges(listener: Listener): void;
+}

src/repositories/notifications/RedisNotificationRepository.ts

@@ -0,0 +1,113 @@
+import { TupleKey } from '../../types/TupleKey';
+import {
+  Listener,
+  NotificationEvent,
+  NotificationLookupArguments,
+  NotificationRepository,
+  ScheduledNotification
+} from "./NotificationRepository";
+import { BaseRedisRepository } from "../BaseRedisRepository";
+
+export class RedisNotificationRepository extends BaseRedisRepository implements NotificationRepository {
+  private listeners: Listener[] = [];
+  private readonly NOTIFICATION_KEY_PREFIX = 'notification:';
+
+  private getNotificationKey = (shuttleId: string, stopId: string): string => {
+    const tuple = new TupleKey(shuttleId, stopId);
+    return `${this.NOTIFICATION_KEY_PREFIX}${tuple.toString()}`;
+  };
+
+  public addOrUpdateNotification = async (notification: ScheduledNotification): Promise<void> => {
+    const { shuttleId, stopId, deviceId, secondsThreshold } = notification;
+    const key = this.getNotificationKey(shuttleId, stopId);
+
+    await this.redisClient.hSet(key, deviceId, secondsThreshold.toString());
+
+    this.listeners.forEach((listener: Listener) => {
+      const event: NotificationEvent = {
+        event: 'addOrUpdate',
+        notification
+      };
+      listener(event);
+    });
+  };
+
+  public deleteNotificationIfExists = async (lookupArguments: NotificationLookupArguments): Promise<void> => {
+    const { shuttleId, stopId, deviceId } = lookupArguments;
+    const key = this.getNotificationKey(shuttleId, stopId);
+
+    const secondsThreshold = await this.redisClient.hGet(key, deviceId);
+    if (secondsThreshold) {
+      await this.redisClient.hDel(key, deviceId);
+
+      // Check if hash is empty and delete it if so
+      const remainingFields = await this.redisClient.hLen(key);
+      if (remainingFields === 0) {
+        await this.redisClient.del(key);
+      }
+
+      this.listeners.forEach((listener) => {
+        const event: NotificationEvent = {
+          event: 'delete',
+          notification: {
+            deviceId,
+            shuttleId,
+            stopId,
+            secondsThreshold: parseInt(secondsThreshold)
+          }
+        };
+        listener(event);
+      });
+    }
+  };
+
+  public getAllNotificationsForShuttleAndStopId = async (
+    shuttleId: string,
+    stopId: string
+  ): Promise<ScheduledNotification[]> => {
+    const key = this.getNotificationKey(shuttleId, stopId);
+    const allNotifications = await this.redisClient.hGetAll(key);
+
+    return Object.entries(allNotifications).map(([deviceId, secondsThreshold]) => ({
+      shuttleId,
+      stopId,
+      deviceId,
+      secondsThreshold: parseInt(secondsThreshold)
+    }));
+  };
+
+  public getSecondsThresholdForNotificationIfExists = async (
+    lookupArguments: NotificationLookupArguments
+  ): Promise<number | null> => {
+    const { shuttleId, stopId, deviceId } = lookupArguments;
+    const key = this.getNotificationKey(shuttleId, stopId);
+
+    const threshold = await this.redisClient.hGet(key, deviceId);
+    return threshold ? parseInt(threshold) : null;
+  };
+
+  public isNotificationScheduled = async (
+    lookupArguments: NotificationLookupArguments
+  ): Promise<boolean> => {
+    const threshold = await this.getSecondsThresholdForNotificationIfExists(lookupArguments);
+    return threshold !== null;
+  };
+
+  public subscribeToNotificationChanges = (listener: Listener): void => {
+    const index = this.listeners.findIndex(
+      (existingListener) => existingListener === listener
+    );
+    if (index < 0) {
+      this.listeners.push(listener);
+    }
+  };
+
+  public unsubscribeFromNotificationChanges = (listener: Listener): void => {
+    const index = this.listeners.findIndex(
+      (existingListener) => existingListener === listener
+    );
+    if (index >= 0) {
+      this.listeners.splice(index, 1);
+    }
+  };
+}