From f441de806fdbf100b5874dfb7791e56d96083c34 Mon Sep 17 00:00:00 2001
From: Natty <natty.sh.git@gmail.com>
Date: Thu, 18 Jan 2024 03:26:26 +0100
Subject: [PATCH] Frontend: Basic notification receiving via SSE

---
 ...MkNotification.vue => MagNotification.vue} |   0
 ...tionToast.vue => MagNotificationToast.vue} |   2 +-
 ...Notifications.vue => MagNotifications.vue} |  16 +-
 fe_calckey/frontend/client/src/config.ts      |   1 +
 .../client/src/pages/notifications.vue        |   2 +-
 fe_calckey/frontend/client/src/stream.ts      |   7 +-
 .../client/src/ui/_common_/common.vue         |  12 +-
 .../src/ui/deck/notifications-column.vue      |   4 +-
 .../client/src/widgets/notifications.vue      |  10 +-
 .../frontend/magnetar-common/package.json     |   3 +
 .../frontend/magnetar-common/src/index.ts     |  12 +-
 .../magnetar-common/src/sse-listener.ts       | 142 ++++++++++++++++++
 .../frontend/magnetar-common/src/types.ts     |   1 +
 .../magnetar-common/src/types/ChannelEvent.ts |   4 +
 fe_calckey/frontend/pnpm-lock.yaml            |   8 +
 src/api_v1/streaming.rs                       |  43 ++++--
 16 files changed, 225 insertions(+), 42 deletions(-)
 rename fe_calckey/frontend/client/src/components/{MkNotification.vue => MagNotification.vue} (100%)
 rename fe_calckey/frontend/client/src/components/{MkNotificationToast.vue => MagNotificationToast.vue} (96%)
 rename fe_calckey/frontend/client/src/components/{MkNotifications.vue => MagNotifications.vue} (90%)
 create mode 100644 fe_calckey/frontend/magnetar-common/src/sse-listener.ts
 create mode 100644 fe_calckey/frontend/magnetar-common/src/types/ChannelEvent.ts

diff --git a/fe_calckey/frontend/client/src/components/MkNotification.vue b/fe_calckey/frontend/client/src/components/MagNotification.vue
similarity index 100%
rename from fe_calckey/frontend/client/src/components/MkNotification.vue
rename to fe_calckey/frontend/client/src/components/MagNotification.vue
diff --git a/fe_calckey/frontend/client/src/components/MkNotificationToast.vue b/fe_calckey/frontend/client/src/components/MagNotificationToast.vue
similarity index 96%
rename from fe_calckey/frontend/client/src/components/MkNotificationToast.vue
rename to fe_calckey/frontend/client/src/components/MagNotificationToast.vue
index 8fbefbd..15c1ffa 100644
--- a/fe_calckey/frontend/client/src/components/MkNotificationToast.vue
+++ b/fe_calckey/frontend/client/src/components/MagNotificationToast.vue
@@ -16,7 +16,7 @@
 
 <script lang="ts" setup>
 import { onMounted } from "vue";
-import XNotification from "@/components/MkNotification.vue";
+import XNotification from "@/components/MagNotification.vue";
 import * as os from "@/os";
 
 defineProps<{
diff --git a/fe_calckey/frontend/client/src/components/MkNotifications.vue b/fe_calckey/frontend/client/src/components/MagNotifications.vue
similarity index 90%
rename from fe_calckey/frontend/client/src/components/MkNotifications.vue
rename to fe_calckey/frontend/client/src/components/MagNotifications.vue
index d059cb1..33623e3 100644
--- a/fe_calckey/frontend/client/src/components/MkNotifications.vue
+++ b/fe_calckey/frontend/client/src/components/MagNotifications.vue
@@ -44,10 +44,10 @@
 
 <script lang="ts" setup>
 import { onMounted, onUnmounted, ref } from "vue";
-import XNotification from "@/components/MkNotification.vue";
+import XNotification from "@/components/MagNotification.vue";
 import XList from "@/components/MkDateSeparatedList.vue";
 import XNote from "@/components/MagNote.vue";
-import { stream } from "@/stream";
+import { magStream, stream } from "@/stream";
 import { $i } from "@/account";
 import MagPagination, { Paging } from "@/components/MagPagination.vue";
 import { endpoints, types } from "magnetar-common";
@@ -89,18 +89,20 @@ const onNotification = (notification) => {
     }
 };
 
+let notifStream;
 let connection;
 
 onMounted(() => {
+    notifStream = magStream.useFiltered("Notification", onNotification);
+
     connection = stream.useChannel("main");
-    connection.on("notification", onNotification);
     connection.on("readAllNotifications", () => {
         if (pagingComponent.value) {
             for (const item of pagingComponent.value.queue) {
-                item.isRead = true;
+                item.is_read = true;
             }
             for (const item of pagingComponent.value.items) {
-                item.isRead = true;
+                item.is_read = true;
             }
         }
     });
@@ -110,7 +112,7 @@ onMounted(() => {
                 if (
                     notificationIds.includes(pagingComponent.value.queue[i].id)
                 ) {
-                    pagingComponent.value.queue[i].isRead = true;
+                    pagingComponent.value.queue[i].is_read = true;
                 }
             }
             for (
@@ -121,7 +123,7 @@ onMounted(() => {
                 if (
                     notificationIds.includes(pagingComponent.value.items[i].id)
                 ) {
-                    pagingComponent.value.items[i].isRead = true;
+                    pagingComponent.value.items[i].is_read = true;
                 }
             }
         }
diff --git a/fe_calckey/frontend/client/src/config.ts b/fe_calckey/frontend/client/src/config.ts
index 158f10c..6ff92fd 100644
--- a/fe_calckey/frontend/client/src/config.ts
+++ b/fe_calckey/frontend/client/src/config.ts
@@ -7,6 +7,7 @@ export const host = _HOST || address.host;
 export const hostname = address.hostname;
 export const url = _REMOTE_URL || address.origin;
 export const apiUrl = `${url}/api`;
+export const magStreamingUrl = `${url}/mag/v1`;
 export const feApiUrl = `${url}/fe-api`;
 export const wsUrl = `${url
     .replace("http://", "ws://")
diff --git a/fe_calckey/frontend/client/src/pages/notifications.vue b/fe_calckey/frontend/client/src/pages/notifications.vue
index 077254b..1a5d81d 100644
--- a/fe_calckey/frontend/client/src/pages/notifications.vue
+++ b/fe_calckey/frontend/client/src/pages/notifications.vue
@@ -62,7 +62,7 @@ import { computed, ref, watch } from "vue";
 import { Virtual } from "swiper";
 import { Swiper, SwiperSlide } from "swiper/vue";
 import { notificationTypes } from "calckey-js";
-import XNotifications from "@/components/MkNotifications.vue";
+import XNotifications from "@/components/MagNotifications.vue";
 import XNotes from "@/components/MkNotes.vue";
 import * as os from "@/os";
 import { i18n } from "@/i18n";
diff --git a/fe_calckey/frontend/client/src/stream.ts b/fe_calckey/frontend/client/src/stream.ts
index e1d4a27..f0fe9f7 100644
--- a/fe_calckey/frontend/client/src/stream.ts
+++ b/fe_calckey/frontend/client/src/stream.ts
@@ -1,7 +1,8 @@
 import * as Misskey from "calckey-js";
 import { markRaw } from "vue";
 import { $i } from "@/account";
-import { url } from "@/config";
+import { magStreamingUrl, url } from "@/config";
+import { MagEventChannel } from "magnetar-common";
 
 export const stream = markRaw(
     new Misskey.Stream(
@@ -22,3 +23,7 @@ function heartbeat(): void {
     }
     window.setTimeout(heartbeat, 1000 * 60);
 }
+
+export const magStream = markRaw(
+    new MagEventChannel(magStreamingUrl, $i ? $i.token : null)
+);
diff --git a/fe_calckey/frontend/client/src/ui/_common_/common.vue b/fe_calckey/frontend/client/src/ui/_common_/common.vue
index a90bc17..c5de0b0 100644
--- a/fe_calckey/frontend/client/src/ui/_common_/common.vue
+++ b/fe_calckey/frontend/client/src/ui/_common_/common.vue
@@ -19,11 +19,12 @@
 <script lang="ts" setup>
 import { defineAsyncComponent } from "vue";
 import { swInject } from "./sw-inject";
-import { popup, popups, pendingApiRequestsCount } from "@/os";
+import { popup, popups } from "@/os";
 import { uploads } from "@/scripts/upload";
 import * as sound from "@/scripts/sound";
 import { $i } from "@/account";
-import { stream } from "@/stream";
+import { magStream, stream } from "@/stream";
+import { PackNotification } from "magnetar-common/built/types/PackNotification";
 
 const XStreamIndicator = defineAsyncComponent(
     () => import("./stream-indicator.vue")
@@ -32,7 +33,7 @@ const XUpload = defineAsyncComponent(() => import("./upload.vue"));
 
 const dev = _DEV_;
 
-const onNotification = (notification) => {
+const onNotification = (notification: PackNotification) => {
     if ($i.mutingNotificationTypes.includes(notification.type)) return;
 
     if (document.visibilityState === "visible") {
@@ -42,7 +43,7 @@ const onNotification = (notification) => {
 
         popup(
             defineAsyncComponent(
-                () => import("@/components/MkNotificationToast.vue")
+                () => import("@/components/MagNotificationToast.vue")
             ),
             {
                 notification,
@@ -56,8 +57,7 @@ const onNotification = (notification) => {
 };
 
 if ($i) {
-    const connection = stream.useChannel("main", null, "UI");
-    connection.on("notification", onNotification);
+    const connection = magStream.useFiltered("Notification", onNotification);
 
     //#region Listen message from SW
     if ("serviceWorker" in navigator) {
diff --git a/fe_calckey/frontend/client/src/ui/deck/notifications-column.vue b/fe_calckey/frontend/client/src/ui/deck/notifications-column.vue
index 58699a4..987ac18 100644
--- a/fe_calckey/frontend/client/src/ui/deck/notifications-column.vue
+++ b/fe_calckey/frontend/client/src/ui/deck/notifications-column.vue
@@ -17,9 +17,9 @@
 <script lang="ts" setup>
 import { defineAsyncComponent } from "vue";
 import XColumn from "./column.vue";
-import { updateColumn } from "./deck-store";
 import type { Column } from "./deck-store";
-import XNotifications from "@/components/MkNotifications.vue";
+import { updateColumn } from "./deck-store";
+import XNotifications from "@/components/MagNotifications.vue";
 import * as os from "@/os";
 import { i18n } from "@/i18n";
 
diff --git a/fe_calckey/frontend/client/src/widgets/notifications.vue b/fe_calckey/frontend/client/src/widgets/notifications.vue
index 72dd0ce..c089adb 100644
--- a/fe_calckey/frontend/client/src/widgets/notifications.vue
+++ b/fe_calckey/frontend/client/src/widgets/notifications.vue
@@ -26,16 +26,10 @@
 
 <script lang="ts" setup>
 import { defineAsyncComponent } from "vue";
-import {
-    useWidgetPropsManager,
-    Widget,
-    WidgetComponentEmits,
-    WidgetComponentExpose,
-    WidgetComponentProps,
-} from "./widget";
+import { useWidgetPropsManager, Widget, WidgetComponentExpose } from "./widget";
 import { GetFormResultType } from "@/scripts/form";
 import MkContainer from "@/components/MkContainer.vue";
-import XNotifications from "@/components/MkNotifications.vue";
+import XNotifications from "@/components/MagNotifications.vue";
 import * as os from "@/os";
 import { i18n } from "@/i18n";
 
diff --git a/fe_calckey/frontend/magnetar-common/package.json b/fe_calckey/frontend/magnetar-common/package.json
index 32dfc4c..739d19e 100644
--- a/fe_calckey/frontend/magnetar-common/package.json
+++ b/fe_calckey/frontend/magnetar-common/package.json
@@ -11,5 +11,8 @@
   "description": "A library with common utilities for Magnetar application development",
   "devDependencies": {
     "typescript": "^5.1.6"
+  },
+  "dependencies": {
+    "eventemitter3": "^5.0.1"
   }
 }
diff --git a/fe_calckey/frontend/magnetar-common/src/index.ts b/fe_calckey/frontend/magnetar-common/src/index.ts
index c6a753c..c2da0e8 100644
--- a/fe_calckey/frontend/magnetar-common/src/index.ts
+++ b/fe_calckey/frontend/magnetar-common/src/index.ts
@@ -13,9 +13,11 @@ import {
     FrontendApiEndpoints,
 } from "./fe-api";
 
-export * as types from "./types";
-export * as packed from "./packed";
-export * as endpoints from "./endpoints";
+import { MagEventChannel } from "./sse-listener";
+
+import * as types from "./types";
+import * as packed from "./packed";
+import * as endpoints from "./endpoints";
 
 export {
     Method,
@@ -27,4 +29,8 @@ export {
     feEndpoints,
     FrontendApiEndpoint,
     FrontendApiEndpoints,
+    MagEventChannel,
+    types,
+    packed,
+    endpoints,
 };
diff --git a/fe_calckey/frontend/magnetar-common/src/sse-listener.ts b/fe_calckey/frontend/magnetar-common/src/sse-listener.ts
new file mode 100644
index 0000000..db951c1
--- /dev/null
+++ b/fe_calckey/frontend/magnetar-common/src/sse-listener.ts
@@ -0,0 +1,142 @@
+import { EventEmitter } from "eventemitter3";
+import { ChannelEvent } from "./types/ChannelEvent";
+
+export type MagChannelState = "connected" | "exponentialBackoff" | "failed";
+
+export class MagEventChannel extends EventEmitter<{
+    stateChange: MagChannelState;
+    message: ChannelEvent;
+    close: "cancelled";
+}> {
+    private readonly baseUrl: string;
+    private attempts = 0;
+    private readonly maxAttempts: number;
+    private readonly token: string | null;
+    private readonly backoffFactor: number;
+    private readonly backoffBase: number;
+    private readonly closePromise: Promise<"cancelled">;
+
+    public constructor(
+        baseUrl: string,
+        token: string | null,
+        maxReconnectAttempts: number = 12,
+        backoffFactor: number = 1.618,
+        backoffBase: number = 500.0
+    ) {
+        super();
+
+        this.baseUrl = baseUrl;
+        this.token = token;
+        this.maxAttempts = maxReconnectAttempts;
+        this.backoffFactor = backoffFactor;
+        this.backoffBase = backoffBase;
+        this.closePromise = new Promise((resolve) => {
+            this.on("close", resolve);
+        });
+        this.connect().then();
+    }
+
+    public useFiltered<T extends ChannelEvent["type"]>(
+        messageType: T,
+        listener: (val: (ChannelEvent & { type: T })["body"]) => void
+    ) {
+        const cb = (val: ChannelEvent) => {
+            if (val.type != messageType) return;
+
+            listener((val as ChannelEvent & { type: T }).body);
+        };
+        this.on("message", cb);
+        return cb;
+    }
+
+    private async connect() {
+        if (this.attempts >= this.maxAttempts) {
+            this.emit("stateChange", "failed");
+            return;
+        }
+
+        const authorization = this.token ? `Bearer ${this.token}` : undefined;
+        const baseUrl = this.baseUrl.replace(/\/+$/, "");
+
+        const response = await fetch(`${baseUrl}/streaming`, {
+            method: "GET",
+            headers: authorization ? { authorization } : {},
+            credentials: "omit",
+            cache: "no-cache",
+        }).catch((e) => {
+            console.error(e);
+            return null;
+        });
+
+        if (
+            response === null ||
+            response.status >= 500 ||
+            response.body === null
+        ) {
+            this.emit("stateChange", "exponentialBackoff");
+            setTimeout(
+                () => this.connect(),
+                this.backoffBase * Math.pow(this.backoffFactor, this.attempts)
+            );
+            this.attempts++;
+            return;
+        }
+
+        if (response.status >= 400 && response.status < 500) {
+            this.emit("stateChange", "failed");
+            return;
+        }
+
+        this.attempts = 0;
+
+        const decoderStream = new TextDecoderStream();
+        const reader = response.body.pipeThrough(decoderStream).getReader();
+        this.emit("stateChange", "connected");
+
+        let buf = "";
+
+        while (true) {
+            const res = await Promise.race([reader.read(), this.closePromise]);
+
+            if (res === "cancelled") break;
+
+            if (res.done) {
+                this.emit("stateChange", "exponentialBackoff");
+                setTimeout(
+                    () => this.connect(),
+                    this.backoffBase *
+                        Math.pow(this.backoffFactor, this.attempts)
+                );
+                this.attempts++;
+                break;
+            }
+
+            buf += res.value;
+
+            const splitIndex = buf.indexOf("\n\n");
+            if (splitIndex === -1) {
+                continue;
+            }
+
+            const rawValue = buf.substring(0, splitIndex);
+            buf = buf.substring(splitIndex + 2);
+
+            if (rawValue.startsWith(":")) continue;
+
+            const text = rawValue
+                .split("\n")
+                .filter((l) => l.startsWith("data: "))
+                .map((l) => l.substring("data: ".length))
+                .join("\n");
+
+            if (!text) continue;
+
+            const data = JSON.parse(text) as ChannelEvent;
+            this.emit("message", data);
+        }
+    }
+
+    public async close() {
+        this.emit("close", "cancelled");
+    }
+}
diff --git a/fe_calckey/frontend/magnetar-common/src/types.ts b/fe_calckey/frontend/magnetar-common/src/types.ts
index 84ca16e..8236127 100644
--- a/fe_calckey/frontend/magnetar-common/src/types.ts
+++ b/fe_calckey/frontend/magnetar-common/src/types.ts
@@ -55,3 +55,4 @@ export { NotificationNoteExt } from "./types/NotificationNoteExt";
 export { NotificationReactionExt } from "./types/NotificationReactionExt";
 export { NotificationUserExt } from "./types/NotificationUserExt";
 export { NotificationsReq } from "./types/NotificationsReq";
+export { ChannelEvent } from "./types/ChannelEvent";
diff --git a/fe_calckey/frontend/magnetar-common/src/types/ChannelEvent.ts b/fe_calckey/frontend/magnetar-common/src/types/ChannelEvent.ts
new file mode 100644
index 0000000..94395c5
--- /dev/null
+++ b/fe_calckey/frontend/magnetar-common/src/types/ChannelEvent.ts
@@ -0,0 +1,4 @@
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+import type { PackNotification } from "./PackNotification";
+
+export type ChannelEvent = { "type": "Notification", "body": PackNotification };
\ No newline at end of file
diff --git a/fe_calckey/frontend/pnpm-lock.yaml b/fe_calckey/frontend/pnpm-lock.yaml
index 1769d0e..713e842 100644
--- a/fe_calckey/frontend/pnpm-lock.yaml
+++ b/fe_calckey/frontend/pnpm-lock.yaml
@@ -358,6 +358,10 @@ importers:
         version: 4.1.0(vue@3.3.4)
 
   magnetar-common:
+    dependencies:
+      eventemitter3:
+        specifier: ^5.0.1
+        version: 5.0.1
     devDependencies:
       typescript:
         specifier: ^5.1.6
@@ -3269,6 +3273,10 @@ packages:
   /eventemitter3@4.0.7:
     resolution: {integrity: sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw==}
 
+  /eventemitter3@5.0.1:
+    resolution: {integrity: sha512-GWkBvjiSZK87ELrYOSESUYeVIc9mvLLf/nXalMOS5dYrgZq9o5OVkbZAVM06CVxYsCwH9BDZFPlQTlPA1j4ahA==}
+    dev: false
+
   /events@3.3.0:
     resolution: {integrity: sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==}
     engines: {node: '>=0.8.x'}
diff --git a/src/api_v1/streaming.rs b/src/api_v1/streaming.rs
index fca38d2..b69fa6c 100644
--- a/src/api_v1/streaming.rs
+++ b/src/api_v1/streaming.rs
@@ -11,6 +11,7 @@ use futures_util::StreamExt as _;
 use magnetar_calckey_model::model_ext::IdShape;
 use magnetar_calckey_model::{CalckeySub, MainStreamMessage, SubMessage};
 use magnetar_sdk::types::streaming::ChannelEvent;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc;
@@ -47,6 +48,11 @@ pub async fn handle_streaming(
                 };
 
                 if id != user_id {
+                    trace!(
+                        "Skipping message intended for {} in channel {}",
+                        id,
+                        user_id
+                    );
                     return;
                 }
 
@@ -59,11 +65,15 @@ pub async fn handle_streaming(
 
     drop_on_close(sub, tx);
 
+    let event_counter = Arc::new(AtomicU64::default());
     let stream = ReceiverStream::new(rx).filter_map(move |m| {
+        trace!("Processing raw message: {:?}", m);
+
         let service = service.clone();
         let self_user = self_user.clone();
+        let event_counter = event_counter.clone();
         async move {
-            match m {
+            let message = match m {
                 MainStreamMessage::Notification(IdShape { id }) => {
                     let ctx = PackingContext::new(service, Some(self_user.clone()))
                         .await
@@ -75,20 +85,27 @@ pub async fn handle_streaming(
                     let notification_model = NotificationModel;
 
                     Some(
-                        Event::default().json_data(ChannelEvent::Notification(
-                            notification_model
-                                .get_notification(&ctx, &id, &self_user.id)
-                                .await
-                                .map_err(|e| {
-                                    error!("Failed to fetch notification: {}", e);
-                                    e
-                                })
-                                .ok()
-                                .flatten()?,
-                        )),
+                        Event::default()
+                            .id(event_counter.fetch_add(1, Ordering::Relaxed).to_string())
+                            .event("message")
+                            .json_data(ChannelEvent::Notification(
+                                notification_model
+                                    .get_notification(&ctx, &id, &self_user.id)
+                                    .await
+                                    .map_err(|e| {
+                                        error!("Failed to fetch notification: {}", e);
+                                        e
+                                    })
+                                    .ok()
+                                    .flatten()?,
+                            )),
                     )
                 }
-            }
+            };
+
+            trace!("Sending message: {:?}", message);
+
+            message
         }
     });