feat: realtime change notifications

Author: Terbau

apps/rpc/src/bin/server.ts

@@ -59,10 +59,24 @@ controller.signal.addEventListener("abort", () => {
 })
 
 export async function createFastifyContext({ req }: CreateFastifyContextOptions) {
+  let rawToken: string | undefined
+
   const bearer = req.headers.authorization
   if (bearer !== undefined) {
-    const token = bearer.substring("Bearer ".length)
-    const principal = await dependencies.rpcJwtService.verify(token)
+    rawToken = bearer.substring("Bearer ".length)
+  } else if (typeof req.query === "object" && req.query !== null && "connectionParams" in req.query) {
+    try {
+      const params = JSON.parse(req.query.connectionParams as string)
+      if (typeof params.token === "string") {
+        rawToken = params.token
+      }
+    } catch {
+      // malformed connectionParams — treat as unauthenticated
+    }
+  }
+
+  if (rawToken !== undefined) {
+    const principal = await dependencies.rpcJwtService.verify(rawToken)
     const subject = principal.payload.sub
     if (subject === undefined) {
       return createTrpcContext(null, serviceLayer)

apps/rpc/src/modules/core.ts

@@ -189,7 +189,7 @@ export async function createServiceLayer(
   const feedbackFormAnswerRepository = getFeedbackFormAnswerRepository()
   const notificationRepository = getNotificationRepository()
 
-  const notificationService = getNotificationService(notificationRepository, userRepository, attendanceRepository)
+  const notificationService = getNotificationService(notificationRepository, userRepository, attendanceRepository, eventEmitter)
   const membershipService = getMembershipService()
   const emailService = isAmazonSesEmailFeatureEnabled(configuration)
     ? getEmailService(clients.sesClient, clients.sqsClient, configuration)

apps/rpc/src/modules/notification/notification-router.ts

@@ -1,6 +1,6 @@
+import { on } from "node:events"
 import type { inferProcedureInput, inferProcedureOutput } from "@trpc/server"
 import { z } from "zod"
-import { isEditor } from "../../authorization"
 import { withAuditLogEntry, withAuthentication, withAuthorization, withDatabaseTransaction } from "../../middlewares"
 import { procedure, t } from "../../trpc"
 import { BasePaginateInputSchema, PaginateInputSchema } from "@dotkomonline/utils"
@@ -11,6 +11,8 @@ import {
   UserNotificationDTOSchema,
   UserNotificationSchema,
 } from "./notification-types"
+import type { Notification } from "./notification-types"
+import { isCommitteeMember } from "src/authorization"
 
 export type GetNotificationInput = inferProcedureInput<typeof getNotificationProcedure>
 export type GetNotificationOutput = inferProcedureOutput<typeof getNotificationProcedure>
@@ -28,11 +30,20 @@ const createNotificationProcedure = procedure
   .input(NotificationWriteSchema)
   .output(NotificationDTOSchema)
   .use(withAuthentication())
-  .use(withAuthorization(isEditor()))
+  .use(withAuthorization(isCommitteeMember()))
   .use(withDatabaseTransaction())
   .use(withAuditLogEntry())
   .mutation(({ input, ctx }) => {
-    return ctx.notificationService.createWithRecipients(ctx.handle, input)
+    return ctx.notificationService.create(
+      ctx.handle,
+      input.recipientIds,
+      input.type,
+      input.title,
+      input.shortDescription ?? input.title,
+      input.actorGroupId,
+      input.payloadType,
+      input.payload
+    )
   })
 
 export type EditNotificationInput = inferProcedureInput<typeof editNotificationProcedure>
@@ -46,7 +57,8 @@ const editNotificationProcedure = procedure
   )
   .output(NotificationDTOSchema)
   .use(withAuthentication())
-  .use(withAuthorization(isEditor()))
+  .use(withAuthorization(isCommitteeMember()))
+
   .use(withDatabaseTransaction())
   .use(withAuditLogEntry())
   .mutation(async ({ input: changes, ctx }) => {
@@ -59,7 +71,8 @@ const deleteNotificationProcedure = procedure
   .input(NotificationSchema.shape.id)
   .output(z.boolean())
   .use(withAuthentication())
-  .use(withAuthorization(isEditor()))
+  .use(withAuthorization(isCommitteeMember()))
+
   .use(withDatabaseTransaction())
   .use(withAuditLogEntry())
   .mutation(async ({ input, ctx }) => {
@@ -118,7 +131,8 @@ const findNotificationsProcedure = procedure
   .input(PaginateInputSchema)
   .output(z.object({ items: z.array(NotificationDTOSchema), nextCursor: NotificationSchema.shape.id.optional() }))
   .use(withAuthentication())
-  .use(withAuthorization(isEditor()))
+  .use(withAuthorization(isCommitteeMember()))
+
   .use(withDatabaseTransaction())
   .query(async ({ input, ctx }) => {
     const items = await ctx.notificationService.findMany(ctx.handle, input)
@@ -129,6 +143,20 @@ const findNotificationsProcedure = procedure
     }
   })
 
+export type OnNewNotificationInput = inferProcedureInput<typeof onNewNotificationProcedure>
+export type OnNewNotificationOutput = inferProcedureOutput<typeof onNewNotificationProcedure>
+const onNewNotificationProcedure = procedure
+  .use(withAuthentication())
+  .subscription(async function* ({ ctx, signal }) {
+    for await (const [data] of on(ctx.eventEmitter, "notification:new", { signal })) {
+      const { userId, notification } = data as { userId: string; notification: Notification }
+      if (userId !== ctx.principal.subject) {
+        continue
+      }
+      yield notification
+    }
+  })
+
 export const notificationRouter = t.router({
   get: getNotificationProcedure,
   create: createNotificationProcedure,
@@ -139,4 +167,5 @@ export const notificationRouter = t.router({
   markAsRead: markAsReadProcedure,
   markAllAsRead: markAllAsReadProcedure,
   findMany: findNotificationsProcedure,
+  onNewNotification: onNewNotificationProcedure,
 })

apps/rpc/src/modules/notification/notification-service.ts

@@ -1,3 +1,4 @@
+import type EventEmitter from "node:events"
 import type { DBHandle } from "@dotkomonline/db"
 import type { UserId } from "@dotkomonline/types"
 import type { NotificationRepository } from "./notification-repository"
@@ -57,7 +58,8 @@ export interface NotificationService {
 export function getNotificationService(
   notificationRepository: NotificationRepository,
   userRepository: UserRepository,
-  attendanceRepository: AttendanceRepository
+  attendanceRepository: AttendanceRepository,
+  eventEmitter: EventEmitter
 ): NotificationService {
   return {
     async findById(handle, notificationId) {
@@ -69,7 +71,7 @@ export function getNotificationService(
     },
 
     async create(handle, recipientIds, notificationType, title, shortDescription, actorGroupId, payloadType, payload) {
-      return await notificationRepository.createWithRecipients(handle, {
+      const notification = await notificationRepository.createWithRecipients(handle, {
         title,
         shortDescription,
         content: shortDescription ?? title,
@@ -80,6 +82,10 @@ export function getNotificationService(
         taskId: null,
         recipientIds,
       })
+      for (const userId of recipientIds) {
+        eventEmitter.emit("notification:new", { userId, notification })
+      }
+      return notification
     },
 
     async retrieveIntendedRecipientIds(handle, notificationType, eventId) {

apps/web/src/components/Navbar/Notifications/NotificationDropdown.tsx

@@ -6,6 +6,7 @@ import { cn } from "@dotkomonline/ui"
 import { useTRPC } from "@/utils/trpc/client"
 import { useInfiniteQuery, useMutation, useQueryClient } from "@tanstack/react-query"
 import type { UserNotificationDTO } from "@dotkomonline/rpc"
+import { useSubscription } from "@trpc/tanstack-react-query"
 
 interface NotificationDropdownProps extends ComponentProps<typeof DropdownMenu.Root> {
   open?: boolean
@@ -16,6 +17,17 @@ export const NotificationDropdown = ({ open, amountUnread, ...props }: Notificat
   const trpc = useTRPC()
   const queryClient = useQueryClient()
 
+  useSubscription(
+    trpc.notification.onNewNotification.subscriptionOptions(undefined, {
+      onData: () => {
+        queryClient.invalidateQueries(trpc.notification.getUnreadCount.queryOptions())
+        queryClient.invalidateQueries({
+          queryKey: trpc.notification.getMyNotifications.infiniteQueryOptions({ take: 10 }).queryKey,
+        })
+      },
+    })
+  )
+
   const { data, fetchNextPage, hasNextPage, isFetchingNextPage, isLoading } = useInfiniteQuery({
     ...trpc.notification.getMyNotifications.infiniteQueryOptions({ take: 10 }),
     enabled: open,
@@ -109,7 +121,7 @@ export const NotificationDropdown = ({ open, amountUnread, ...props }: Notificat
             "mx-4 xs:ml-4 xs:w-102 xs:-mr-16 lg:-mr-4",
             "rounded-3xl shadow-sm",
             "bg-blue-50 border border-blue-100",
-            "dark:border-white/10 dark:bg-stone-700"
+            "dark:border-white/10 dark:bg-stone-800"
           )}
         >
           <div className="p-5 border-b border-black/10 dark:border-white/10">

apps/web/src/utils/trpc/QueryProvider.tsx

@@ -1,7 +1,7 @@
 "use client"
 
 import { env } from "@/env"
-import { getAccessToken } from "@auth0/nextjs-auth0"
+import { getAccessToken, useUser } from "@auth0/nextjs-auth0"
 import type { AppRouter } from "@dotkomonline/rpc"
 import { QueryClient, QueryClientProvider } from "@tanstack/react-query"
 import {
@@ -45,11 +45,16 @@ export const useTRPCSSERegisterChangeConnectionState = () => {
 }
 
 export const QueryProvider = ({ children }: PropsWithChildren) => {
+  const { user } = useUser()
+  const userId = user?.sub
+
   const [trpcSSERegisterChangeConnectionState, setTRPCSSERegisterChangeConnectionState] =
     useState<TRPCSSEConnectionState>("connecting")
 
-  const trpcConfig: CreateTRPCClientOptions<AppRouter> = useMemo(
-    () => ({
+  const trpcConfig: CreateTRPCClientOptions<AppRouter> = useMemo(() => {
+    const hasAuthenticatedUser = typeof userId === "string" && userId !== ""
+
+    return {
       links: [
         loggerLink({
           enabled: (opts) => opts.direction === "down" && opts.result instanceof Error,
@@ -59,6 +64,23 @@ export const QueryProvider = ({ children }: PropsWithChildren) => {
           true: httpSubscriptionLink({
             transformer: superjson,
             url: `${env.NEXT_PUBLIC_RPC_HOST}/api/trpc`,
+            async connectionParams() {
+              if (!hasAuthenticatedUser) {
+                return {}
+              }
+
+              try {
+                const token = await getAccessToken()
+
+                if (typeof token === "string" && token !== "") {
+                  return { token }
+                }
+              } catch {
+                // not authenticated
+              }
+
+              return {}
+            },
           }),
           false: httpBatchLink({
             transformer: superjson,
@@ -85,9 +107,8 @@ export const QueryProvider = ({ children }: PropsWithChildren) => {
           }),
         }),
       ],
-    }),
-    []
-  )
+    }
+  }, [userId])
 
   const trpcClient = useMemo(() => createTRPCClient(trpcConfig), [trpcConfig])