improve(api): update api to fastify v5

apps/api/src/controllers/event.controller.ts

@@ -4,7 +4,7 @@ import type { FastifyReply, FastifyRequest } from 'fastify';
 import { generateDeviceId } from '@openpanel/common/server';
 import { getSalts } from '@openpanel/db';
 import { eventsQueue } from '@openpanel/queue';
-import { getRedisCache } from '@openpanel/redis';
+import { getLock } from '@openpanel/redis';
 import type { PostEventPayload } from '@openpanel/sdk';
 
 import { getStringHeaders, getTimestamp } from './track.controller';
@@ -42,12 +42,10 @@ export async function postEvent(
   const isScreenView = request.body.name === 'screen_view';
   // this will ensure that we don't have multiple events creating sessions
   const LOCK_DURATION = 1000;
-  const locked = await getRedisCache().set(
+  const locked = await getLock(
     `request:priority:${currentDeviceId}-${previousDeviceId}:${isScreenView ? 'screen_view' : 'other'}`,
     'locked',
-    'PX',
     LOCK_DURATION,
-    'NX',
   );
 
   await eventsQueue.add(
@@ -65,7 +63,7 @@ export async function postEvent(
         geo,
         currentDeviceId,
         previousDeviceId,
-        priority: locked === 'OK',
+        priority: locked,
       },
     },
     {

apps/api/src/controllers/live.controller.ts

@@ -1,8 +1,7 @@
-import type { SocketStream } from '@fastify/websocket';
 import type { FastifyRequest } from 'fastify';
 import superjson from 'superjson';
-import type { WebSocket } from 'ws';
 
+import type { WebSocket } from '@fastify/websocket';
 import {
   eventBuffer,
   getProfileByIdCached,
@@ -16,20 +15,12 @@ import {
 import { getProjectAccess } from '@openpanel/trpc';
 import { getOrganizationAccess } from '@openpanel/trpc/src/access';
 
-type WebSocketConnection = SocketStream & {
-  socket: WebSocket & {
-    on(event: 'close', listener: () => void): void;
-    send(data: string): void;
-    close(): void;
-  };
-};
-
 export function getLiveEventInfo(key: string) {
   return key.split(':').slice(2) as [string, string];
 }
 
 export function wsVisitors(
-  connection: WebSocketConnection,
+  socket: WebSocket,
   req: FastifyRequest<{
     Params: {
       projectId: string;
@@ -37,11 +28,10 @@ export function wsVisitors(
   }>,
 ) {
   const { params } = req;
-
   const unsubscribe = subscribeToPublishedEvent('events', 'saved', (event) => {
     if (event?.projectId === params.projectId) {
       eventBuffer.getActiveVisitorCount(params.projectId).then((count) => {
-        connection.socket.send(String(count));
+        socket.send(String(count));
       });
     }
   });
@@ -52,20 +42,20 @@ export function wsVisitors(
       const [projectId] = getLiveEventInfo(key);
       if (projectId && projectId === params.projectId) {
         eventBuffer.getActiveVisitorCount(params.projectId).then((count) => {
-          connection.socket.send(String(count));
+          socket.send(String(count));
         });
       }
     },
   );
 
-  connection.socket.on('close', () => {
+  socket.on('close', () => {
     unsubscribe();
     punsubscribe();
   });
 }
 
 export async function wsProjectEvents(
-  connection: WebSocketConnection,
+  socket: WebSocket,
   req: FastifyRequest<{
     Params: {
       projectId: string;
@@ -80,15 +70,15 @@ export async function wsProjectEvents(
   const type = query.type || 'saved';
 
   if (!['saved', 'received'].includes(type)) {
-    connection.socket.send('Invalid type');
-    connection.socket.close();
+    socket.send('Invalid type');
+    socket.close();
     return;
   }
 
   const userId = req.session?.userId;
   if (!userId) {
-    connection.socket.send('No active session');
-    connection.socket.close();
+    socket.send('No active session');
+    socket.close();
     return;
   }
 
@@ -106,7 +96,7 @@ export async function wsProjectEvents(
           event.profileId,
           event.projectId,
         );
-        connection.socket.send(
+        socket.send(
           superjson.stringify(
             access
               ? {
@@ -120,11 +110,11 @@ export async function wsProjectEvents(
     },
   );
 
-  connection.socket.on('close', () => unsubscribe());
+  socket.on('close', () => unsubscribe());
 }
 
 export async function wsProjectNotifications(
-  connection: WebSocketConnection,
+  socket: WebSocket,
   req: FastifyRequest<{
     Params: {
       projectId: string;
@@ -135,8 +125,8 @@ export async function wsProjectNotifications(
   const userId = req.session?.userId;
 
   if (!userId) {
-    connection.socket.send('No active session');
-    connection.socket.close();
+    socket.send('No active session');
+    socket.close();
     return;
   }
 
@@ -146,8 +136,8 @@ export async function wsProjectNotifications(
   });
 
   if (!access) {
-    connection.socket.send('No access');
-    connection.socket.close();
+    socket.send('No access');
+    socket.close();
     return;
   }
 
@@ -156,16 +146,16 @@ export async function wsProjectNotifications(
     'created',
     (notification) => {
       if (notification.projectId === params.projectId) {
-        connection.socket.send(superjson.stringify(notification));
+        socket.send(superjson.stringify(notification));
       }
     },
   );
 
-  connection.socket.on('close', () => unsubscribe());
+  socket.on('close', () => unsubscribe());
 }
 
 export async function wsOrganizationEvents(
-  connection: WebSocketConnection,
+  socket: WebSocket,
   req: FastifyRequest<{
     Params: {
       organizationId: string;
@@ -176,8 +166,8 @@ export async function wsOrganizationEvents(
   const userId = req.session?.userId;
 
   if (!userId) {
-    connection.socket.send('No active session');
-    connection.socket.close();
+    socket.send('No active session');
+    socket.close();
     return;
   }
 
@@ -187,8 +177,8 @@ export async function wsOrganizationEvents(
   });
 
   if (!access) {
-    connection.socket.send('No access');
-    connection.socket.close();
+    socket.send('No access');
+    socket.close();
     return;
   }
 
@@ -196,9 +186,9 @@ export async function wsOrganizationEvents(
     'organization',
     'subscription_updated',
     (message) => {
-      connection.socket.send(setSuperJson(message));
+      socket.send(setSuperJson(message));
     },
   );
 
-  connection.socket.on('close', () => unsubscribe());
+  socket.on('close', () => unsubscribe());
 }

apps/api/src/controllers/oauth-callback.controller.tsx

@@ -76,7 +76,7 @@ async function handleExistingUser({
     sessionToken,
     session.expiresAt,
   );
-  return reply.status(302).redirect(process.env.NEXT_PUBLIC_DASHBOARD_URL!);
+  return reply.redirect(process.env.NEXT_PUBLIC_DASHBOARD_URL!);
 }
 
 async function handleNewUser({
@@ -138,7 +138,7 @@ async function handleNewUser({
     sessionToken,
     session.expiresAt,
   );
-  return reply.status(302).redirect(process.env.NEXT_PUBLIC_DASHBOARD_URL!);
+  return reply.redirect(process.env.NEXT_PUBLIC_DASHBOARD_URL!);
 }
 
 // Provider-specific user fetching

apps/api/src/controllers/track.controller.ts

@@ -6,7 +6,7 @@ import { path, assocPath, pathOr, pick } from 'ramda';
 import { generateDeviceId, parseUserAgent } from '@openpanel/common/server';
 import { getProfileById, getSalts, upsertProfile } from '@openpanel/db';
 import { eventsQueue } from '@openpanel/queue';
-import { getRedisCache } from '@openpanel/redis';
+import { getLock } from '@openpanel/redis';
 import type {
   DecrementPayload,
   IdentifyPayload,
@@ -230,12 +230,10 @@ async function track({
   const isScreenView = payload.name === 'screen_view';
   // this will ensure that we don't have multiple events creating sessions
   const LOCK_DURATION = 1000;
-  const locked = await getRedisCache().set(
+  const locked = await getLock(
     `request:priority:${currentDeviceId}-${previousDeviceId}:${isScreenView ? 'screen_view' : 'other'}`,
     'locked',
-    'PX',
     LOCK_DURATION,
-    'NX',
   );
 
   await eventsQueue.add(
@@ -253,7 +251,7 @@ async function track({
         geo,
         currentDeviceId,
         previousDeviceId,
-        priority: locked === 'OK',
+        priority: locked,
       },
     },
     {