implement RedisNotificationRepository

src/repositories/RedisNotificationRepository.ts

@@ -1,5 +1,7 @@
+import { TupleKey } from '../types/TupleKey';
 import {
   Listener,
+  NotificationEvent,
   NotificationLookupArguments,
   NotificationRepository,
   ScheduledNotification
@@ -7,6 +9,9 @@ import {
 import { createClient } from "redis";
 
 export class RedisNotificationRepository implements NotificationRepository {
+  private listeners: Listener[] = [];
+  private readonly NOTIFICATION_KEY_PREFIX = 'notification:';
+
   constructor(
     private redisClient = createClient({
       url: process.env.REDIS_URL,
@@ -37,27 +42,102 @@ export class RedisNotificationRepository implements NotificationRepository {
     await this.redisClient.flushAll();
   }
 
+  private getNotificationKey(shuttleId: string, stopId: string): string {
+    const tuple = new TupleKey(shuttleId, stopId);
+    return `${this.NOTIFICATION_KEY_PREFIX}${tuple.toString()}`;
+  }
+
   public async addOrUpdateNotification(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 async deleteNotificationIfExists(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);
       }
 
-  public async getAllNotificationsForShuttleAndStopId(shuttleId: string, stopId: string): Promise<ScheduledNotification[]> {
-    return [];
+      this.listeners.forEach((listener) => {
+        const event: NotificationEvent = {
+          event: 'delete',
+          notification: {
+            deviceId,
+            shuttleId,
+            stopId,
+            secondsThreshold: parseInt(secondsThreshold)
+          }
+        };
+        listener(event);
+      });
+    }
   }
 
-  public async getSecondsThresholdForNotificationIfExists(lookupArguments: NotificationLookupArguments): Promise<number | null> {
-    return null;
+  public async getAllNotificationsForShuttleAndStopId(
+    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 async isNotificationScheduled(lookupArguments: NotificationLookupArguments): Promise<boolean> {
-    return false;
+  public async getSecondsThresholdForNotificationIfExists(
+    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;
   }
 
-  subscribeToNotificationChanges(listener: Listener): void {
+  public async isNotificationScheduled(
+    lookupArguments: NotificationLookupArguments
+  ): Promise<boolean> {
+    const threshold = await this.getSecondsThresholdForNotificationIfExists(lookupArguments);
+    return threshold !== null;
   }
 
-  unsubscribeFromNotificationChanges(listener: Listener): void {
+  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);
+    }
   }
 }

test/repositories/RedisNotificationRepositoryTests.test.ts

@@ -1,6 +0,0 @@
-// Test additional edge cases like Redis failing to connect, etc.
-
-import { describe, it } from "@jest/globals";
-
-describe("RedisNotificationRepository", () => {
-});