✨ Realtime Notify

pkg/grpc/notify.go

@@ -32,12 +32,13 @@ func (v *Server) NotifyUser(_ context.Context, in *proto.NotifyRequest) (*proto.
 		Content:     in.GetContent(),
 		Links:       links,
 		IsImportant: in.GetIsImportant(),
+		IsRealtime:  in.GetIsRealtime(),
 		ReadAt:      nil,
 		RecipientID: user.ID,
 		SenderID:    &client.ID,
 	}
 
-	if in.GetRealtime() {
+	if in.GetIsRealtime() {
 		if err := services.PushNotification(notification); err != nil {
 			return nil, err
 		}

pkg/grpc/proto/notify.pb.go

@@ -87,7 +87,7 @@ type NotifyRequest struct {
 	RecipientId  uint64        `protobuf:"varint,5,opt,name=recipient_id,json=recipientId,proto3" json:"recipient_id,omitempty"`
 	ClientId     string        `protobuf:"bytes,6,opt,name=client_id,json=clientId,proto3" json:"client_id,omitempty"`
 	ClientSecret string        `protobuf:"bytes,7,opt,name=client_secret,json=clientSecret,proto3" json:"client_secret,omitempty"`
-	Realtime     bool          `protobuf:"varint,8,opt,name=realtime,proto3" json:"realtime,omitempty"`
+	IsRealtime   bool          `protobuf:"varint,8,opt,name=is_realtime,json=isRealtime,proto3" json:"is_realtime,omitempty"`
 }
 
 func (x *NotifyRequest) Reset() {
@@ -171,9 +171,9 @@ func (x *NotifyRequest) GetClientSecret() string {
 	return ""
 }
 
-func (x *NotifyRequest) GetRealtime() bool {
+func (x *NotifyRequest) GetIsRealtime() bool {
 	if x != nil {
-		return x.Realtime
+		return x.IsRealtime
 	}
 	return false
 }
@@ -232,7 +232,7 @@ var file_notify_proto_rawDesc = []byte{
 	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x34, 0x0a, 0x0a, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x4c,
 	0x69, 0x6e, 0x6b, 0x12, 0x14, 0x0a, 0x05, 0x6c, 0x61, 0x62, 0x65, 0x6c, 0x18, 0x01, 0x20, 0x01,
 	0x28, 0x09, 0x52, 0x05, 0x6c, 0x61, 0x62, 0x65, 0x6c, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x72, 0x6c,
-	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x72, 0x6c, 0x22, 0x90, 0x02, 0x0a, 0x0d,
+	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x72, 0x6c, 0x22, 0x95, 0x02, 0x0a, 0x0d,
 	0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x18, 0x0a,
 	0x07, 0x73, 0x75, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07,
 	0x73, 0x75, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x12, 0x18, 0x0a, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65,
@@ -248,16 +248,17 @@ var file_notify_proto_rawDesc = []byte{
 	0x01, 0x28, 0x09, 0x52, 0x08, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x23, 0x0a,
 	0x0d, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x18, 0x07,
 	0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x53, 0x65, 0x63, 0x72,
-	0x65, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x72, 0x65, 0x61, 0x6c, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x08,
+	0x65, 0x74, 0x12, 0x1f, 0x0a, 0x0b, 0x69, 0x73, 0x5f, 0x72, 0x65, 0x61, 0x6c, 0x74, 0x69, 0x6d,
-	0x20, 0x01, 0x28, 0x08, 0x52, 0x08, 0x72, 0x65, 0x61, 0x6c, 0x74, 0x69, 0x6d, 0x65, 0x22, 0x26,
+	0x65, 0x18, 0x08, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0a, 0x69, 0x73, 0x52, 0x65, 0x61, 0x6c, 0x74,
-	0x0a, 0x0b, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x17, 0x0a,
+	0x69, 0x6d, 0x65, 0x22, 0x26, 0x0a, 0x0b, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x52, 0x65, 0x70,
-	0x07, 0x69, 0x73, 0x5f, 0x73, 0x65, 0x6e, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x08, 0x52, 0x06,
+	0x6c, 0x79, 0x12, 0x17, 0x0a, 0x07, 0x69, 0x73, 0x5f, 0x73, 0x65, 0x6e, 0x74, 0x18, 0x01, 0x20,
-	0x69, 0x73, 0x53, 0x65, 0x6e, 0x74, 0x32, 0x42, 0x0a, 0x06, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79,
+	0x01, 0x28, 0x08, 0x52, 0x06, 0x69, 0x73, 0x53, 0x65, 0x6e, 0x74, 0x32, 0x42, 0x0a, 0x06, 0x4e,
-	0x12, 0x38, 0x0a, 0x0a, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x55, 0x73, 0x65, 0x72, 0x12, 0x14,
+	0x6f, 0x74, 0x69, 0x66, 0x79, 0x12, 0x38, 0x0a, 0x0a, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x55,
-	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x52, 0x65, 0x71,
+	0x73, 0x65, 0x72, 0x12, 0x14, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x4e, 0x6f, 0x74, 0x69,
-	0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x4e, 0x6f, 0x74,
+	0x66, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x70, 0x72, 0x6f, 0x74,
-	0x69, 0x66, 0x79, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x22, 0x00, 0x42, 0x09, 0x5a, 0x07, 0x2e, 0x3b,
+	0x6f, 0x2e, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x22, 0x00, 0x42,
-	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x09, 0x5a, 0x07, 0x2e, 0x3b, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x33,
 }
 
 var (

pkg/grpc/proto/notify.proto

@@ -21,7 +21,7 @@ message NotifyRequest {
   uint64 recipient_id = 5;
   string client_id = 6;
   string client_secret = 7;
-  bool realtime = 8;
+  bool is_realtime = 8;
 }
 
 message NotifyReply {

pkg/models/notifications.go

@@ -12,6 +12,7 @@ type Notification struct {
 	Content     string                                `json:"content"`
 	Links       datatypes.JSONSlice[NotificationLink] `json:"links"`
 	IsImportant bool                                  `json:"is_important"`
+	IsRealtime  bool                                  `json:"is_realtime" gorm:"-"`
 	ReadAt      *time.Time                            `json:"read_at"`
 	SenderID    *uint                                 `json:"sender_id"`
 	RecipientID uint                                  `json:"recipient_id"`

pkg/server/notify_api.go

@@ -14,6 +14,7 @@ func notifyUser(c *fiber.Ctx) error {
 		Content      string                    `json:"content" validate:"required,max=3072"`
 		Links        []models.NotificationLink `json:"links"`
 		IsImportant  bool                      `json:"is_important"`
+		IsRealtime   bool                      `json:"is_realtime"`
 		UserID       uint                      `json:"user_id" validate:"required"`
 	}
 
@@ -33,16 +34,23 @@ func notifyUser(c *fiber.Ctx) error {
 
 	notification := models.Notification{
 		Subject:     data.Subject,
-		Content:     data.Subject,
+		Content:     data.Content,
 		Links:       data.Links,
 		IsImportant: data.IsImportant,
+		IsRealtime:  data.IsRealtime,
 		ReadAt:      nil,
 		RecipientID: user.ID,
 		SenderID:    &client.ID,
 	}
 
-	if err := services.NewNotification(notification); err != nil {
+	if data.IsRealtime {
-		return fiber.NewError(fiber.StatusBadRequest, err.Error())
+		if err := services.PushNotification(notification); err != nil {
+			return fiber.NewError(fiber.StatusBadRequest, err.Error())
+		}
+	} else {
+		if err := services.NewNotification(notification); err != nil {
+			return fiber.NewError(fiber.StatusBadRequest, err.Error())
+		}
 	}
 
 	return c.SendStatus(fiber.StatusOK)

pkg/server/notify_ws.go

@@ -16,12 +16,8 @@ func listenNotifications(c *websocket.Conn) {
 	// Event loop
 	var err error
 	for {
-		message := services.WsNotifyQueue[user.ID]
+		if _, _, err = c.ReadMessage(); err != nil {
-
+			break
-		if message != nil {
-			if err = c.WriteMessage(1, message); err != nil {
-				break
-			}
 		}
 	}
 

pkg/services/connections.go

@@ -5,8 +5,12 @@ import (
 	"github.com/gofiber/contrib/websocket"
 )
 
+type WsPushRequest struct {
+	Payload     []byte
+	RecipientID uint
+}
+
 var WsConn = make(map[uint][]*websocket.Conn)
-var WsNotifyQueue = make(map[uint][]byte)
 
 func CheckOnline(user models.Account) bool {
 	return len(WsConn[user.ID]) > 0

pkg/services/notifications.go

@@ -31,14 +31,19 @@ func NewNotification(notification models.Notification) error {
 
 	go func() {
 		err := PushNotification(notification)
-		log.Error().Err(err).Msg("Unexpected error occurred during the notification.")
+		if err != nil {
+			log.Error().Err(err).Msg("Unexpected error occurred during the notification.")
+		}
 	}()
 
 	return nil
 }
 
 func PushNotification(notification models.Notification) error {
-	WsNotifyQueue[notification.RecipientID], _ = jsoniter.Marshal(notification)
+	raw, _ := jsoniter.Marshal(notification)
+	for _, conn := range WsConn[notification.RecipientID] {
+		_ = conn.WriteMessage(1, raw)
+	}
 
 	var subscribers []models.NotificationSubscriber
 	if err := database.C.Where(&models.NotificationSubscriber{

pkg/views/src/components/NotificationList.vue

@@ -1,8 +1,8 @@
 <template>
   <v-menu eager :close-on-content-click="false">
     <template #activator="{ props }">
-      <v-btn v-bind="props" stacked rounded="circle" size="small" variant="text" :loading="loading">
+      <v-btn v-bind="props" icon size="small" variant="text" :loading="loading">
-        <v-badge v-if="pagination.total > 0" color="error" :content="pagination.total">
+        <v-badge v-if="notify.total > 0" color="error" :content="notify.total">
           <v-icon icon="mdi-bell" />
         </v-badge>
 
@@ -10,19 +10,19 @@
       </v-btn>
     </template>
 
-    <v-list v-if="notifications.length <= 0" class="w-[380px]" density="compact">
+    <v-list v-if="notify.notifications.length <= 0" class="w-[380px]" density="compact">
       <v-list-item>
         <v-alert class="text-sm" variant="tonal" type="info">You are done! There is no unread notifications for you.</v-alert>
       </v-list-item>
     </v-list>
 
     <v-list v-else class="w-[380px]" density="compact" lines="three">
-      <v-list-item v-for="item in notifications">
+      <v-list-item v-for="(item, idx) in notify.notifications">
         <template #title>{{ item.subject }}</template>
         <template #subtitle>{{ item.content }}</template>
 
         <template #append>
-          <v-btn icon="mdi-check" size="x-small" variant="text" :disabled="loading" @click="markAsRead(item)" />
+          <v-btn icon="mdi-check" size="x-small" variant="text" :disabled="loading" @click="markAsRead(item, idx)" />
         </template>
 
         <div class="flex text-xs gap-1">
@@ -39,39 +39,17 @@
 <script setup lang="ts">
 import { request } from "@/scripts/request"
 import { getAtk } from "@/stores/userinfo"
-import { reactive, ref } from "vue"
+import { computed, onMounted, onUnmounted, ref } from "vue";
+import { useNotifications } from "@/stores/notifications";
 
-const loading = ref(false)
+const notify = useNotifications()
 
 const error = ref<string | null>(null)
+const submitting = ref(false)
+const loading = computed(() => notify.loading || submitting.value)
 
-const notifications = ref<any[]>([])
+async function markAsRead(item: any, idx: number) {
-const pagination = reactive({ page: 1, pageSize: 25, total: 0 })
+  submitting.value = true
-
-async function readNotifications() {
-  loading.value = true
-  const res = await request(
-    "/api/notifications?" +
-      new URLSearchParams({
-        take: pagination.pageSize.toString(),
-        offset: ((pagination.page - 1) * pagination.pageSize).toString(),
-      }),
-    {
-      headers: { Authorization: `Bearer ${getAtk()}` },
-    },
-  )
-  if (res.status === 200) {
-    const data = await res.json()
-    notifications.value = data["data"]
-    pagination.total = data["count"]
-  }
-  loading.value = false
-}
-
-readNotifications()
-
-async function markAsRead(item: any) {
-  loading.value = true
   const res = await request(`/api/notifications/${item.id}/read`, {
     method: "PUT",
     headers: { Authorization: `Bearer ${getAtk()}` },
@@ -79,9 +57,14 @@ async function markAsRead(item: any) {
   if (res.status !== 200) {
     error.value = await res.text()
   } else {
-    await readNotifications()
+    notify.remove(idx)
     error.value = null
   }
-  loading.value = false
+  submitting.value = false
 }
+
+notify.list()
+
+onMounted(() => notify.connect())
+onUnmounted(() => notify.disconnect())
 </script>

pkg/views/src/stores/notifications.ts

@@ -0,0 +1,64 @@
+import { defineStore } from "pinia";
+import { ref } from "vue";
+import { checkLoggedIn, getAtk } from "@/stores/userinfo";
+import { request } from "@/scripts/request";
+
+export const useNotifications = defineStore("notifications", () => {
+  let socket: WebSocket;
+
+  const loading = ref(false);
+
+  const notifications = ref<any[]>([]);
+  const total = ref(0)
+
+  async function list() {
+    loading.value = true;
+    const res = await request(
+      "/api/notifications?" +
+      new URLSearchParams({
+        take: (25).toString(),
+        offset: (0).toString()
+      }),
+      {
+        headers: { Authorization: `Bearer ${getAtk()}` }
+      }
+    );
+    if (res.status === 200) {
+      const data = await res.json();
+      notifications.value = data["data"];
+      total.value = data["count"];
+    }
+    loading.value = false;
+  }
+
+  function remove(idx: number) {
+    notifications.value.splice(idx, 1)
+    total.value--;
+  }
+
+  async function connect() {
+    if (!(checkLoggedIn())) return;
+
+    const uri = `ws://${window.location.host}/api/notifications/listen`;
+
+    socket = new WebSocket(uri + `?tk=${getAtk() as string}`);
+
+    socket.addEventListener("open", (event) => {
+      console.log("[NOTIFICATIONS] The listen websocket has been established... ", event.type);
+    });
+    socket.addEventListener("close", (event) => {
+      console.warn("[NOTIFICATIONS] The listen websocket is disconnected... ", event.reason, event.code);
+    });
+    socket.addEventListener("message", (event) => {
+      const data = JSON.parse(event.data);
+      notifications.value.push(data);
+      total.value++;
+    });
+  }
+
+  function disconnect() {
+    socket.close();
+  }
+
+  return { loading, notifications, total, list, remove, connect, disconnect };
+});

pkg/views/vite.config.ts

@@ -1,22 +1,27 @@
-import { fileURLToPath, URL } from "node:url"
+import { fileURLToPath, URL } from "node:url";
 
-import { defineConfig } from "vite"
+import { defineConfig } from "vite";
-import vue from "@vitejs/plugin-vue"
+import vue from "@vitejs/plugin-vue";
-import vueJsx from "@vitejs/plugin-vue-jsx"
+import vueJsx from "@vitejs/plugin-vue-jsx";
-import unocss from "unocss/vite"
+import unocss from "unocss/vite";
 
 // https://vitejs.dev/config/
 export default defineConfig({
   plugins: [vue(), vueJsx(), unocss()],
   resolve: {
     alias: {
-      "@": fileURLToPath(new URL("./src", import.meta.url)),
+      "@": fileURLToPath(new URL("./src", import.meta.url))
-    },
+    }
   },
   server: {
     proxy: {
+      "/api/notifications/listen": {
+        target: "ws://localhost:8444",
+        ws: true
+      },
+
       "/api": "http://localhost:8444",
       "/.well-known": "http://localhost:8444"
     }
   }
-})
+});