Complete persistent notifications migration (#16476)

* Complete migration of persistent notifications

* Complete migration of persistent notifications

* revert

src/data/persistent_notification.ts

@@ -1,7 +1,7 @@
 import {
   Connection,
-  createCollection,
   HassEntity,
+  UnsubscribeFunc,
 } from "home-assistant-js-websocket";
 
 export interface PersitentNotificationEntity extends HassEntity {
@@ -19,25 +19,52 @@ export interface PersistentNotification {
   status: "read" | "unread";
 }
 
-const fetchNotifications = (conn) =>
-  conn.sendMessagePromise({
-    type: "persistent_notification/get",
-  });
+export interface PersistentNotifications {
+  [notificationId: string]: PersistentNotification;
+}
 
-const subscribeUpdates = (conn, store) =>
-  conn.subscribeEvents(
-    () => fetchNotifications(conn).then((ntf) => store.setState(ntf, true)),
-    "persistent_notifications_updated"
-  );
+export interface PersistentNotificationMessage {
+  type: "added" | "removed" | "current" | "updated";
+  notifications: PersistentNotifications;
+}
 
 export const subscribeNotifications = (
   conn: Connection,
   onChange: (notifications: PersistentNotification[]) => void
-) =>
-  createCollection<PersistentNotification[]>(
-    "_ntf",
-    fetchNotifications,
-    subscribeUpdates,
-    conn,
-    onChange
+): UnsubscribeFunc => {
+  const params = {
+    type: "persistent_notification/subscribe",
+  };
+  const stream = new NotificationStream();
+  const subscription = conn.subscribeMessage<PersistentNotificationMessage>(
+    (message) => onChange(stream.processMessage(message)),
+    params
   );
+  return () => {
+    subscription.then((unsub) => unsub?.());
+  };
+};
+
+class NotificationStream {
+  notifications: PersistentNotifications;
+
+  constructor() {
+    this.notifications = {};
+  }
+
+  processMessage(
+    streamMessage: PersistentNotificationMessage
+  ): PersistentNotification[] {
+    if (streamMessage.type === "removed") {
+      for (const notificationId of Object.keys(streamMessage.notifications)) {
+        delete this.notifications[notificationId];
+      }
+    } else {
+      this.notifications = {
+        ...this.notifications,
+        ...streamMessage.notifications,
+      };
+    }
+    return Object.values(this.notifications);
+  }
+}