diff --git a/apps/api/package.json b/apps/api/package.json
index 3cb8b99f5..5f63bc5d9 100644
--- a/apps/api/package.json
+++ b/apps/api/package.json
@@ -30,7 +30,6 @@
     "@openpanel/logger": "workspace:*",
     "@openpanel/payments": "workspace:*",
     "@openpanel/queue": "workspace:*",
-    "groupmq": "catalog:",
     "@openpanel/redis": "workspace:*",
     "@openpanel/trpc": "workspace:*",
     "@openpanel/validation": "workspace:*",
@@ -40,6 +39,7 @@
     "fastify": "^5.6.1",
     "fastify-metrics": "^12.1.0",
     "fastify-raw-body": "^5.0.0",
+    "groupmq": "catalog:",
     "jsonwebtoken": "^9.0.2",
     "ramda": "^0.29.1",
     "sharp": "^0.33.5",
diff --git a/apps/api/src/bots/index.ts b/apps/api/src/bots/index.ts
index 12d6b7387..391aa2900 100644
--- a/apps/api/src/bots/index.ts
+++ b/apps/api/src/bots/index.ts
@@ -1,4 +1,4 @@
-import { cacheable, cacheableLru } from '@openpanel/redis';
+import { cacheable } from '@openpanel/redis';
 import bots from './bots';
 
 // Pre-compile regex patterns at module load time
@@ -15,7 +15,7 @@ const compiledBots = bots.map((bot) => {
 const regexBots = compiledBots.filter((bot) => 'compiledRegex' in bot);
 const includesBots = compiledBots.filter((bot) => 'includes' in bot);
 
-export const isBot = cacheableLru(
+export const isBot = cacheable(
   'is-bot',
   (ua: string) => {
     // Check simple string patterns first (fast)
@@ -40,8 +40,5 @@ export const isBot = cacheableLru(
 
     return null;
   },
-  {
-    maxSize: 1000,
-    ttl: 60 * 5,
-  },
+  60 * 5
 );
diff --git a/apps/api/src/controllers/live.controller.ts b/apps/api/src/controllers/live.controller.ts
index 29931aa84..488e67138 100644
--- a/apps/api/src/controllers/live.controller.ts
+++ b/apps/api/src/controllers/live.controller.ts
@@ -1,12 +1,5 @@
-import type { FastifyRequest } from 'fastify';
-import superjson from 'superjson';
-
 import type { WebSocket } from '@fastify/websocket';
-import {
-  eventBuffer,
-  getProfileById,
-  transformMinimalEvent,
-} from '@openpanel/db';
+import { eventBuffer } from '@openpanel/db';
 import { setSuperJson } from '@openpanel/json';
 import {
   psubscribeToPublishedEvent,
@@ -14,10 +7,7 @@ import {
 } from '@openpanel/redis';
 import { getProjectAccess } from '@openpanel/trpc';
 import { getOrganizationAccess } from '@openpanel/trpc/src/access';
-
-export function getLiveEventInfo(key: string) {
-  return key.split(':').slice(2) as [string, string];
-}
+import type { FastifyRequest } from 'fastify';
 
 export function wsVisitors(
   socket: WebSocket,
@@ -25,27 +15,38 @@ export function wsVisitors(
     Params: {
       projectId: string;
     };
-  }>,
+  }>
 ) {
   const { params } = req;
-  const unsubscribe = subscribeToPublishedEvent('events', 'saved', (event) => {
-    if (event?.projectId === params.projectId) {
-      eventBuffer.getActiveVisitorCount(params.projectId).then((count) => {
+  const sendCount = () => {
+    eventBuffer
+      .getActiveVisitorCount(params.projectId)
+      .then((count) => {
         socket.send(String(count));
+      })
+      .catch(() => {
+        socket.send('0');
       });
+  };
+
+  const unsubscribe = subscribeToPublishedEvent(
+    'events',
+    'batch',
+    ({ projectId }) => {
+      if (projectId === params.projectId) {
+        sendCount();
+      }
     }
-  });
+  );
 
   const punsubscribe = psubscribeToPublishedEvent(
     '__keyevent@0__:expired',
     (key) => {
-      const [projectId] = getLiveEventInfo(key);
-      if (projectId && projectId === params.projectId) {
-        eventBuffer.getActiveVisitorCount(params.projectId).then((count) => {
-          socket.send(String(count));
-        });
+      const [, , projectId] = key.split(':');
+      if (projectId === params.projectId) {
+        sendCount();
       }
-    },
+    }
   );
 
   socket.on('close', () => {
@@ -62,18 +63,10 @@ export async function wsProjectEvents(
     };
     Querystring: {
       token?: string;
-      type?: 'saved' | 'received';
     };
-  }>,
+  }>
 ) {
-  const { params, query } = req;
-  const type = query.type || 'saved';
-
-  if (!['saved', 'received'].includes(type)) {
-    socket.send('Invalid type');
-    socket.close();
-    return;
-  }
+  const { params } = req;
 
   const userId = req.session?.userId;
   if (!userId) {
@@ -87,24 +80,20 @@ export async function wsProjectEvents(
     projectId: params.projectId,
   });
 
+  if (!access) {
+    socket.send('No access');
+    socket.close();
+    return;
+  }
+
   const unsubscribe = subscribeToPublishedEvent(
     'events',
-    type,
-    async (event) => {
-      if (event.projectId === params.projectId) {
-        const profile = await getProfileById(event.profileId, event.projectId);
-        socket.send(
-          superjson.stringify(
-            access
-              ? {
-                  ...event,
-                  profile,
-                }
-              : transformMinimalEvent(event),
-          ),
-        );
+    'batch',
+    ({ projectId, count }) => {
+      if (projectId === params.projectId) {
+        socket.send(setSuperJson({ count }));
       }
-    },
+    }
   );
 
   socket.on('close', () => unsubscribe());
@@ -116,7 +105,7 @@ export async function wsProjectNotifications(
     Params: {
       projectId: string;
     };
-  }>,
+  }>
 ) {
   const { params } = req;
   const userId = req.session?.userId;
@@ -143,9 +132,9 @@ export async function wsProjectNotifications(
     'created',
     (notification) => {
       if (notification.projectId === params.projectId) {
-        socket.send(superjson.stringify(notification));
+        socket.send(setSuperJson(notification));
       }
-    },
+    }
   );
 
   socket.on('close', () => unsubscribe());
@@ -157,7 +146,7 @@ export async function wsOrganizationEvents(
     Params: {
       organizationId: string;
     };
-  }>,
+  }>
 ) {
   const { params } = req;
   const userId = req.session?.userId;
@@ -184,7 +173,7 @@ export async function wsOrganizationEvents(
     'subscription_updated',
     (message) => {
       socket.send(setSuperJson(message));
-    },
+    }
   );
 
   socket.on('close', () => unsubscribe());
diff --git a/apps/api/src/controllers/manage.controller.ts b/apps/api/src/controllers/manage.controller.ts
index e1d3bf67f..1d162851a 100644
--- a/apps/api/src/controllers/manage.controller.ts
+++ b/apps/api/src/controllers/manage.controller.ts
@@ -1,5 +1,4 @@
 import crypto from 'node:crypto';
-import { HttpError } from '@/utils/errors';
 import { stripTrailingSlash } from '@openpanel/common';
 import { hashPassword } from '@openpanel/common/server';
 import {
@@ -10,6 +9,7 @@ import {
 } from '@openpanel/db';
 import type { FastifyReply, FastifyRequest } from 'fastify';
 import { z } from 'zod';
+import { HttpError } from '@/utils/errors';
 
 // Validation schemas
 const zCreateProject = z.object({
@@ -57,7 +57,7 @@ const zUpdateReference = z.object({
 // Projects CRUD
 export async function listProjects(
   request: FastifyRequest,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const projects = await db.project.findMany({
     where: {
@@ -74,7 +74,7 @@ export async function listProjects(
 
 export async function getProject(
   request: FastifyRequest<{ Params: { id: string } }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const project = await db.project.findFirst({
     where: {
@@ -92,7 +92,7 @@ export async function getProject(
 
 export async function createProject(
   request: FastifyRequest<{ Body: z.infer<typeof zCreateProject> }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const parsed = zCreateProject.safeParse(request.body);
 
@@ -139,12 +139,9 @@ export async function createProject(
     },
   });
 
-  // Clear cache
   await Promise.all([
     getProjectByIdCached.clear(project.id),
-    project.clients.map((client) => {
-      getClientByIdCached.clear(client.id);
-    }),
+    ...project.clients.map((client) => getClientByIdCached.clear(client.id)),
   ]);
 
   reply.send({
@@ -165,7 +162,7 @@ export async function updateProject(
     Params: { id: string };
     Body: z.infer<typeof zUpdateProject>;
   }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const parsed = zUpdateProject.safeParse(request.body);
 
@@ -223,12 +220,9 @@ export async function updateProject(
     data: updateData,
   });
 
-  // Clear cache
   await Promise.all([
     getProjectByIdCached.clear(project.id),
-    existing.clients.map((client) => {
-      getClientByIdCached.clear(client.id);
-    }),
+    ...existing.clients.map((client) => getClientByIdCached.clear(client.id)),
   ]);
 
   reply.send({ data: project });
@@ -236,7 +230,7 @@ export async function updateProject(
 
 export async function deleteProject(
   request: FastifyRequest<{ Params: { id: string } }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const project = await db.project.findFirst({
     where: {
@@ -266,7 +260,7 @@ export async function deleteProject(
 // Clients CRUD
 export async function listClients(
   request: FastifyRequest<{ Querystring: { projectId?: string } }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const where: any = {
     organizationId: request.client!.organizationId,
@@ -300,7 +294,7 @@ export async function listClients(
 
 export async function getClient(
   request: FastifyRequest<{ Params: { id: string } }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const client = await db.client.findFirst({
     where: {
@@ -318,7 +312,7 @@ export async function getClient(
 
 export async function createClient(
   request: FastifyRequest<{ Body: z.infer<typeof zCreateClient> }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const parsed = zCreateClient.safeParse(request.body);
 
@@ -374,7 +368,7 @@ export async function updateClient(
     Params: { id: string };
     Body: z.infer<typeof zUpdateClient>;
   }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const parsed = zUpdateClient.safeParse(request.body);
 
@@ -417,7 +411,7 @@ export async function updateClient(
 
 export async function deleteClient(
   request: FastifyRequest<{ Params: { id: string } }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const client = await db.client.findFirst({
     where: {
@@ -444,7 +438,7 @@ export async function deleteClient(
 // References CRUD
 export async function listReferences(
   request: FastifyRequest<{ Querystring: { projectId?: string } }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const where: any = {};
 
@@ -488,7 +482,7 @@ export async function listReferences(
 
 export async function getReference(
   request: FastifyRequest<{ Params: { id: string } }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const reference = await db.reference.findUnique({
     where: {
@@ -516,7 +510,7 @@ export async function getReference(
 
 export async function createReference(
   request: FastifyRequest<{ Body: z.infer<typeof zCreateReference> }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const parsed = zCreateReference.safeParse(request.body);
 
@@ -559,7 +553,7 @@ export async function updateReference(
     Params: { id: string };
     Body: z.infer<typeof zUpdateReference>;
   }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const parsed = zUpdateReference.safeParse(request.body);
 
@@ -616,7 +610,7 @@ export async function updateReference(
 
 export async function deleteReference(
   request: FastifyRequest<{ Params: { id: string } }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const reference = await db.reference.findUnique({
     where: {
diff --git a/apps/api/src/controllers/track.controller.ts b/apps/api/src/controllers/track.controller.ts
index f7a3c61f0..0e530f933 100644
--- a/apps/api/src/controllers/track.controller.ts
+++ b/apps/api/src/controllers/track.controller.ts
@@ -7,7 +7,10 @@ import {
   upsertProfile,
 } from '@openpanel/db';
 import { type GeoLocation, getGeoLocation } from '@openpanel/geo';
-import { getEventsGroupQueueShard } from '@openpanel/queue';
+import {
+  type EventsQueuePayloadIncomingEvent,
+  getEventsGroupQueueShard,
+} from '@openpanel/queue';
 import { getRedisCache } from '@openpanel/redis';
 import {
   type IDecrementPayload,
@@ -112,6 +115,7 @@ interface TrackContext {
   identity?: IIdentifyPayload;
   deviceId: string;
   sessionId: string;
+  session?: EventsQueuePayloadIncomingEvent['payload']['session'];
   geo: GeoLocation;
 }
 
@@ -141,19 +145,21 @@ async function buildContext(
     validatedBody.payload.profileId = profileId;
   }
 
+  const overrideDeviceId =
+    validatedBody.type === 'track' &&
+    typeof validatedBody.payload?.properties?.__deviceId === 'string'
+      ? validatedBody.payload?.properties.__deviceId
+      : undefined;
+
   // Get geo location (needed for track and identify)
   const [geo, salts] = await Promise.all([getGeoLocation(ip), getSalts()]);
 
-  const { deviceId, sessionId } = await getDeviceId({
+  const deviceIdResult = await getDeviceId({
     projectId,
     ip,
     ua,
     salts,
-    overrideDeviceId:
-      validatedBody.type === 'track' &&
-      typeof validatedBody.payload?.properties?.__deviceId === 'string'
-        ? validatedBody.payload?.properties.__deviceId
-        : undefined,
+    overrideDeviceId,
   });
 
   return {
@@ -166,8 +172,9 @@ async function buildContext(
       isFromPast: timestamp.isTimestampFromThePast,
     },
     identity,
-    deviceId,
-    sessionId,
+    deviceId: deviceIdResult.deviceId,
+    sessionId: deviceIdResult.sessionId,
+    session: deviceIdResult.session,
     geo,
   };
 }
@@ -176,13 +183,14 @@ async function handleTrack(
   payload: ITrackPayload,
   context: TrackContext
 ): Promise<void> {
-  const { projectId, deviceId, geo, headers, timestamp, sessionId } = context;
+  const { projectId, deviceId, geo, headers, timestamp, sessionId, session } =
+    context;
 
   const uaInfo = parseUserAgent(headers['user-agent'], payload.properties);
   const groupId = uaInfo.isServer
     ? payload.profileId
       ? `${projectId}:${payload.profileId}`
-      : `${projectId}:${generateId()}`
+      : undefined
     : deviceId;
   const jobId = [
     slug(payload.name),
@@ -203,7 +211,7 @@ async function handleTrack(
   }
 
   promises.push(
-    getEventsGroupQueueShard(groupId).add({
+    getEventsGroupQueueShard(groupId || generateId()).add({
       orderMs: timestamp.value,
       data: {
         projectId,
@@ -217,6 +225,7 @@ async function handleTrack(
         geo,
         deviceId,
         sessionId,
+        session,
       },
       groupId,
       jobId,
diff --git a/apps/api/src/hooks/is-bot.hook.ts b/apps/api/src/hooks/is-bot.hook.ts
index c01a231a6..09346fbde 100644
--- a/apps/api/src/hooks/is-bot.hook.ts
+++ b/apps/api/src/hooks/is-bot.hook.ts
@@ -1,20 +1,19 @@
-import { isBot } from '@/bots';
 import { createBotEvent } from '@openpanel/db';
 import type {
   DeprecatedPostEventPayload,
   ITrackHandlerPayload,
 } from '@openpanel/validation';
-
 import type { FastifyReply, FastifyRequest } from 'fastify';
+import { isBot } from '@/bots';
 
 export async function isBotHook(
   req: FastifyRequest<{
     Body: ITrackHandlerPayload | DeprecatedPostEventPayload;
   }>,
-  reply: FastifyReply,
+  reply: FastifyReply
 ) {
   const bot = req.headers['user-agent']
-    ? isBot(req.headers['user-agent'])
+    ? await isBot(req.headers['user-agent'])
     : null;
 
   if (bot && req.client?.projectId) {
@@ -44,6 +43,6 @@ export async function isBotHook(
       }
     }
 
-    return reply.status(202).send();
+    return reply.status(202).send({ bot });
   }
 }
diff --git a/apps/api/src/routes/track.router.ts b/apps/api/src/routes/track.router.ts
index 2f94a8eb4..1bb04c4b9 100644
--- a/apps/api/src/routes/track.router.ts
+++ b/apps/api/src/routes/track.router.ts
@@ -1,6 +1,5 @@
-import { fetchDeviceId, handler } from '@/controllers/track.controller';
 import type { FastifyPluginCallback } from 'fastify';
-
+import { fetchDeviceId, handler } from '@/controllers/track.controller';
 import { clientHook } from '@/hooks/client.hook';
 import { duplicateHook } from '@/hooks/duplicate.hook';
 import { isBotHook } from '@/hooks/is-bot.hook';
@@ -13,7 +12,7 @@ const trackRouter: FastifyPluginCallback = async (fastify) => {
   fastify.route({
     method: 'POST',
     url: '/',
-    handler: handler,
+    handler,
   });
 
   fastify.route({
diff --git a/apps/api/src/utils/ids.ts b/apps/api/src/utils/ids.ts
index db6b9aa5e..f2bd9a453 100644
--- a/apps/api/src/utils/ids.ts
+++ b/apps/api/src/utils/ids.ts
@@ -1,7 +1,12 @@
 import crypto from 'node:crypto';
 import { generateDeviceId } from '@openpanel/common/server';
 import { getSafeJson } from '@openpanel/json';
+import type {
+  EventsQueuePayloadCreateSessionEnd,
+  EventsQueuePayloadIncomingEvent,
+} from '@openpanel/queue';
 import { getRedisCache } from '@openpanel/redis';
+import { pick } from 'ramda';
 
 export async function getDeviceId({
   projectId,
@@ -37,14 +42,20 @@ export async function getDeviceId({
     ua,
   });
 
-  return await getDeviceIdFromSession({
+  return await getInfoFromSession({
     projectId,
     currentDeviceId,
     previousDeviceId,
   });
 }
 
-async function getDeviceIdFromSession({
+interface DeviceIdResult {
+  deviceId: string;
+  sessionId: string;
+  session?: EventsQueuePayloadIncomingEvent['payload']['session'];
+}
+
+async function getInfoFromSession({
   projectId,
   currentDeviceId,
   previousDeviceId,
@@ -52,7 +63,7 @@ async function getDeviceIdFromSession({
   projectId: string;
   currentDeviceId: string;
   previousDeviceId: string;
-}) {
+}): Promise<DeviceIdResult> {
   try {
     const multi = getRedisCache().multi();
     multi.hget(
@@ -65,21 +76,33 @@ async function getDeviceIdFromSession({
     );
     const res = await multi.exec();
     if (res?.[0]?.[1]) {
-      const data = getSafeJson<{ payload: { sessionId: string } }>(
+      const data = getSafeJson<EventsQueuePayloadCreateSessionEnd>(
         (res?.[0]?.[1] as string) ?? ''
       );
       if (data) {
-        const sessionId = data.payload.sessionId;
-        return { deviceId: currentDeviceId, sessionId };
+        return {
+          deviceId: currentDeviceId,
+          sessionId: data.payload.sessionId,
+          session: pick(
+            ['referrer', 'referrerName', 'referrerType'],
+            data.payload
+          ),
+        };
       }
     }
     if (res?.[1]?.[1]) {
-      const data = getSafeJson<{ payload: { sessionId: string } }>(
+      const data = getSafeJson<EventsQueuePayloadCreateSessionEnd>(
         (res?.[1]?.[1] as string) ?? ''
       );
       if (data) {
-        const sessionId = data.payload.sessionId;
-        return { deviceId: previousDeviceId, sessionId };
+        return {
+          deviceId: previousDeviceId,
+          sessionId: data.payload.sessionId,
+          session: pick(
+            ['referrer', 'referrerName', 'referrerType'],
+            data.payload
+          ),
+        };
       }
     }
   } catch (error) {
diff --git a/apps/start/src/components/events/event-listener.tsx b/apps/start/src/components/events/event-listener.tsx
index defabb7d1..08df9b61d 100644
--- a/apps/start/src/components/events/event-listener.tsx
+++ b/apps/start/src/components/events/event-listener.tsx
@@ -1,3 +1,4 @@
+import { AnimatedNumber } from '../animated-number';
 import {
   Tooltip,
   TooltipContent,
@@ -8,71 +9,53 @@ import { useDebounceState } from '@/hooks/use-debounce-state';
 import useWS from '@/hooks/use-ws';
 import { cn } from '@/utils/cn';
 
-import type { IServiceEvent, IServiceEventMinimal } from '@openpanel/db';
-import { useParams } from '@tanstack/react-router';
-import { AnimatedNumber } from '../animated-number';
-
 export default function EventListener({
   onRefresh,
 }: {
   onRefresh: () => void;
 }) {
-  const params = useParams({
-    strict: false,
-  });
   const { projectId } = useAppParams();
   const counter = useDebounceState(0, 1000);
-  useWS<IServiceEventMinimal | IServiceEvent>(
+  useWS<{ count: number }>(
     `/live/events/${projectId}`,
-    (event) => {
-      if (event) {
-        const isProfilePage = !!params?.profileId;
-        if (isProfilePage) {
-          const profile = 'profile' in event ? event.profile : null;
-          if (profile?.id === params?.profileId) {
-            counter.set((prev) => prev + 1);
-          }
-          return;
-        }
-
-        counter.set((prev) => prev + 1);
-      }
+    ({ count }) => {
+      counter.set((prev) => prev + count);
     },
     {
       debounce: {
         delay: 1000,
         maxWait: 5000,
       },
-    },
+    }
   );
 
   return (
     <Tooltip>
       <TooltipTrigger asChild>
         <button
-          type="button"
+          className="flex h-8 items-center gap-2 rounded-md border border-border bg-card px-3 font-medium leading-none"
           onClick={() => {
             counter.set(0);
             onRefresh();
           }}
-          className="flex h-8 items-center gap-2 rounded-md border border-border bg-card px-3 font-medium leading-none"
+          type="button"
         >
           <div className="relative">
             <div
               className={cn(
-                'h-3 w-3 animate-ping rounded-full bg-emerald-500 opacity-100 transition-all',
+                'h-3 w-3 animate-ping rounded-full bg-emerald-500 opacity-100 transition-all'
               )}
             />
             <div
               className={cn(
-                'absolute left-0 top-0 h-3 w-3 rounded-full bg-emerald-500 transition-all',
+                'absolute top-0 left-0 h-3 w-3 rounded-full bg-emerald-500 transition-all'
               )}
             />
           </div>
           {counter.debounced === 0 ? (
             'Listening'
           ) : (
-            <AnimatedNumber value={counter.debounced} suffix=" new events" />
+            <AnimatedNumber suffix=" new events" value={counter.debounced} />
           )}
         </button>
       </TooltipTrigger>
diff --git a/apps/start/src/components/events/table/index.tsx b/apps/start/src/components/events/table/index.tsx
index 300713997..95ed8e2d7 100644
--- a/apps/start/src/components/events/table/index.tsx
+++ b/apps/start/src/components/events/table/index.tsx
@@ -35,6 +35,7 @@ type Props = {
     >,
     unknown
   >;
+  showEventListener?: boolean;
 };
 
 const LOADING_DATA = [{}, {}, {}, {}, {}, {}, {}, {}, {}] as IServiceEvent[];
@@ -215,7 +216,7 @@ const VirtualizedEventsTable = ({
   );
 };
 
-export const EventsTable = ({ query }: Props) => {
+export const EventsTable = ({ query, showEventListener = false }: Props) => {
   const { isLoading } = query;
   const columns = useColumns();
 
@@ -272,7 +273,7 @@ export const EventsTable = ({ query }: Props) => {
 
   return (
     <>
-      <EventsTableToolbar query={query} table={table} />
+      <EventsTableToolbar query={query} table={table} showEventListener={showEventListener} />
       <VirtualizedEventsTable table={table} data={data} isLoading={isLoading} />
       <div className="w-full h-10 center-center pt-4" ref={inViewportRef}>
         <div
@@ -291,9 +292,11 @@ export const EventsTable = ({ query }: Props) => {
 function EventsTableToolbar({
   query,
   table,
+  showEventListener,
 }: {
   query: Props['query'];
   table: Table<IServiceEvent>;
+  showEventListener: boolean;
 }) {
   const { projectId } = useAppParams();
   const [startDate, setStartDate] = useQueryState(
@@ -305,7 +308,7 @@ function EventsTableToolbar({
   return (
     <DataTableToolbarContainer>
       <div className="flex flex-1 flex-wrap items-center gap-2">
-        <EventListener onRefresh={() => query.refetch()} />
+        {showEventListener && <EventListener onRefresh={() => query.refetch()} />}
         <Button
           variant="outline"
           size="sm"
diff --git a/apps/start/src/components/onboarding/onboarding-verify-listener.tsx b/apps/start/src/components/onboarding/onboarding-verify-listener.tsx
index 538551b0c..0360e1a3c 100644
--- a/apps/start/src/components/onboarding/onboarding-verify-listener.tsx
+++ b/apps/start/src/components/onboarding/onboarding-verify-listener.tsx
@@ -1,31 +1,13 @@
-import type {
-  IServiceClient,
-  IServiceEvent,
-  IServiceProject,
-} from '@openpanel/db';
+import type { IServiceEvent } from '@openpanel/db';
 import { CheckCircle2Icon, CheckIcon, Loader2 } from 'lucide-react';
-import { useState } from 'react';
-import useWS from '@/hooks/use-ws';
 import { cn } from '@/utils/cn';
 import { timeAgo } from '@/utils/date';
 
 interface Props {
-  project: IServiceProject;
-  client: IServiceClient | null;
   events: IServiceEvent[];
-  onVerified: (verified: boolean) => void;
 }
 
-const VerifyListener = ({ client, events: _events, onVerified }: Props) => {
-  const [events, setEvents] = useState<IServiceEvent[]>(_events ?? []);
-  useWS<IServiceEvent>(
-    `/live/events/${client?.projectId}?type=received`,
-    (data) => {
-      setEvents((prev) => [...prev, data]);
-      onVerified(true);
-    }
-  );
-
+const VerifyListener = ({ events }: Props) => {
   const isConnected = events.length > 0;
 
   const renderIcon = () => {
@@ -49,16 +31,18 @@ const VerifyListener = ({ client, events: _events, onVerified }: Props) => {
       <div
         className={cn(
           'flex gap-6 rounded-xl p-4 md:p-6',
-          isConnected ? 'bg-emerald-100 dark:bg-emerald-700' : 'bg-blue-500/10'
+          isConnected
+            ? 'bg-emerald-100 dark:bg-emerald-700/10'
+            : 'bg-blue-500/10'
         )}
       >
         {renderIcon()}
         <div className="flex-1">
           <div className="font-semibold text-foreground/90 text-lg leading-normal">
-            {isConnected ? 'Success' : 'Waiting for events'}
+            {isConnected ? 'Successfully connected' : 'Waiting for events'}
           </div>
           {isConnected ? (
-            <div className="flex flex-col-reverse">
+            <div className="mt-2 flex flex-col-reverse gap-1">
               {events.length > 5 && (
                 <div className="flex items-center gap-2">
                   <CheckIcon size={14} />{' '}
@@ -69,7 +53,7 @@ const VerifyListener = ({ client, events: _events, onVerified }: Props) => {
                 <div className="flex items-center gap-2" key={event.id}>
                   <CheckIcon size={14} />{' '}
                   <span className="font-medium">{event.name}</span>{' '}
-                  <span className="ml-auto text-emerald-800">
+                  <span className="ml-auto text-foreground/50 text-sm">
                     {timeAgo(event.createdAt, 'round')}
                   </span>
                 </div>
diff --git a/apps/start/src/components/realtime/realtime-active-sessions.tsx b/apps/start/src/components/realtime/realtime-active-sessions.tsx
index 3a6db1e90..797f1d2f7 100644
--- a/apps/start/src/components/realtime/realtime-active-sessions.tsx
+++ b/apps/start/src/components/realtime/realtime-active-sessions.tsx
@@ -1,11 +1,9 @@
-import { useTRPC } from '@/integrations/trpc/react';
 import { useQuery } from '@tanstack/react-query';
 import { AnimatePresence, motion } from 'framer-motion';
-import { useEffect, useState } from 'react';
-
-import useWS from '@/hooks/use-ws';
-import type { IServiceEvent } from '@openpanel/db';
-import { EventItem } from '../events/table/item';
+import { ProjectLink } from '../links';
+import { SerieIcon } from '../report-chart/common/serie-icon';
+import { useTRPC } from '@/integrations/trpc/react';
+import { formatTimeAgoOrDateTime } from '@/utils/date';
 
 interface RealtimeActiveSessionsProps {
   projectId: string;
@@ -17,64 +15,52 @@ export function RealtimeActiveSessions({
   limit = 10,
 }: RealtimeActiveSessionsProps) {
   const trpc = useTRPC();
-  const activeSessionsQuery = useQuery(
-    trpc.realtime.activeSessions.queryOptions({
-      projectId,
-    }),
-  );
-
-  const [state, setState] = useState<IServiceEvent[]>([]);
-
-  // Update state when initial data loads
-  useEffect(() => {
-    if (activeSessionsQuery.data && state.length === 0) {
-      setState(activeSessionsQuery.data);
-    }
-  }, [activeSessionsQuery.data, state]);
-
-  // Set up WebSocket connection for real-time updates
-  useWS<IServiceEvent>(
-    `/live/events/${projectId}`,
-    (session) => {
-      setState((prev) => {
-        // Add new session and remove duplicates, keeping most recent
-        const filtered = prev.filter((s) => s.id !== session.id);
-        return [session, ...filtered].slice(0, limit);
-      });
-    },
-    {
-      debounce: {
-        delay: 1000,
-        maxWait: 5000,
-      },
-    },
+  const { data: sessions = [] } = useQuery(
+    trpc.realtime.activeSessions.queryOptions(
+      { projectId },
+      { refetchInterval: 5000 }
+    )
   );
 
-  const sessions = state.length > 0 ? state : (activeSessionsQuery.data ?? []);
-
   return (
-    <div className="col h-full max-md:hidden">
-      <div className="hide-scrollbar h-full overflow-y-auto pb-10">
-        <AnimatePresence mode="popLayout" initial={false}>
-          <div className="col gap-4">
-            {sessions.map((session) => (
+    <div className="col card h-full max-md:hidden">
+      <div className="hide-scrollbar h-full overflow-y-auto">
+        <AnimatePresence initial={false} mode="popLayout">
+          <div className="col divide-y">
+            {sessions.slice(0, limit).map((session) => (
               <motion.div
-                key={session.id}
-                layout
-                // initial={{ opacity: 0, x: -200, scale: 0.8 }}
                 animate={{ opacity: 1, x: 0, scale: 1 }}
                 exit={{ opacity: 0, x: 200, scale: 0.8 }}
+                key={session.id}
+                layout
                 transition={{ duration: 0.4, type: 'spring', stiffness: 300 }}
               >
-                <EventItem
-                  event={session}
-                  viewOptions={{
-                    properties: false,
-                    origin: false,
-                    queryString: false,
-                  }}
-                  className="w-full"
-                />
+                <ProjectLink
+                  className="relative block p-4 py-3 pr-14"
+                  href={`/sessions/${session.sessionId}`}
+                >
+                  <div className="col flex-1 gap-1">
+                    {session.name === 'screen_view' && (
+                      <span className="text-muted-foreground text-xs leading-normal/80">
+                        {session.origin}
+                      </span>
+                    )}
+                    <span className="font-medium text-sm leading-normal">
+                      {session.name === 'screen_view'
+                        ? session.path
+                        : session.name}
+                    </span>
+                    <span className="text-muted-foreground text-xs">
+                      {formatTimeAgoOrDateTime(session.createdAt)}
+                    </span>
+                  </div>
+                  <div className="row absolute top-1/2 right-4 origin-right -translate-y-1/2 scale-50 gap-2">
+                    <SerieIcon name={session.referrerName} />
+                    <SerieIcon name={session.os} />
+                    <SerieIcon name={session.browser} />
+                    <SerieIcon name={session.device} />
+                  </div>
+                </ProjectLink>
               </motion.div>
             ))}
           </div>
diff --git a/apps/start/src/routes/_app.$organizationId.$projectId.events._tabs.events.tsx b/apps/start/src/routes/_app.$organizationId.$projectId.events._tabs.events.tsx
index 0454a830a..a9ab7694c 100644
--- a/apps/start/src/routes/_app.$organizationId.$projectId.events._tabs.events.tsx
+++ b/apps/start/src/routes/_app.$organizationId.$projectId.events._tabs.events.tsx
@@ -42,5 +42,5 @@ function Component() {
     ),
   );
 
-  return <EventsTable query={query} />;
+  return <EventsTable query={query} showEventListener />;
 }
diff --git a/apps/start/src/routes/_app.$organizationId.$projectId.realtime.tsx b/apps/start/src/routes/_app.$organizationId.$projectId.realtime.tsx
index fc7b2e8c3..4fa81b4c9 100644
--- a/apps/start/src/routes/_app.$organizationId.$projectId.realtime.tsx
+++ b/apps/start/src/routes/_app.$organizationId.$projectId.realtime.tsx
@@ -1,3 +1,5 @@
+import { keepPreviousData, useQuery } from '@tanstack/react-query';
+import { createFileRoute } from '@tanstack/react-router';
 import { Fullscreen, FullscreenClose } from '@/components/fullscreen-toggle';
 import RealtimeMap from '@/components/realtime/map';
 import { RealtimeActiveSessions } from '@/components/realtime/realtime-active-sessions';
@@ -7,12 +9,10 @@ import { RealtimePaths } from '@/components/realtime/realtime-paths';
 import { RealtimeReferrals } from '@/components/realtime/realtime-referrals';
 import RealtimeReloader from '@/components/realtime/realtime-reloader';
 import { useTRPC } from '@/integrations/trpc/react';
-import { PAGE_TITLES, createProjectTitle } from '@/utils/title';
-import { keepPreviousData, useQuery } from '@tanstack/react-query';
-import { createFileRoute } from '@tanstack/react-router';
+import { createProjectTitle, PAGE_TITLES } from '@/utils/title';
 
 export const Route = createFileRoute(
-  '/_app/$organizationId/$projectId/realtime',
+  '/_app/$organizationId/$projectId/realtime'
 )({
   component: Component,
   head: () => {
@@ -36,8 +36,8 @@ function Component() {
       },
       {
         placeholderData: keepPreviousData,
-      },
-    ),
+      }
+    )
   );
 
   return (
@@ -47,7 +47,7 @@ function Component() {
         <RealtimeReloader projectId={projectId} />
 
         <div className="row relative">
-          <div className="overflow-hidden aspect-[4/2] w-full">
+          <div className="aspect-[4/2] w-full overflow-hidden">
             <RealtimeMap
               markers={coordinatesQuery.data ?? []}
               sidebarConfig={{
@@ -56,18 +56,17 @@ function Component() {
               }}
             />
           </div>
-          <div className="absolute top-8 left-8 bottom-0 col gap-4">
-            <div className="card p-4 w-72 bg-background/90">
+          <div className="col absolute top-8 bottom-4 left-8 gap-4">
+            <div className="card w-72 bg-background/90 p-4">
               <RealtimeLiveHistogram projectId={projectId} />
             </div>
-            <div className="w-72 flex-1 min-h-0 relative">
+            <div className="relative min-h-0 w-72 flex-1">
               <RealtimeActiveSessions projectId={projectId} />
-              <div className="absolute bottom-0 left-0 right-0 h-10 bg-gradient-to-t from-def-100 to-transparent" />
             </div>
           </div>
         </div>
 
-        <div className="grid grid-cols-1 md:grid-cols-2 xl:grid-cols-3 gap-4 p-4 pt-4 md:p-8 md:pt-0">
+        <div className="grid grid-cols-1 gap-4 p-4 pt-4 md:grid-cols-2 md:p-8 md:pt-0 xl:grid-cols-3">
           <div>
             <RealtimeGeo projectId={projectId} />
           </div>
diff --git a/apps/start/src/routes/_steps.onboarding.$projectId.verify.tsx b/apps/start/src/routes/_steps.onboarding.$projectId.verify.tsx
index e1ed746ad..f600605c4 100644
--- a/apps/start/src/routes/_steps.onboarding.$projectId.verify.tsx
+++ b/apps/start/src/routes/_steps.onboarding.$projectId.verify.tsx
@@ -1,7 +1,6 @@
 import { useQuery } from '@tanstack/react-query';
 import { createFileRoute, Link, redirect } from '@tanstack/react-router';
 import { BoxSelectIcon } from 'lucide-react';
-import { useEffect, useState } from 'react';
 import { ButtonContainer } from '@/components/button-container';
 import { FullPageEmptyState } from '@/components/full-page-empty-state';
 import FullPageLoadingState from '@/components/full-page-loading-state';
@@ -33,22 +32,21 @@ export const Route = createFileRoute('/_steps/onboarding/$projectId/verify')({
 });
 
 function Component() {
-  const [isVerified, setIsVerified] = useState(false);
   const { projectId } = Route.useParams();
   const trpc = useTRPC();
-  const { data: events, refetch } = useQuery(
-    trpc.event.events.queryOptions({ projectId })
+  const { data: events } = useQuery(
+    trpc.event.events.queryOptions(
+      { projectId },
+      {
+        refetchInterval: 2500,
+      }
+    )
   );
+  const isVerified = events?.data && events.data.length > 0;
   const { data: project } = useQuery(
     trpc.project.getProjectWithClients.queryOptions({ projectId })
   );
 
-  useEffect(() => {
-    if (events && events.data.length > 0) {
-      setIsVerified(true);
-    }
-  }, [events]);
-
   if (!project) {
     return (
       <FullPageEmptyState icon={BoxSelectIcon} title="Project not found" />
@@ -64,15 +62,7 @@ function Component() {
     <div className="flex min-h-0 flex-1 flex-col">
       <div className="scrollbar-thin flex-1 overflow-y-auto">
         <div className="col gap-8 p-4">
-          <VerifyListener
-            client={client}
-            events={events?.data ?? []}
-            onVerified={() => {
-              refetch();
-              setIsVerified(true);
-            }}
-            project={project}
-          />
+          <VerifyListener events={events?.data ?? []} />
 
           <VerifyFaq project={project} />
         </div>
diff --git a/apps/worker/package.json b/apps/worker/package.json
index 60366df0a..643afc5ab 100644
--- a/apps/worker/package.json
+++ b/apps/worker/package.json
@@ -16,11 +16,11 @@
     "@openpanel/common": "workspace:*",
     "@openpanel/db": "workspace:*",
     "@openpanel/email": "workspace:*",
+    "@openpanel/importer": "workspace:*",
     "@openpanel/integrations": "workspace:^",
     "@openpanel/js-runtime": "workspace:*",
     "@openpanel/json": "workspace:*",
     "@openpanel/logger": "workspace:*",
-    "@openpanel/importer": "workspace:*",
     "@openpanel/payments": "workspace:*",
     "@openpanel/queue": "workspace:*",
     "@openpanel/redis": "workspace:*",
diff --git a/apps/worker/src/boot-workers.ts b/apps/worker/src/boot-workers.ts
index 5b4285cad..f63955650 100644
--- a/apps/worker/src/boot-workers.ts
+++ b/apps/worker/src/boot-workers.ts
@@ -1,10 +1,9 @@
-import type { Queue, WorkerOptions } from 'bullmq';
-import { Worker } from 'bullmq';
-
+import { performance } from 'node:perf_hooks';
+import { setTimeout as sleep } from 'node:timers/promises';
 import {
+  cronQueue,
   EVENTS_GROUP_QUEUES_SHARDS,
   type EventsQueuePayloadIncomingEvent,
-  cronQueue,
   eventsGroupQueues,
   gscQueue,
   importQueue,
@@ -15,14 +14,12 @@ import {
   sessionsQueue,
 } from '@openpanel/queue';
 import { getRedisQueue } from '@openpanel/redis';
-
-import { performance } from 'node:perf_hooks';
-import { setTimeout as sleep } from 'node:timers/promises';
+import type { Queue, WorkerOptions } from 'bullmq';
+import { Worker } from 'bullmq';
 import { Worker as GroupWorker } from 'groupmq';
-
 import { cronJob } from './jobs/cron';
-import { gscJob } from './jobs/gsc';
 import { incomingEvent } from './jobs/events.incoming-event';
+import { gscJob } from './jobs/gsc';
 import { importJob } from './jobs/import';
 import { insightsProjectJob } from './jobs/insights';
 import { miscJob } from './jobs/misc';
@@ -95,7 +92,7 @@ function getConcurrencyFor(queueName: string, defaultValue = 1): number {
   return defaultValue;
 }
 
-export async function bootWorkers() {
+export function bootWorkers() {
   const enabledQueues = getEnabledQueues();
 
   const workers: (Worker | GroupWorker<any>)[] = [];
@@ -119,12 +116,14 @@ export async function bootWorkers() {
 
   for (const index of eventQueuesToStart) {
     const queue = eventsGroupQueues[index];
-    if (!queue) continue;
+    if (!queue) {
+      continue;
+    }
 
     const queueName = `events_${index}`;
     const concurrency = getConcurrencyFor(
       queueName,
-      Number.parseInt(process.env.EVENT_JOB_CONCURRENCY || '10', 10),
+      Number.parseInt(process.env.EVENT_JOB_CONCURRENCY || '10', 10)
     );
 
     const worker = new GroupWorker<EventsQueuePayloadIncomingEvent['payload']>({
@@ -132,7 +131,7 @@ export async function bootWorkers() {
       concurrency,
       logger: process.env.NODE_ENV === 'production' ? queueLogger : undefined,
       blockingTimeoutSec: Number.parseFloat(
-        process.env.EVENT_BLOCKING_TIMEOUT_SEC || '1',
+        process.env.EVENT_BLOCKING_TIMEOUT_SEC || '1'
       ),
       handler: async (job) => {
         return await incomingEvent(job.data);
@@ -172,7 +171,7 @@ export async function bootWorkers() {
     const notificationWorker = new Worker(
       notificationQueue.name,
       notificationJob,
-      { ...workerOptions, concurrency },
+      { ...workerOptions, concurrency }
     );
     workers.push(notificationWorker);
     logger.info('Started worker for notification', { concurrency });
@@ -224,7 +223,7 @@ export async function bootWorkers() {
 
   if (workers.length === 0) {
     logger.warn(
-      'No workers started. Check ENABLED_QUEUES environment variable.',
+      'No workers started. Check ENABLED_QUEUES environment variable.'
     );
   }
 
@@ -254,7 +253,7 @@ export async function bootWorkers() {
           const elapsed = job.finishedOn - job.processedOn;
           eventsGroupJobDuration.observe(
             { name: worker.name, status: 'failed' },
-            elapsed,
+            elapsed
           );
         }
         logger.error('job failed', {
@@ -267,23 +266,6 @@ export async function bootWorkers() {
       }
     });
 
-    (worker as Worker).on('completed', (job) => {
-      if (job) {
-        if (job.processedOn && job.finishedOn) {
-          const elapsed = job.finishedOn - job.processedOn;
-          logger.info('job completed', {
-            jobId: job.id,
-            worker: worker.name,
-            elapsed,
-          });
-          eventsGroupJobDuration.observe(
-            { name: worker.name, status: 'success' },
-            elapsed,
-          );
-        }
-      }
-    });
-
     (worker as Worker).on('ioredis:close', () => {
       logger.error('worker closed due to ioredis:close', {
         worker: worker.name,
@@ -293,7 +275,7 @@ export async function bootWorkers() {
 
   async function exitHandler(
     eventName: string,
-    evtOrExitCodeOrError: number | string | Error,
+    evtOrExitCodeOrError: number | string | Error
   ) {
     // Log the actual error details for unhandled rejections/exceptions
     if (evtOrExitCodeOrError instanceof Error) {
@@ -339,7 +321,7 @@ export async function bootWorkers() {
       process.on(evt, (code) => {
         exitHandler(evt, code);
       });
-    },
+    }
   );
 
   return workers;
diff --git a/apps/worker/src/jobs/cron.salt.ts b/apps/worker/src/jobs/cron.salt.ts
index e7575f9fe..8b5b02e4f 100644
--- a/apps/worker/src/jobs/cron.salt.ts
+++ b/apps/worker/src/jobs/cron.salt.ts
@@ -33,7 +33,7 @@ async function generateNewSalt() {
     return created;
   });
 
-  getSalts.clear();
+  await getSalts.clear();
 
   return newSalt;
 }
diff --git a/apps/worker/src/jobs/events.incoming-event.ts b/apps/worker/src/jobs/events.incoming-event.ts
index 21b169c0b..1f15e5d34 100644
--- a/apps/worker/src/jobs/events.incoming-event.ts
+++ b/apps/worker/src/jobs/events.incoming-event.ts
@@ -1,9 +1,5 @@
 import { getTime, isSameDomain, parsePath } from '@openpanel/common';
-import {
-  getReferrerWithQuery,
-  parseReferrer,
-  parseUserAgent,
-} from '@openpanel/common/server';
+import { getReferrerWithQuery, parseReferrer } from '@openpanel/common/server';
 import type { IServiceCreateEventPayload, IServiceEvent } from '@openpanel/db';
 import {
   checkNotificationRulesForEvent,
@@ -14,10 +10,12 @@ import {
 } from '@openpanel/db';
 import type { ILogger } from '@openpanel/logger';
 import type { EventsQueuePayloadIncomingEvent } from '@openpanel/queue';
-import { getLock } from '@openpanel/redis';
 import { anyPass, isEmpty, isNil, mergeDeepRight, omit, reject } from 'ramda';
 import { logger as baseLogger } from '@/utils/logger';
-import { createSessionEndJob, getSessionEnd } from '@/utils/session-handler';
+import {
+  createSessionEndJob,
+  extendSessionEndJob,
+} from '@/utils/session-handler';
 
 const GLOBAL_PROPERTIES = ['__path', '__referrer', '__timestamp', '__revenue'];
 
@@ -93,7 +91,8 @@ export async function incomingEvent(
     projectId,
     deviceId,
     sessionId,
-    uaInfo: _uaInfo,
+    uaInfo,
+    session,
   } = jobPayload;
   const properties = body.properties ?? {};
   const reqId = headers['request-id'] ?? 'unknown';
@@ -121,16 +120,15 @@ export async function incomingEvent(
     ? null
     : parseReferrer(getProperty('__referrer'));
   const utmReferrer = getReferrerWithQuery(query);
-  const userAgent = headers['user-agent'];
   const sdkName = headers['openpanel-sdk-name'];
   const sdkVersion = headers['openpanel-sdk-version'];
-  // TODO: Remove both user-agent and parseUserAgent
-  const uaInfo = _uaInfo ?? parseUserAgent(userAgent, properties);
 
-  const baseEvent = {
+  const baseEvent: IServiceCreateEventPayload = {
     name: body.name,
     profileId,
     projectId,
+    deviceId,
+    sessionId,
     properties: omit(GLOBAL_PROPERTIES, {
       ...properties,
       __hash: hash,
@@ -149,7 +147,7 @@ export async function incomingEvent(
     origin,
     referrer: referrer?.url || '',
     referrerName: utmReferrer?.name || referrer?.name || referrer?.url,
-    referrerType: referrer?.type || utmReferrer?.type || '',
+    referrerType: utmReferrer?.type || referrer?.type || '',
     os: uaInfo.os,
     osVersion: uaInfo.osVersion,
     browser: uaInfo.browser,
@@ -161,16 +159,17 @@ export async function incomingEvent(
       body.name === 'revenue' && '__revenue' in properties
         ? parseRevenue(properties.__revenue)
         : undefined,
-  } as const;
+  };
 
   // if timestamp is from the past we dont want to create a new session
   if (uaInfo.isServer || isTimestampFromThePast) {
-    const session = profileId
-      ? await sessionBuffer.getExistingSession({
-          profileId,
-          projectId,
-        })
-      : null;
+    const session =
+      profileId && !isTimestampFromThePast
+        ? await sessionBuffer.getExistingSession({
+            profileId,
+            projectId,
+          })
+        : null;
 
     const payload = {
       ...baseEvent,
@@ -198,82 +197,48 @@ export async function incomingEvent(
     return createEventAndNotify(payload as IServiceEvent, logger, projectId);
   }
 
-  const sessionEnd = await getSessionEnd({
-    projectId,
-    deviceId,
-    profileId,
-  });
-  const activeSession = sessionEnd
-    ? await sessionBuffer.getExistingSession({
-        sessionId: sessionEnd.sessionId,
-      })
-    : null;
-
   const payload: IServiceCreateEventPayload = merge(baseEvent, {
-    deviceId: sessionEnd?.deviceId ?? deviceId,
-    sessionId: sessionEnd?.sessionId ?? sessionId,
-    referrer: sessionEnd?.referrer ?? baseEvent.referrer,
-    referrerName: sessionEnd?.referrerName ?? baseEvent.referrerName,
-    referrerType: sessionEnd?.referrerType ?? baseEvent.referrerType,
-    // if the path is not set, use the last screen view path
-    path: baseEvent.path || activeSession?.exit_path || '',
-    origin: baseEvent.origin || activeSession?.exit_origin || '',
+    referrer: session?.referrer ?? baseEvent.referrer,
+    referrerName: session?.referrerName ?? baseEvent.referrerName,
+    referrerType: session?.referrerType ?? baseEvent.referrerType,
   } as Partial<IServiceCreateEventPayload>) as IServiceCreateEventPayload;
 
-  // If the triggering event is filtered, do not create session_start or the event (issue #2)
   const isExcluded = await isEventExcludedByProjectFilter(payload, projectId);
   if (isExcluded) {
     logger.info(
       'Skipping session_start and event (excluded by project filter)',
-      {
-        event: payload.name,
-        projectId,
-      }
+      { event: payload.name, projectId }
     );
     return null;
   }
 
-  if (!sessionEnd) {
-    const locked = await getLock(
-      `session_start:${projectId}:${sessionId}`,
-      '1',
-      1000
-    );
-    if (locked) {
-      logger.info('Creating session start event', { event: payload });
-      await createEventAndNotify(
-        {
-          ...payload,
-          name: 'session_start',
-          createdAt: new Date(getTime(payload.createdAt) - 100),
-        },
-        logger,
-        projectId
-      ).catch((error) => {
-        logger.error('Error creating session start event', { event: payload });
-        throw error;
-      });
-    } else {
-      logger.info('Session start already claimed by another worker', {
-        event: payload,
-      });
-    }
-  }
-
-  const event = await createEventAndNotify(payload, logger, projectId);
-
-  if (!event) {
-    // Skip creating session end when event was excluded
-    return null;
-  }
+  if (session) {
+    await extendSessionEndJob({
+      projectId,
+      deviceId,
+    }).catch((error) => {
+      logger.error('Error finding and extending session end job', { error });
+      throw error;
+    });
+  } else {
+    await createEventAndNotify(
+      {
+        ...payload,
+        name: 'session_start',
+        createdAt: new Date(getTime(payload.createdAt) - 100),
+      },
+      logger,
+      projectId
+    ).catch((error) => {
+      logger.error('Error creating session start event', { event: payload });
+      throw error;
+    });
 
-  if (!sessionEnd) {
-    logger.info('Creating session end job', { event: payload });
     await createSessionEndJob({ payload }).catch((error) => {
       logger.error('Error creating session end job', { event: payload });
       throw error;
     });
   }
 
-  return event;
+  return createEventAndNotify(payload, logger, projectId);
 }
diff --git a/apps/worker/src/jobs/events.incoming-events.test.ts b/apps/worker/src/jobs/events.incoming-events.test.ts
index f1c24feb2..f483fb741 100644
--- a/apps/worker/src/jobs/events.incoming-events.test.ts
+++ b/apps/worker/src/jobs/events.incoming-events.test.ts
@@ -186,6 +186,11 @@ describe('incomingEvent', () => {
       projectId,
       deviceId,
       sessionId: 'session-123',
+      session: {
+        referrer: '',
+        referrerName: '',
+        referrerType: '',
+      },
     };
 
     const changeDelay = vi.fn();
diff --git a/apps/worker/src/metrics.ts b/apps/worker/src/metrics.ts
index eaea78d2d..0eabf89d6 100644
--- a/apps/worker/src/metrics.ts
+++ b/apps/worker/src/metrics.ts
@@ -1,5 +1,3 @@
-import client from 'prom-client';
-
 import {
   botBuffer,
   eventBuffer,
@@ -8,6 +6,7 @@ import {
   sessionBuffer,
 } from '@openpanel/db';
 import { cronQueue, eventsGroupQueues, sessionsQueue } from '@openpanel/queue';
+import client from 'prom-client';
 
 const Registry = client.Registry;
 
@@ -20,7 +19,7 @@ export const eventsGroupJobDuration = new client.Histogram({
   name: 'job_duration_ms',
   help: 'Duration of job processing (in ms)',
   labelNames: ['name', 'status'],
-  buckets: [10, 25, 50, 100, 250, 500, 750, 1000, 2000, 5000, 10000, 30000], // 10ms to 30s
+  buckets: [10, 25, 50, 100, 250, 500, 750, 1000, 2000, 5000, 10_000, 30_000], // 10ms to 30s
 });
 
 register.registerMetric(eventsGroupJobDuration);
@@ -28,57 +27,61 @@ register.registerMetric(eventsGroupJobDuration);
 queues.forEach((queue) => {
   register.registerMetric(
     new client.Gauge({
-      name: `${queue.name.replace(/[\{\}]/g, '')}_active_count`,
+      name: `${queue.name.replace(/[{}]/g, '')}_active_count`,
       help: 'Active count',
       async collect() {
         const metric = await queue.getActiveCount();
         this.set(metric);
       },
-    }),
+    })
   );
 
   register.registerMetric(
     new client.Gauge({
-      name: `${queue.name.replace(/[\{\}]/g, '')}_delayed_count`,
+      name: `${queue.name.replace(/[{}]/g, '')}_delayed_count`,
       help: 'Delayed count',
       async collect() {
-        const metric = await queue.getDelayedCount();
-        this.set(metric);
+        if ('getDelayedCount' in queue) {
+          const metric = await queue.getDelayedCount();
+          this.set(metric);
+        } else {
+          this.set(0);
+        }
       },
-    }),
+    })
   );
 
   register.registerMetric(
     new client.Gauge({
-      name: `${queue.name.replace(/[\{\}]/g, '')}_failed_count`,
+      name: `${queue.name.replace(/[{}]/g, '')}_failed_count`,
       help: 'Failed count',
       async collect() {
         const metric = await queue.getFailedCount();
         this.set(metric);
       },
-    }),
+    })
   );
 
   register.registerMetric(
     new client.Gauge({
-      name: `${queue.name.replace(/[\{\}]/g, '')}_completed_count`,
+      name: `${queue.name.replace(/[{}]/g, '')}_completed_count`,
       help: 'Completed count',
       async collect() {
         const metric = await queue.getCompletedCount();
         this.set(metric);
       },
-    }),
+    })
   );
 
   register.registerMetric(
     new client.Gauge({
-      name: `${queue.name.replace(/[\{\}]/g, '')}_waiting_count`,
+      name: `${queue.name.replace(/[{}]/g, '')}_waiting_count`,
       help: 'Waiting count',
       async collect() {
         const metric = await queue.getWaitingCount();
         this.set(metric);
       },
-    }),
+    })
   );
 });
 
@@ -90,7 +93,7 @@ register.registerMetric(
       const metric = await eventBuffer.getBufferSize();
       this.set(metric);
     },
-  }),
+  })
 );
 
 register.registerMetric(
@@ -101,7 +104,7 @@ register.registerMetric(
       const metric = await profileBuffer.getBufferSize();
       this.set(metric);
     },
-  }),
+  })
 );
 
 register.registerMetric(
@@ -112,7 +115,7 @@ register.registerMetric(
       const metric = await botBuffer.getBufferSize();
       this.set(metric);
     },
-  }),
+  })
 );
 
 register.registerMetric(
@@ -123,7 +126,7 @@ register.registerMetric(
       const metric = await sessionBuffer.getBufferSize();
       this.set(metric);
     },
-  }),
+  })
 );
 
 register.registerMetric(
@@ -134,5 +137,5 @@ register.registerMetric(
       const metric = await replayBuffer.getBufferSize();
       this.set(metric);
     },
-  }),
+  })
 );
diff --git a/apps/worker/src/utils/session-handler.ts b/apps/worker/src/utils/session-handler.ts
index 01008c1fc..0ab9511ea 100644
--- a/apps/worker/src/utils/session-handler.ts
+++ b/apps/worker/src/utils/session-handler.ts
@@ -1,13 +1,39 @@
 import type { IServiceCreateEventPayload } from '@openpanel/db';
-import {
-  type EventsQueuePayloadCreateSessionEnd,
-  sessionsQueue,
-} from '@openpanel/queue';
-import type { Job } from 'bullmq';
-import { logger } from './logger';
+import { sessionsQueue } from '@openpanel/queue';
 
 export const SESSION_TIMEOUT = 1000 * 60 * 30;
 
+const CHANGE_DELAY_THROTTLE_MS = process.env.CHANGE_DELAY_THROTTLE_MS
+  ? Number.parseInt(process.env.CHANGE_DELAY_THROTTLE_MS, 10)
+  : 60_000; // 1 minute
+
+const CHANGE_DELAY_THROTTLE_MAP = new Map<string, number>();
+
+export async function extendSessionEndJob({
+  projectId,
+  deviceId,
+}: {
+  projectId: string;
+  deviceId: string;
+}) {
+  const last = CHANGE_DELAY_THROTTLE_MAP.get(`${projectId}:${deviceId}`) ?? 0;
+  const isThrottled = Date.now() - last < CHANGE_DELAY_THROTTLE_MS;
+
+  if (isThrottled) {
+    return;
+  }
+
+  const jobId = getSessionEndJobId(projectId, deviceId);
+  const job = await sessionsQueue.getJob(jobId);
+
+  if (!job) {
+    return;
+  }
+
+  await job.changeDelay(SESSION_TIMEOUT);
+  CHANGE_DELAY_THROTTLE_MAP.set(`${projectId}:${deviceId}`, Date.now());
+}
+
 const getSessionEndJobId = (projectId: string, deviceId: string) =>
   `sessionEnd:${projectId}:${deviceId}`;
 
@@ -33,106 +59,3 @@ export function createSessionEndJob({
     }
   );
 }
-
-export async function getSessionEnd({
-  projectId,
-  deviceId,
-  profileId,
-}: {
-  projectId: string;
-  deviceId: string;
-  profileId: string;
-}) {
-  const sessionEnd = await getSessionEndJob({
-    projectId,
-    deviceId,
-  });
-
-  if (sessionEnd) {
-    const existingSessionIsAnonymous =
-      sessionEnd.job.data.payload.profileId ===
-      sessionEnd.job.data.payload.deviceId;
-
-    const eventIsIdentified =
-      profileId && sessionEnd.job.data.payload.profileId !== profileId;
-
-    if (existingSessionIsAnonymous && eventIsIdentified) {
-      await sessionEnd.job.updateData({
-        ...sessionEnd.job.data,
-        payload: {
-          ...sessionEnd.job.data.payload,
-          profileId,
-        },
-      });
-    }
-
-    await sessionEnd.job.changeDelay(SESSION_TIMEOUT);
-    return sessionEnd.job.data.payload;
-  }
-
-  return null;
-}
-
-export async function getSessionEndJob(args: {
-  projectId: string;
-  deviceId: string;
-  retryCount?: number;
-}): Promise<{
-  deviceId: string;
-  job: Job<EventsQueuePayloadCreateSessionEnd>;
-} | null> {
-  const { retryCount = 0 } = args;
-
-  if (retryCount >= 6) {
-    throw new Error('Failed to get session end');
-  }
-
-  async function handleJobStates(
-    job: Job<EventsQueuePayloadCreateSessionEnd>,
-    deviceId: string
-  ): Promise<{
-    deviceId: string;
-    job: Job<EventsQueuePayloadCreateSessionEnd>;
-  } | null> {
-    const state = await job.getState();
-    if (state !== 'delayed') {
-      logger.debug(`[session-handler] Session end job is in "${state}" state`, {
-        state,
-        retryCount,
-        jobTimestamp: new Date(job.timestamp).toISOString(),
-        jobDelta: Date.now() - job.timestamp,
-        jobId: job.id,
-        payload: job.data.payload,
-      });
-    }
-
-    if (state === 'delayed' || state === 'waiting') {
-      return { deviceId, job };
-    }
-
-    if (state === 'active') {
-      await new Promise((resolve) => setTimeout(resolve, 100));
-      return getSessionEndJob({
-        ...args,
-        retryCount: retryCount + 1,
-      });
-    }
-
-    if (state === 'completed') {
-      await job.remove();
-    }
-
-    return null;
-  }
-
-  // Check current device job
-  const currentJob = await sessionsQueue.getJob(
-    getSessionEndJobId(args.projectId, args.deviceId)
-  );
-  if (currentJob) {
-    return await handleJobStates(currentJob, args.deviceId);
-  }
-
-  // Create session
-  return null;
-}
diff --git a/biome.json b/biome.json
index 347e1b482..9e7b023e8 100644
--- a/biome.json
+++ b/biome.json
@@ -42,6 +42,7 @@
         "useSemanticElements": "off"
       },
       "style": {
+        "noNestedTernary": "off",
         "noNonNullAssertion": "off",
         "noParameterAssign": "error",
         "useAsConstAssertion": "error",
diff --git a/packages/db/src/buffers/event-buffer.test.ts b/packages/db/src/buffers/event-buffer.test.ts
index 95852bd29..50600c704 100644
--- a/packages/db/src/buffers/event-buffer.test.ts
+++ b/packages/db/src/buffers/event-buffer.test.ts
@@ -71,8 +71,9 @@ describe('EventBuffer', () => {
     // Get initial count
     const initialCount = await eventBuffer.getBufferSize();
 
-    // Add event
-    await eventBuffer.add(event);
+    // Add event and flush (events are micro-batched)
+    eventBuffer.add(event);
+    await eventBuffer.flush();
 
     // Buffer counter should increase by 1
     const newCount = await eventBuffer.getBufferSize();
@@ -109,7 +110,8 @@ describe('EventBuffer', () => {
 
     // Add first screen_view
     const count1 = await eventBuffer.getBufferSize();
-    await eventBuffer.add(view1);
+    eventBuffer.add(view1);
+    await eventBuffer.flush();
 
     // Should be stored as "last" but NOT in queue yet
     const count2 = await eventBuffer.getBufferSize();
@@ -124,7 +126,8 @@ describe('EventBuffer', () => {
     expect(last1!.createdAt.toISOString()).toBe(view1.created_at);
 
     // Add second screen_view
-    await eventBuffer.add(view2);
+    eventBuffer.add(view2);
+    await eventBuffer.flush();
 
     // Now view1 should be in buffer
     const count3 = await eventBuffer.getBufferSize();
@@ -138,7 +141,8 @@ describe('EventBuffer', () => {
     expect(last2!.createdAt.toISOString()).toBe(view2.created_at);
 
     // Add third screen_view
-    await eventBuffer.add(view3);
+    eventBuffer.add(view3);
+    await eventBuffer.flush();
 
     // Now view2 should also be in buffer
     const count4 = await eventBuffer.getBufferSize();
@@ -174,14 +178,16 @@ describe('EventBuffer', () => {
 
     // Add screen_view
     const count1 = await eventBuffer.getBufferSize();
-    await eventBuffer.add(view);
+    eventBuffer.add(view);
+    await eventBuffer.flush();
 
     // Should be stored as "last", not in buffer yet
     const count2 = await eventBuffer.getBufferSize();
     expect(count2).toBe(count1);
 
     // Add session_end
-    await eventBuffer.add(sessionEnd);
+    eventBuffer.add(sessionEnd);
+    await eventBuffer.flush();
 
     // Both should now be in buffer (+2)
     const count3 = await eventBuffer.getBufferSize();
@@ -207,7 +213,8 @@ describe('EventBuffer', () => {
     } as any;
 
     const count1 = await eventBuffer.getBufferSize();
-    await eventBuffer.add(sessionEnd);
+    eventBuffer.add(sessionEnd);
+    await eventBuffer.flush();
 
     // Only session_end should be in buffer (+1)
     const count2 = await eventBuffer.getBufferSize();
@@ -224,7 +231,8 @@ describe('EventBuffer', () => {
       created_at: new Date().toISOString(),
     } as any;
 
-    await eventBuffer.add(view);
+    eventBuffer.add(view);
+    await eventBuffer.flush();
 
     // Query by profileId
     const result = await eventBuffer.getLastScreenView({
@@ -248,7 +256,8 @@ describe('EventBuffer', () => {
       created_at: new Date().toISOString(),
     } as any;
 
-    await eventBuffer.add(view);
+    eventBuffer.add(view);
+    await eventBuffer.flush();
 
     // Query by sessionId
     const result = await eventBuffer.getLastScreenView({
@@ -275,43 +284,47 @@ describe('EventBuffer', () => {
     expect(await eventBuffer.getBufferSize()).toBe(0);
 
     // Add regular event
-    await eventBuffer.add({
+    eventBuffer.add({
       project_id: 'p6',
       name: 'event1',
       created_at: new Date().toISOString(),
     } as any);
+    await eventBuffer.flush();
 
     expect(await eventBuffer.getBufferSize()).toBe(1);
 
     // Add another regular event
-    await eventBuffer.add({
+    eventBuffer.add({
       project_id: 'p6',
       name: 'event2',
       created_at: new Date().toISOString(),
     } as any);
+    await eventBuffer.flush();
 
     expect(await eventBuffer.getBufferSize()).toBe(2);
 
     // Add screen_view (not counted until flushed)
-    await eventBuffer.add({
+    eventBuffer.add({
       project_id: 'p6',
       profile_id: 'u6',
       session_id: 'session_6',
       name: 'screen_view',
       created_at: new Date().toISOString(),
     } as any);
+    await eventBuffer.flush();
 
     // Still 2 (screen_view is pending)
     expect(await eventBuffer.getBufferSize()).toBe(2);
 
     // Add another screen_view (first one gets flushed)
-    await eventBuffer.add({
+    eventBuffer.add({
       project_id: 'p6',
       profile_id: 'u6',
       session_id: 'session_6',
       name: 'screen_view',
       created_at: new Date(Date.now() + 1000).toISOString(),
     } as any);
+    await eventBuffer.flush();
 
     // Now 3 (2 regular + 1 flushed screen_view)
     expect(await eventBuffer.getBufferSize()).toBe(3);
@@ -330,8 +343,9 @@ describe('EventBuffer', () => {
       created_at: new Date(Date.now() + 1000).toISOString(),
     } as any;
 
-    await eventBuffer.add(event1);
-    await eventBuffer.add(event2);
+    eventBuffer.add(event1);
+    eventBuffer.add(event2);
+    await eventBuffer.flush();
 
     expect(await eventBuffer.getBufferSize()).toBe(2);
 
@@ -361,12 +375,13 @@ describe('EventBuffer', () => {
 
     // Add 4 events
     for (let i = 0; i < 4; i++) {
-      await eb.add({
+      eb.add({
         project_id: 'p8',
         name: `event${i}`,
         created_at: new Date(Date.now() + i).toISOString(),
       } as any);
     }
+    await eb.flush();
 
     const insertSpy = vi
       .spyOn(ch, 'insert')
@@ -396,7 +411,8 @@ describe('EventBuffer', () => {
       created_at: new Date().toISOString(),
     } as any;
 
-    await eventBuffer.add(event);
+    eventBuffer.add(event);
+    await eventBuffer.flush();
 
     const count = await eventBuffer.getActiveVisitorCount('p9');
     expect(count).toBeGreaterThanOrEqual(1);
@@ -439,10 +455,11 @@ describe('EventBuffer', () => {
       created_at: new Date(t0 + 2000).toISOString(),
     } as any;
 
-    await eventBuffer.add(view1a);
-    await eventBuffer.add(view2a);
-    await eventBuffer.add(view1b); // Flushes view1a
-    await eventBuffer.add(view2b); // Flushes view2a
+    eventBuffer.add(view1a);
+    eventBuffer.add(view2a);
+    eventBuffer.add(view1b); // Flushes view1a
+    eventBuffer.add(view2b); // Flushes view2a
+    await eventBuffer.flush();
 
     // Should have 2 events in buffer (one from each session)
     expect(await eventBuffer.getBufferSize()).toBe(2);
@@ -470,7 +487,8 @@ describe('EventBuffer', () => {
     } as any;
 
     const count1 = await eventBuffer.getBufferSize();
-    await eventBuffer.add(view);
+    eventBuffer.add(view);
+    await eventBuffer.flush();
 
     // Should go directly to buffer (no session_id)
     const count2 = await eventBuffer.getBufferSize();
@@ -498,8 +516,9 @@ describe('EventBuffer', () => {
       created_at: new Date(t0 + 1000).toISOString(),
     } as any;
 
-    await eventBuffer.add(view1);
-    await eventBuffer.add(view2);
+    eventBuffer.add(view1);
+    eventBuffer.add(view2);
+    await eventBuffer.flush();
 
     // Both sessions should have their own "last"
     const lastSession1 = await eventBuffer.getLastScreenView({
diff --git a/packages/db/src/buffers/event-buffer.ts b/packages/db/src/buffers/event-buffer.ts
index d305372aa..306b69d1b 100644
--- a/packages/db/src/buffers/event-buffer.ts
+++ b/packages/db/src/buffers/event-buffer.ts
@@ -2,7 +2,6 @@ import { getSafeJson } from '@openpanel/json';
 import {
   type Redis,
   getRedisCache,
-  getRedisPub,
   publishEvent,
 } from '@openpanel/redis';
 import { ch } from '../clickhouse/client';
@@ -14,9 +13,8 @@ import {
 import { BaseBuffer } from './base-buffer';
 
 /**
- * Simplified Event Buffer
+ * Event Buffer
  *
- * Rules:
  * 1. All events go into a single list buffer (event_buffer:queue)
  * 2. screen_view events are handled specially:
  *    - Store current screen_view as "last" for the session
@@ -24,11 +22,16 @@ import { BaseBuffer } from './base-buffer';
  * 3. session_end events:
  *    - Retrieve the last screen_view (don't modify it)
  *    - Push both screen_view and session_end to buffer
- * 4. Flush: Simply process all events from the list buffer
+ * 4. Flush: Process all events from the list buffer
  */
+interface PendingEvent {
+  event: IClickhouseEvent;
+  eventJson: string;
+  eventWithTimestamp?: string;
+  type: 'regular' | 'screen_view' | 'session_end';
+}
 
 export class EventBuffer extends BaseBuffer {
-  // Configurable limits
   private batchSize = process.env.EVENT_BUFFER_BATCH_SIZE
     ? Number.parseInt(process.env.EVENT_BUFFER_BATCH_SIZE, 10)
     : 4000;
@@ -36,37 +39,44 @@ export class EventBuffer extends BaseBuffer {
     ? Number.parseInt(process.env.EVENT_BUFFER_CHUNK_SIZE, 10)
     : 1000;
 
-  private activeVisitorsExpiration = 60 * 5; // 5 minutes
+  private microBatchIntervalMs = process.env.EVENT_BUFFER_MICRO_BATCH_MS
+    ? Number.parseInt(process.env.EVENT_BUFFER_MICRO_BATCH_MS, 10)
+    : 10;
+  private microBatchMaxSize = process.env.EVENT_BUFFER_MICRO_BATCH_SIZE
+    ? Number.parseInt(process.env.EVENT_BUFFER_MICRO_BATCH_SIZE, 10)
+    : 100;
 
-  // LIST - Stores all events ready to be flushed
-  private queueKey = 'event_buffer:queue';
+  private pendingEvents: PendingEvent[] = [];
+  private flushTimer: ReturnType<typeof setTimeout> | null = null;
+  private isFlushing = false;
+  /** Tracks consecutive flush failures for observability; reset on success. */
+  private flushRetryCount = 0;
 
-  // STRING - Tracks total buffer size incrementally
+  private activeVisitorsExpiration = 60 * 5; // 5 minutes
+  /** How often (ms) we refresh the heartbeat key + zadd per visitor. */
+  private heartbeatRefreshMs = 60_000; // 1 minute
+  private lastHeartbeat = new Map<string, number>();
+  private queueKey = 'event_buffer:queue';
   protected bufferCounterKey = 'event_buffer:total_count';
 
-  // Script SHAs for loaded Lua scripts
   private scriptShas: {
     addScreenView?: string;
     addSessionEnd?: string;
   } = {};
 
-  // Hash key for storing last screen_view per session
   private getLastScreenViewKeyBySession(sessionId: string) {
     return `event_buffer:last_screen_view:session:${sessionId}`;
   }
 
-  // Hash key for storing last screen_view per profile
   private getLastScreenViewKeyByProfile(projectId: string, profileId: string) {
     return `event_buffer:last_screen_view:profile:${projectId}:${profileId}`;
   }
 
   /**
-   * Lua script for handling screen_view addition - RACE-CONDITION SAFE without GroupMQ
+   * Lua script for screen_view addition.
+   * Uses GETDEL for atomic get-and-delete to prevent race conditions.
    *
-   * Strategy: Use Redis GETDEL (atomic get-and-delete) to ensure only ONE thread
-   * can process the "last" screen_view at a time.
-   *
-   * KEYS[1] = last screen_view key (by session) - stores both event and timestamp as JSON
+   * KEYS[1] = last screen_view key (by session)
    * KEYS[2] = last screen_view key (by profile, may be empty)
    * KEYS[3] = queue key
    * KEYS[4] = buffer counter key
@@ -81,24 +91,18 @@ local counterKey = KEYS[4]
 local newEventData = ARGV[1]
 local ttl = tonumber(ARGV[2])
 
--- GETDEL is atomic: get previous and delete in one operation
--- This ensures only ONE thread gets the previous event
 local previousEventData = redis.call("GETDEL", sessionKey)
 
--- Store new screen_view as last for session
 redis.call("SET", sessionKey, newEventData, "EX", ttl)
 
--- Store new screen_view as last for profile (if key provided)
 if profileKey and profileKey ~= "" then
   redis.call("SET", profileKey, newEventData, "EX", ttl)
 end
 
--- If there was a previous screen_view, add it to queue with calculated duration
 if previousEventData then
   local prev = cjson.decode(previousEventData)
   local curr = cjson.decode(newEventData)
   
-  -- Calculate duration (ensure non-negative to handle clock skew)
   if prev.ts and curr.ts then
     prev.event.duration = math.max(0, curr.ts - prev.ts)
   end
@@ -112,9 +116,8 @@ return 0
 `;
 
   /**
-   * Lua script for handling session_end - RACE-CONDITION SAFE
-   *
-   * Uses GETDEL to atomically retrieve and delete the last screen_view
+   * Lua script for session_end.
+   * Uses GETDEL to atomically retrieve and delete the last screen_view.
    *
    * KEYS[1] = last screen_view key (by session)
    * KEYS[2] = last screen_view key (by profile, may be empty)
@@ -129,11 +132,9 @@ local queueKey = KEYS[3]
 local counterKey = KEYS[4]
 local sessionEndJson = ARGV[1]
 
--- GETDEL is atomic: only ONE thread gets the last screen_view
 local previousEventData = redis.call("GETDEL", sessionKey)
 local added = 0
 
--- If there was a previous screen_view, add it to queue
 if previousEventData then
   local prev = cjson.decode(previousEventData)
   redis.call("RPUSH", queueKey, cjson.encode(prev.event))
@@ -141,12 +142,10 @@ if previousEventData then
   added = added + 1
 end
 
--- Add session_end to queue
 redis.call("RPUSH", queueKey, sessionEndJson)
 redis.call("INCR", counterKey)
 added = added + 1
 
--- Delete profile key
 if profileKey and profileKey ~= "" then
   redis.call("DEL", profileKey)
 end
@@ -161,14 +160,9 @@ return added
         await this.processBuffer();
       },
     });
-    // Load Lua scripts into Redis on startup
     this.loadScripts();
   }
 
-  /**
-   * Load Lua scripts into Redis and cache their SHAs.
-   * This avoids sending the entire script on every call.
-   */
   private async loadScripts() {
     try {
       const redis = getRedisCache();
@@ -190,105 +184,153 @@ return added
   }
 
   bulkAdd(events: IClickhouseEvent[]) {
-    const redis = getRedisCache();
-    const multi = redis.multi();
     for (const event of events) {
-      this.add(event, multi);
+      this.add(event);
     }
-    return multi.exec();
   }
 
-  /**
-   * Add an event into Redis buffer.
-   *
-   * Logic:
-   * - screen_view: Store as "last" for session, flush previous if exists
-   * - session_end: Flush last screen_view + session_end
-   * - Other events: Add directly to queue
-   */
-  async add(event: IClickhouseEvent, _multi?: ReturnType<Redis['multi']>) {
+  add(event: IClickhouseEvent) {
+    const eventJson = JSON.stringify(event);
+
+    let type: PendingEvent['type'] = 'regular';
+    let eventWithTimestamp: string | undefined;
+
+    if (event.session_id && event.name === 'screen_view') {
+      type = 'screen_view';
+      const timestamp = new Date(event.created_at || Date.now()).getTime();
+      eventWithTimestamp = JSON.stringify({
+        event: event,
+        ts: timestamp,
+      });
+    } else if (event.session_id && event.name === 'session_end') {
+      type = 'session_end';
+    }
+
+    const pendingEvent: PendingEvent = {
+      event,
+      eventJson,
+      eventWithTimestamp,
+      type,
+    };
+
+    this.pendingEvents.push(pendingEvent);
+
+    if (this.pendingEvents.length >= this.microBatchMaxSize) {
+      this.flushLocalBuffer();
+      return;
+    }
+
+    if (!this.flushTimer) {
+      this.flushTimer = setTimeout(() => {
+        this.flushTimer = null;
+        this.flushLocalBuffer();
+      }, this.microBatchIntervalMs);
+    }
+  }
+
+  private addToMulti(multi: ReturnType<Redis['multi']>, pending: PendingEvent) {
+    const { event, eventJson, eventWithTimestamp, type } = pending;
+
+    if (type === 'screen_view' && event.session_id) {
+      const sessionKey = this.getLastScreenViewKeyBySession(event.session_id);
+      const profileKey = event.profile_id
+        ? this.getLastScreenViewKeyByProfile(event.project_id, event.profile_id)
+        : '';
+
+      this.evalScript(
+        multi,
+        'addScreenView',
+        this.addScreenViewScript,
+        4,
+        sessionKey,
+        profileKey,
+        this.queueKey,
+        this.bufferCounterKey,
+        eventWithTimestamp!,
+        '3600',
+      );
+    } else if (type === 'session_end' && event.session_id) {
+      const sessionKey = this.getLastScreenViewKeyBySession(event.session_id);
+      const profileKey = event.profile_id
+        ? this.getLastScreenViewKeyByProfile(event.project_id, event.profile_id)
+        : '';
+
+      this.evalScript(
+        multi,
+        'addSessionEnd',
+        this.addSessionEndScript,
+        4,
+        sessionKey,
+        profileKey,
+        this.queueKey,
+        this.bufferCounterKey,
+        eventJson,
+      );
+    } else {
+      multi.rpush(this.queueKey, eventJson).incr(this.bufferCounterKey);
+    }
+
+    if (event.profile_id) {
+      this.incrementActiveVisitorCount(
+        multi,
+        event.project_id,
+        event.profile_id,
+      );
+    }
+  }
+
+  public async flush() {
+    if (this.flushTimer) {
+      clearTimeout(this.flushTimer);
+      this.flushTimer = null;
+    }
+    await this.flushLocalBuffer();
+  }
+
+  private async flushLocalBuffer() {
+    if (this.isFlushing || this.pendingEvents.length === 0) {
+      return;
+    }
+
+    this.isFlushing = true;
+
+    const eventsToFlush = this.pendingEvents;
+    this.pendingEvents = [];
+
     try {
       const redis = getRedisCache();
-      const eventJson = JSON.stringify(event);
-      const multi = _multi || redis.multi();
-
-      if (event.session_id && event.name === 'screen_view') {
-        // Handle screen_view
-        const sessionKey = this.getLastScreenViewKeyBySession(event.session_id);
-        const profileKey = event.profile_id
-          ? this.getLastScreenViewKeyByProfile(
-              event.project_id,
-              event.profile_id,
-            )
-          : '';
-        const timestamp = new Date(event.created_at || Date.now()).getTime();
-
-        // Combine event and timestamp into single JSON for atomic operations
-        const eventWithTimestamp = JSON.stringify({
-          event: event,
-          ts: timestamp,
-        });
-
-        this.evalScript(
-          multi,
-          'addScreenView',
-          this.addScreenViewScript,
-          4,
-          sessionKey,
-          profileKey,
-          this.queueKey,
-          this.bufferCounterKey,
-          eventWithTimestamp,
-          '3600', // 1 hour TTL
-        );
-      } else if (event.session_id && event.name === 'session_end') {
-        // Handle session_end
-        const sessionKey = this.getLastScreenViewKeyBySession(event.session_id);
-        const profileKey = event.profile_id
-          ? this.getLastScreenViewKeyByProfile(
-              event.project_id,
-              event.profile_id,
-            )
-          : '';
-
-        this.evalScript(
-          multi,
-          'addSessionEnd',
-          this.addSessionEndScript,
-          4,
-          sessionKey,
-          profileKey,
-          this.queueKey,
-          this.bufferCounterKey,
-          eventJson,
-        );
-      } else {
-        // All other events go directly to queue
-        multi.rpush(this.queueKey, eventJson).incr(this.bufferCounterKey);
-      }
+      const multi = redis.multi();
 
-      if (event.profile_id) {
-        this.incrementActiveVisitorCount(
-          multi,
-          event.project_id,
-          event.profile_id,
-        );
+      for (const pending of eventsToFlush) {
+        this.addToMulti(multi, pending);
       }
 
-      if (!_multi) {
-        await multi.exec();
-      }
+      await multi.exec();
 
-      await publishEvent('events', 'received', transformEvent(event));
+      this.flushRetryCount = 0;
+      this.pruneHeartbeatMap();
     } catch (error) {
-      this.logger.error('Failed to add event to Redis buffer', { error });
+      // Re-queue failed events at the front to preserve order and avoid data loss
+      this.pendingEvents = eventsToFlush.concat(this.pendingEvents);
+
+      this.flushRetryCount += 1;
+      this.logger.warn('Failed to flush local buffer to Redis; events re-queued', {
+        error,
+        eventCount: eventsToFlush.length,
+        flushRetryCount: this.flushRetryCount,
+      });
+    } finally {
+      this.isFlushing = false;
+      // Events may have accumulated while we were flushing; schedule another flush if needed
+      if (this.pendingEvents.length > 0 && !this.flushTimer) {
+        this.flushTimer = setTimeout(() => {
+          this.flushTimer = null;
+          this.flushLocalBuffer();
+        }, this.microBatchIntervalMs);
+      }
     }
   }
 
-  /**
-   * Execute a Lua script using EVALSHA (cached) or fallback to EVAL.
-   * This avoids sending the entire script on every call.
-   */
   private evalScript(
     multi: ReturnType<Redis['multi']>,
     scriptName: keyof typeof this.scriptShas,
@@ -299,32 +341,18 @@ return added
     const sha = this.scriptShas[scriptName];
 
     if (sha) {
-      // Use EVALSHA with cached SHA
       multi.evalsha(sha, numKeys, ...args);
     } else {
-      // Fallback to EVAL and try to reload script
       multi.eval(scriptContent, numKeys, ...args);
       this.logger.warn(`Script ${scriptName} not loaded, using EVAL fallback`);
-      // Attempt to reload scripts in background
       this.loadScripts();
     }
   }
 
-  /**
-   * Process the Redis buffer - simplified version.
-   *
-   * Simply:
-   * 1. Fetch events from the queue (up to batchSize)
-   * 2. Parse and sort them
-   * 3. Insert into ClickHouse in chunks
-   * 4. Publish saved events
-   * 5. Clean up processed events from queue
-   */
   async processBuffer() {
     const redis = getRedisCache();
 
     try {
-      // Fetch events from queue
       const queueEvents = await redis.lrange(
         this.queueKey,
         0,
@@ -336,7 +364,6 @@ return added
         return;
       }
 
-      // Parse events
       const eventsToClickhouse: IClickhouseEvent[] = [];
       for (const eventStr of queueEvents) {
         const event = getSafeJson<IClickhouseEvent>(eventStr);
@@ -350,14 +377,12 @@ return added
         return;
       }
 
-      // Sort events by creation time
       eventsToClickhouse.sort(
         (a, b) =>
           new Date(a.created_at || 0).getTime() -
           new Date(b.created_at || 0).getTime(),
       );
 
-      // Insert events into ClickHouse in chunks
       this.logger.info('Inserting events into ClickHouse', {
         totalEvents: eventsToClickhouse.length,
         chunks: Math.ceil(eventsToClickhouse.length / this.chunkSize),
@@ -371,14 +396,14 @@ return added
         });
       }
 
-      // Publish "saved" events
-      const pubMulti = getRedisPub().multi();
+      const countByProject = new Map<string, number>();
       for (const event of eventsToClickhouse) {
-        await publishEvent('events', 'saved', transformEvent(event), pubMulti);
+        countByProject.set(event.project_id, (countByProject.get(event.project_id) ?? 0) + 1);
+      }
+      for (const [projectId, count] of countByProject) {
+        publishEvent('events', 'batch', { projectId, count });
       }
-      await pubMulti.exec();
 
-      // Clean up processed events from queue
       await redis
         .multi()
         .ltrim(this.queueKey, queueEvents.length, -1)
@@ -394,9 +419,6 @@ return added
     }
   }
 
-  /**
-   * Retrieve the latest screen_view event for a given session or profile
-   */
   public async getLastScreenView(
     params:
       | {
@@ -440,16 +462,32 @@ return added
     });
   }
 
-  private async incrementActiveVisitorCount(
+  private pruneHeartbeatMap() {
+    const cutoff = Date.now() - this.activeVisitorsExpiration * 1000;
+    for (const [key, ts] of this.lastHeartbeat) {
+      if (ts < cutoff) {
+        this.lastHeartbeat.delete(key);
+      }
+    }
+  }
+
+  private incrementActiveVisitorCount(
     multi: ReturnType<Redis['multi']>,
     projectId: string,
     profileId: string,
   ) {
-    // Track active visitors and emit expiry events when inactive for TTL
+    const key = `${projectId}:${profileId}`;
     const now = Date.now();
+    const last = this.lastHeartbeat.get(key) ?? 0;
+
+    if (now - last < this.heartbeatRefreshMs) {
+      return;
+    }
+
+    this.lastHeartbeat.set(key, now);
     const zsetKey = `live:visitors:${projectId}`;
     const heartbeatKey = `live:visitor:${projectId}:${profileId}`;
-    return multi
+    multi
       .zadd(zsetKey, now, profileId)
       .set(heartbeatKey, '1', 'EX', this.activeVisitorsExpiration);
   }
diff --git a/packages/db/src/buffers/profile-buffer.ts b/packages/db/src/buffers/profile-buffer.ts
index de80bd422..d568bb23c 100644
--- a/packages/db/src/buffers/profile-buffer.ts
+++ b/packages/db/src/buffers/profile-buffer.ts
@@ -1,11 +1,11 @@
 import { deepMergeObjects } from '@openpanel/common';
 import { getSafeJson } from '@openpanel/json';
 import type { ILogger } from '@openpanel/logger';
-import { type Redis, getRedisCache } from '@openpanel/redis';
+import { getRedisCache, type Redis } from '@openpanel/redis';
 import shallowEqual from 'fast-deep-equal';
 import { omit } from 'ramda';
 import sqlstring from 'sqlstring';
-import { TABLE_NAMES, ch, chQuery } from '../clickhouse/client';
+import { ch, chQuery, TABLE_NAMES } from '../clickhouse/client';
 import type { IClickhouseProfile } from '../services/profile.service';
 import { BaseBuffer } from './base-buffer';
 
@@ -89,7 +89,7 @@ export class ProfileBuffer extends BaseBuffer {
             'os_version',
             'browser_version',
           ],
-          profile.properties,
+          profile.properties
         );
       }
 
@@ -97,16 +97,16 @@ export class ProfileBuffer extends BaseBuffer {
         ? deepMergeObjects(existingProfile, omit(['created_at'], profile))
         : profile;
 
-      if (profile && existingProfile) {
-        if (
-          shallowEqual(
-            omit(['created_at'], existingProfile),
-            omit(['created_at'], mergedProfile),
-          )
-        ) {
-          this.logger.debug('Profile not changed, skipping');
-          return;
-        }
+      if (
+        profile &&
+        existingProfile &&
+        shallowEqual(
+          omit(['created_at'], existingProfile),
+          omit(['created_at'], mergedProfile)
+        )
+      ) {
+        this.logger.debug('Profile not changed, skipping');
+        return;
       }
 
       this.logger.debug('Merged profile will be inserted', {
@@ -151,11 +151,11 @@ export class ProfileBuffer extends BaseBuffer {
 
   private async fetchProfile(
     profile: IClickhouseProfile,
-    logger: ILogger,
+    logger: ILogger
   ): Promise<IClickhouseProfile | null> {
     const existingProfile = await this.fetchFromCache(
       profile.id,
-      profile.project_id,
+      profile.project_id
     );
     if (existingProfile) {
       logger.debug('Profile found in Redis');
@@ -167,7 +167,7 @@ export class ProfileBuffer extends BaseBuffer {
 
   public async fetchFromCache(
     profileId: string,
-    projectId: string,
+    projectId: string
   ): Promise<IClickhouseProfile | null> {
     const cacheKey = this.getProfileCacheKey({
       profileId,
@@ -182,7 +182,7 @@ export class ProfileBuffer extends BaseBuffer {
 
   private async fetchFromClickhouse(
     profile: IClickhouseProfile,
-    logger: ILogger,
+    logger: ILogger
   ): Promise<IClickhouseProfile | null> {
     logger.debug('Fetching profile from Clickhouse');
     const result = await chQuery<IClickhouseProfile>(
@@ -207,7 +207,7 @@ export class ProfileBuffer extends BaseBuffer {
         }
       GROUP BY id, project_id 
       ORDER BY created_at DESC 
-      LIMIT 1`,
+      LIMIT 1`
     );
     logger.debug('Clickhouse fetch result', {
       found: !!result[0],
@@ -221,7 +221,7 @@ export class ProfileBuffer extends BaseBuffer {
       const profiles = await this.redis.lrange(
         this.redisKey,
         0,
-        this.batchSize - 1,
+        this.batchSize - 1
       );
 
       if (profiles.length === 0) {
@@ -231,7 +231,7 @@ export class ProfileBuffer extends BaseBuffer {
 
       this.logger.debug(`Processing ${profiles.length} profiles in buffer`);
       const parsedProfiles = profiles.map((p) =>
-        getSafeJson<IClickhouseProfile>(p),
+        getSafeJson<IClickhouseProfile>(p)
       );
 
       for (const chunk of this.chunks(parsedProfiles, this.chunkSize)) {
diff --git a/packages/db/src/services/clients.service.ts b/packages/db/src/services/clients.service.ts
index ea5eab17b..6566d219f 100644
--- a/packages/db/src/services/clients.service.ts
+++ b/packages/db/src/services/clients.service.ts
@@ -1,4 +1,4 @@
-import { cacheable, cacheableLru } from '@openpanel/redis';
+import { cacheable } from '@openpanel/redis';
 import type { Client, Prisma } from '../prisma-client';
 import { db } from '../prisma-client';
 
@@ -34,7 +34,4 @@ export async function getClientById(
   });
 }
 
-export const getClientByIdCached = cacheableLru(getClientById, {
-  maxSize: 1000,
-  ttl: 60 * 5,
-});
+export const getClientByIdCached = cacheable(getClientById, 60 * 5);
diff --git a/packages/db/src/services/notification.service.ts b/packages/db/src/services/notification.service.ts
index 57d988f7f..b1cbc1a50 100644
--- a/packages/db/src/services/notification.service.ts
+++ b/packages/db/src/services/notification.service.ts
@@ -90,7 +90,7 @@ export const getNotificationRulesByProjectId = cacheable(
       },
     });
   },
-  60 * 24
+  60 * 24,
 );
 
 function getIntegration(integrationId: string | null) {
diff --git a/packages/db/src/services/project.service.ts b/packages/db/src/services/project.service.ts
index 7e8a997a4..3144adcf9 100644
--- a/packages/db/src/services/project.service.ts
+++ b/packages/db/src/services/project.service.ts
@@ -1,6 +1,6 @@
 import { cacheable } from '@openpanel/redis';
 import sqlstring from 'sqlstring';
-import { TABLE_NAMES, chQuery } from '../clickhouse/client';
+import { chQuery, TABLE_NAMES } from '../clickhouse/client';
 import type { Prisma, Project } from '../prisma-client';
 import { db } from '../prisma-client';
 
@@ -25,6 +25,7 @@ export async function getProjectById(id: string) {
   return res;
 }
 
+/** L1 LRU (60s) + L2 Redis. clear() invalidates Redis + local LRU; other nodes may serve stale from LRU for up to 60s. */
 export const getProjectByIdCached = cacheable(getProjectById, 60 * 60 * 24);
 
 export async function getProjectWithClients(id: string) {
@@ -44,7 +45,7 @@ export async function getProjectWithClients(id: string) {
   return res;
 }
 
-export async function getProjectsByOrganizationId(organizationId: string) {
+export function getProjectsByOrganizationId(organizationId: string) {
   return db.project.findMany({
     where: {
       organizationId,
@@ -95,7 +96,7 @@ export async function getProjects({
 
   if (access.length > 0) {
     return projects.filter((project) =>
-      access.some((a) => a.projectId === project.id),
+      access.some((a) => a.projectId === project.id)
     );
   }
 
@@ -104,7 +105,7 @@ export async function getProjects({
 
 export const getProjectEventsCount = async (projectId: string) => {
   const res = await chQuery<{ count: number }>(
-    `SELECT count(*) as count FROM ${TABLE_NAMES.events} WHERE project_id = ${sqlstring.escape(projectId)} AND name NOT IN ('session_start', 'session_end')`,
+    `SELECT count(*) as count FROM ${TABLE_NAMES.events} WHERE project_id = ${sqlstring.escape(projectId)} AND name NOT IN ('session_start', 'session_end')`
   );
   return res[0]?.count;
 };
diff --git a/packages/db/src/services/salt.service.ts b/packages/db/src/services/salt.service.ts
index ce3b9f594..424b3e6af 100644
--- a/packages/db/src/services/salt.service.ts
+++ b/packages/db/src/services/salt.service.ts
@@ -1,9 +1,9 @@
 import { generateSalt } from '@openpanel/common/server';
 
-import { cacheableLru } from '@openpanel/redis';
+import { cacheable } from '@openpanel/redis';
 import { db } from '../prisma-client';
 
-export const getSalts = cacheableLru(
+export const getSalts = cacheable(
   'op:salt',
   async () => {
     const [curr, prev] = await db.salt.findMany({
@@ -24,10 +24,7 @@ export const getSalts = cacheableLru(
 
     return salts;
   },
-  {
-    maxSize: 2,
-    ttl: 60 * 5,
-  },
+  60 * 5,
 );
 
 export async function createInitialSalts() {
diff --git a/packages/payments/scripts/assign-product-to-org.ts b/packages/payments/scripts/assign-product-to-org.ts
deleted file mode 100644
index 862c8ee39..000000000
--- a/packages/payments/scripts/assign-product-to-org.ts
+++ /dev/null
@@ -1,221 +0,0 @@
-import { db } from '@openpanel/db';
-import { Polar } from '@polar-sh/sdk';
-import inquirer from 'inquirer';
-import inquirerAutocomplete from 'inquirer-autocomplete-prompt';
-import { getSuccessUrl } from '..';
-
-// Register the autocomplete prompt
-inquirer.registerPrompt('autocomplete', inquirerAutocomplete);
-
-interface Answers {
-  isProduction: boolean;
-  polarApiKey: string;
-  productId: string;
-  organizationId: string;
-}
-
-async function promptForInput() {
-  // Get all organizations first
-  const organizations = await db.organization.findMany({
-    select: {
-      id: true,
-      name: true,
-    },
-  });
-
-  // Step 1: Collect Polar credentials first
-  const polarCredentials = await inquirer.prompt<{
-    isProduction: boolean;
-    polarApiKey: string;
-    polarOrganizationId: string;
-  }>([
-    {
-      type: 'list',
-      name: 'isProduction',
-      message: 'Is this for production?',
-      choices: [
-        { name: 'Yes', value: true },
-        { name: 'No', value: false },
-      ],
-      default: true,
-    },
-    {
-      type: 'string',
-      name: 'polarApiKey',
-      message: 'Enter your Polar API key:',
-      validate: (input: string) => {
-        if (!input) return 'API key is required';
-        return true;
-      },
-    },
-  ]);
-
-  // Step 2: Initialize Polar client and fetch products
-  const polar = new Polar({
-    accessToken: polarCredentials.polarApiKey,
-    server: polarCredentials.isProduction ? 'production' : 'sandbox',
-  });
-
-  console.log('Fetching products from Polar...');
-  const productsResponse = await polar.products.list({
-    limit: 100,
-    isArchived: false,
-    sorting: ['price_amount'],
-  });
-
-  const products = productsResponse.result.items;
-
-  if (products.length === 0) {
-    throw new Error('No products found in Polar');
-  }
-
-  // Step 3: Continue with product selection and organization selection
-  const restOfAnswers = await inquirer.prompt<{
-    productId: string;
-    organizationId: string;
-  }>([
-    {
-      type: 'autocomplete',
-      name: 'productId',
-      message: 'Select product:',
-      source: (answersSoFar: any, input = '') => {
-        return products
-          .filter(
-            (product) =>
-              product.name.toLowerCase().includes(input.toLowerCase()) ||
-              product.id.toLowerCase().includes(input.toLowerCase()),
-          )
-          .map((product) => {
-            const price = product.prices?.[0];
-            const priceStr =
-              price && 'priceAmount' in price && price.priceAmount
-                ? `$${(price.priceAmount / 100).toFixed(2)}/${price.recurringInterval || 'month'}`
-                : 'No price';
-            return {
-              name: `${product.name} (${priceStr})`,
-              value: product.id,
-            };
-          });
-      },
-    },
-    {
-      type: 'autocomplete',
-      name: 'organizationId',
-      message: 'Select organization:',
-      source: (answersSoFar: any, input = '') => {
-        return organizations
-          .filter(
-            (org) =>
-              org.name.toLowerCase().includes(input.toLowerCase()) ||
-              org.id.toLowerCase().includes(input.toLowerCase()),
-          )
-          .map((org) => ({
-            name: `${org.name} (${org.id})`,
-            value: org.id,
-          }));
-      },
-    },
-  ]);
-
-  return {
-    ...polarCredentials,
-    ...restOfAnswers,
-  };
-}
-
-async function main() {
-  console.log('Assigning existing product to organization...');
-  const input = await promptForInput();
-
-  const polar = new Polar({
-    accessToken: input.polarApiKey,
-    server: input.isProduction ? 'production' : 'sandbox',
-  });
-
-  const organization = await db.organization.findUniqueOrThrow({
-    where: {
-      id: input.organizationId,
-    },
-    select: {
-      id: true,
-      name: true,
-      createdBy: {
-        select: {
-          id: true,
-          email: true,
-          firstName: true,
-          lastName: true,
-        },
-      },
-      projects: {
-        select: {
-          id: true,
-        },
-      },
-    },
-  });
-
-  if (!organization.createdBy) {
-    throw new Error(
-      `Organization ${organization.name} does not have a creator. Cannot proceed.`,
-    );
-  }
-
-  const user = organization.createdBy;
-
-  // Fetch product details for review
-  const product = await polar.products.get({ id: input.productId });
-  const price = product.prices?.[0];
-  const priceStr =
-    price && 'priceAmount' in price && price.priceAmount
-      ? `$${(price.priceAmount / 100).toFixed(2)}/${price.recurringInterval || 'month'}`
-      : 'No price';
-
-  console.log('\nReview the following settings:');
-  console.table({
-    product: product.name,
-    price: priceStr,
-    organization: organization.name,
-    email: user.email,
-    name:
-      [user.firstName, user.lastName].filter(Boolean).join(' ') || 'No name',
-  });
-
-  const { confirmed } = await inquirer.prompt([
-    {
-      type: 'confirm',
-      name: 'confirmed',
-      message: 'Do you want to proceed?',
-      default: false,
-    },
-  ]);
-
-  if (!confirmed) {
-    console.log('Operation canceled');
-    return;
-  }
-
-  const checkoutLink = await polar.checkoutLinks.create({
-    paymentProcessor: 'stripe',
-    productId: input.productId,
-    allowDiscountCodes: false,
-    metadata: {
-      organizationId: organization.id,
-      userId: user.id,
-    },
-    successUrl: getSuccessUrl(
-      input.isProduction
-        ? 'https://dashboard.openpanel.dev'
-        : 'http://localhost:3000',
-      organization.id,
-    ),
-  });
-
-  console.log('\nCheckout link created:');
-  console.table(checkoutLink);
-  console.log('\nProduct assigned successfully!');
-}
-
-main()
-  .catch(console.error)
-  .finally(() => db.$disconnect());
diff --git a/packages/queue/src/queues.ts b/packages/queue/src/queues.ts
index aa769b486..48a5893e1 100644
--- a/packages/queue/src/queues.ts
+++ b/packages/queue/src/queues.ts
@@ -6,7 +6,7 @@ import type {
 } from '@openpanel/db';
 import { createLogger } from '@openpanel/logger';
 import { getRedisGroupQueue, getRedisQueue } from '@openpanel/redis';
-import { Queue, QueueEvents } from 'bullmq';
+import { Queue } from 'bullmq';
 import { Queue as GroupQueue } from 'groupmq';
 import type { ITrackPayload } from '../../validation';
 
@@ -66,6 +66,10 @@ export interface EventsQueuePayloadIncomingEvent {
     headers: Record<string, string | undefined>;
     deviceId: string;
     sessionId: string;
+    session?: Pick<
+      IServiceCreateEventPayload,
+      'referrer' | 'referrerName' | 'referrerType'
+    >;
   };
 }
 export interface EventsQueuePayloadCreateEvent {
@@ -206,9 +210,6 @@ export const sessionsQueue = new Queue<SessionsQueuePayload>(
     },
   }
 );
-export const sessionsQueueEvents = new QueueEvents(getQueueName('sessions'), {
-  connection: getRedisQueue(),
-});
 
 export const cronQueue = new Queue<CronQueuePayload>(getQueueName('cron'), {
   connection: getRedisQueue(),
diff --git a/packages/redis/cachable.ts b/packages/redis/cachable.ts
index 3878af657..d77ec3a01 100644
--- a/packages/redis/cachable.ts
+++ b/packages/redis/cachable.ts
@@ -1,7 +1,7 @@
 import { LRUCache } from 'lru-cache';
 import { getRedisCache } from './redis';
 
-export const deleteCache = async (key: string) => {
+export const deleteCache = (key: string) => {
   return getRedisCache().del(key);
 };
 
@@ -15,7 +15,7 @@ export async function getCache<T>(
   key: string,
   expireInSec: number,
   fn: () => Promise<T>,
-  useLruCache?: boolean,
+  useLruCache?: boolean
 ): Promise<T> {
   // L1 Cache: Check global LRU cache first (in-memory, instant)
   if (useLruCache) {
@@ -28,15 +28,7 @@ export async function getCache<T>(
   // L2 Cache: Check Redis cache (shared across instances)
   const hit = await getRedisCache().get(key);
   if (hit) {
-    const parsed = JSON.parse(hit, (_, value) => {
-      if (
-        typeof value === 'string' &&
-        /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.*Z$/.test(value)
-      ) {
-        return new Date(value);
-      }
-      return value;
-    });
+    const parsed = parseCache(hit);
 
     // Store in LRU cache for next time
     if (useLruCache) {
@@ -81,12 +73,24 @@ export function getGlobalLruCacheStats() {
 }
 
 function stringify(obj: unknown): string {
-  if (obj === null) return 'null';
-  if (obj === undefined) return 'undefined';
-  if (typeof obj === 'boolean') return obj ? 'true' : 'false';
-  if (typeof obj === 'number') return String(obj);
-  if (typeof obj === 'string') return obj;
-  if (typeof obj === 'function') return obj.toString();
+  if (obj === null) {
+    return 'null';
+  }
+  if (obj === undefined) {
+    return 'undefined';
+  }
+  if (typeof obj === 'boolean') {
+    return obj ? 'true' : 'false';
+  }
+  if (typeof obj === 'number') {
+    return String(obj);
+  }
+  if (typeof obj === 'string') {
+    return obj;
+  }
+  if (typeof obj === 'function') {
+    return obj.toString();
+  }
 
   if (Array.isArray(obj)) {
     return `[${obj.map(stringify).join(',')}]`;
@@ -128,17 +132,29 @@ function hasResult(result: unknown): boolean {
   return true;
 }
 
-export interface CacheableLruOptions {
-  /** TTL in seconds for LRU cache */
-  ttl: number;
-  /** Maximum number of entries in LRU cache */
-  maxSize?: number;
-}
+const DATE_REGEX = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.*Z$/;
+const parseCache = (cached: string) => {
+  try {
+    return JSON.parse(cached, (_, value) => {
+      if (typeof value === 'string' && DATE_REGEX.test(value)) {
+        return new Date(value);
+      }
+      return value;
+    });
+  } catch (error) {
+    console.error('Failed to parse cache', error);
+    return null;
+  }
+};
+
+// L1 cache: short TTL to offload Redis; clear() invalidates Redis, other nodes may serve stale from LRU for up to this long
+const CACHEABLE_LRU_TTL_MS = 60 * 1000; // 60 seconds
+const CACHEABLE_LRU_MAX = 1000;
 
 // Overload 1: cacheable(fn, expireInSec)
 export function cacheable<T extends (...args: any) => any>(
   fn: T,
-  expireInSec: number,
+  expireInSec: number
 ): T & {
   getKey: (...args: Parameters<T>) => string;
   clear: (...args: Parameters<T>) => Promise<number>;
@@ -151,7 +167,7 @@ export function cacheable<T extends (...args: any) => any>(
 export function cacheable<T extends (...args: any) => any>(
   name: string,
   fn: T,
-  expireInSec: number,
+  expireInSec: number
 ): T & {
   getKey: (...args: Parameters<T>) => string;
   clear: (...args: Parameters<T>) => Promise<number>;
@@ -164,7 +180,7 @@ export function cacheable<T extends (...args: any) => any>(
 export function cacheable<T extends (...args: any) => any>(
   fnOrName: T | string,
   fnOrExpireInSec: number | T,
-  _expireInSec?: number,
+  _expireInSec?: number
 ) {
   const name = typeof fnOrName === 'string' ? fnOrName : fnOrName.name;
   const fn =
@@ -195,184 +211,67 @@ export function cacheable<T extends (...args: any) => any>(
 
   const cachePrefix = `cachable:${name}`;
   const getKey = (...args: Parameters<T>) =>
-    `${cachePrefix}:${stringify(args)}`;
+    `${cachePrefix}:${stringify(args)}`.replaceAll(/\s/g, '');
 
-  // Redis-only mode: asynchronous implementation
+  const lruCache = new LRUCache<string, any>({
+    max: CACHEABLE_LRU_MAX,
+    ttl: CACHEABLE_LRU_TTL_MS,
+  });
+
+  // L1 LRU (60s) + L2 Redis. clear() deletes Redis + local LRU; other nodes may serve stale from LRU for up to 60s.
   const cachedFn = async (
     ...args: Parameters<T>
   ): Promise<Awaited<ReturnType<T>>> => {
     const key = getKey(...args);
 
-    // Check Redis cache (shared across instances)
+    // L1: in-memory LRU first (offloads Redis on hot keys)
+    const lruHit = lruCache.get(key);
+    if (lruHit !== undefined && hasResult(lruHit)) {
+      return lruHit as Awaited<ReturnType<T>>;
+    }
+
+    // L2: Redis (shared across instances)
     const cached = await getRedisCache().get(key);
     if (cached) {
-      try {
-        const parsed = JSON.parse(cached, (_, value) => {
-          if (
-            typeof value === 'string' &&
-            /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.*Z$/.test(value)
-          ) {
-            return new Date(value);
-          }
-          return value;
-        });
-        if (hasResult(parsed)) {
-          return parsed;
-        }
-      } catch (e) {
-        console.error('Failed to parse cache', e);
+      const parsed = parseCache(cached);
+      if (hasResult(parsed)) {
+        lruCache.set(key, parsed);
+        return parsed;
       }
     }
 
-    // Cache miss: Execute function
+    // Cache miss: execute function
     const result = await fn(...(args as any));
 
     if (hasResult(result)) {
-      // Don't await Redis write - fire and forget for better performance
+      lruCache.set(key, result);
       getRedisCache()
         .setex(key, expireInSec, JSON.stringify(result))
-        .catch(() => {});
-    }
-
-    return result;
-  };
-
-  cachedFn.getKey = getKey;
-  cachedFn.clear = async (...args: Parameters<T>) => {
-    const key = getKey(...args);
-    return getRedisCache().del(key);
-  };
-  cachedFn.set =
-    (...args: Parameters<T>) =>
-    async (payload: Awaited<ReturnType<T>>) => {
-      const key = getKey(...args);
-      return getRedisCache()
-        .setex(key, expireInSec, JSON.stringify(payload))
-        .catch(() => {});
-    };
-
-  return cachedFn;
-}
-
-// Overload 1: cacheableLru(fn, options)
-export function cacheableLru<T extends (...args: any) => any>(
-  fn: T,
-  options: CacheableLruOptions,
-): T & {
-  getKey: (...args: Parameters<T>) => string;
-  clear: (...args: Parameters<T>) => boolean;
-  set: (...args: Parameters<T>) => (payload: ReturnType<T>) => void;
-};
-
-// Overload 2: cacheableLru(name, fn, options)
-export function cacheableLru<T extends (...args: any) => any>(
-  name: string,
-  fn: T,
-  options: CacheableLruOptions,
-): T & {
-  getKey: (...args: Parameters<T>) => string;
-  clear: (...args: Parameters<T>) => boolean;
-  set: (...args: Parameters<T>) => (payload: ReturnType<T>) => void;
-};
-
-// Implementation for cacheableLru (LRU-only - synchronous)
-export function cacheableLru<T extends (...args: any) => any>(
-  fnOrName: T | string,
-  fnOrOptions: T | CacheableLruOptions,
-  _options?: CacheableLruOptions,
-) {
-  const name = typeof fnOrName === 'string' ? fnOrName : fnOrName.name;
-  const fn =
-    typeof fnOrName === 'function'
-      ? fnOrName
-      : typeof fnOrOptions === 'function'
-        ? fnOrOptions
-        : null;
-
-  let options: CacheableLruOptions;
-
-  // Parse parameters based on function signature
-  if (typeof fnOrName === 'function') {
-    // Overload 1: cacheableLru(fn, options)
-    options =
-      typeof fnOrOptions === 'object' && fnOrOptions !== null
-        ? fnOrOptions
-        : ({} as CacheableLruOptions);
-  } else {
-    // Overload 2: cacheableLru(name, fn, options)
-    options =
-      typeof _options === 'object' && _options !== null
-        ? _options
-        : ({} as CacheableLruOptions);
-  }
-
-  if (typeof fn !== 'function') {
-    throw new Error('fn is not a function');
-  }
-
-  if (typeof options.ttl !== 'number') {
-    throw new Error('options.ttl is required and must be a number');
-  }
-
-  const cachePrefix = `cachable:${name}`;
-  const getKey = (...args: Parameters<T>) =>
-    `${cachePrefix}:${stringify(args)}`;
-
-  const maxSize = options.maxSize ?? 1000;
-  const ttl = options.ttl;
-
-  // Create function-specific LRU cache
-  const functionLruCache = new LRUCache<string, any>({
-    max: maxSize,
-    ttl: ttl * 1000, // Convert seconds to milliseconds for LRU
-  });
-
-  // LRU-only mode: synchronous implementation (or returns promise if fn is async)
-  const cachedFn = ((...args: Parameters<T>): ReturnType<T> => {
-    const key = getKey(...args);
-
-    // Check LRU cache
-    const lruHit = functionLruCache.get(key);
-    if (lruHit !== undefined && hasResult(lruHit)) {
-      return lruHit as ReturnType<T>;
-    }
-
-    // Cache miss: Execute function
-    const result = fn(...(args as any)) as ReturnType<T>;
-
-    // If result is a Promise, handle it asynchronously but cache the resolved value
-    if (result && typeof (result as any).then === 'function') {
-      return (result as Promise<any>).then((resolved: any) => {
-        if (hasResult(resolved)) {
-          functionLruCache.set(key, resolved);
-        }
-        return resolved;
-      }) as ReturnType<T>;
-    }
-
-    // Synchronous result: cache and return
-    if (hasResult(result)) {
-      functionLruCache.set(key, result);
+        .catch(() => {
+          // ignore error
+        });
     }
 
     return result;
-  }) as T & {
-    getKey: (...args: Parameters<T>) => string;
-    clear: (...args: Parameters<T>) => boolean;
-    set: (...args: Parameters<T>) => (payload: ReturnType<T>) => void;
   };
 
   cachedFn.getKey = getKey;
   cachedFn.clear = (...args: Parameters<T>) => {
     const key = getKey(...args);
-    return functionLruCache.delete(key);
+    lruCache.delete(key);
+    return getRedisCache().del(key);
   };
   cachedFn.set =
     (...args: Parameters<T>) =>
-    (payload: ReturnType<T>) => {
+    (payload: Awaited<ReturnType<T>>) => {
       const key = getKey(...args);
       if (hasResult(payload)) {
-        functionLruCache.set(key, payload);
+        lruCache.set(key, payload);
+        return getRedisCache()
+          .setex(key, expireInSec, JSON.stringify(payload))
+          .catch(() => {
+            // ignore error
+          });
       }
     };
 
diff --git a/packages/redis/publisher.ts b/packages/redis/publisher.ts
index 2bcca054a..47d04f4c1 100644
--- a/packages/redis/publisher.ts
+++ b/packages/redis/publisher.ts
@@ -10,8 +10,7 @@ export type IPublishChannels = {
     };
   };
   events: {
-    received: IServiceEvent;
-    saved: IServiceEvent;
+    batch: { projectId: string; count: number };
   };
   notification: {
     created: Prisma.NotificationUncheckedCreateInput;
diff --git a/packages/sdks/react-native/index.ts b/packages/sdks/react-native/index.ts
index 7aa2baf3c..30a616191 100644
--- a/packages/sdks/react-native/index.ts
+++ b/packages/sdks/react-native/index.ts
@@ -1,13 +1,13 @@
+import type { OpenPanelOptions, TrackProperties } from '@openpanel/sdk';
+import { OpenPanel as OpenPanelBase } from '@openpanel/sdk';
 import * as Application from 'expo-application';
 import Constants from 'expo-constants';
 import { AppState, Platform } from 'react-native';
 
-import type { OpenPanelOptions, TrackProperties } from '@openpanel/sdk';
-import { OpenPanel as OpenPanelBase } from '@openpanel/sdk';
-
 export * from '@openpanel/sdk';
 
 export class OpenPanel extends OpenPanelBase {
+  private lastPath = '';
   constructor(public options: OpenPanelOptions) {
     super({
       ...options,
@@ -37,7 +37,12 @@ export class OpenPanel extends OpenPanelBase {
     });
   }
 
-  public screenView(route: string, properties?: TrackProperties): void {
+  track(name: string, properties?: TrackProperties) {
+    return super.track(name, { ...properties, __path: this.lastPath });
+  }
+
+  screenView(route: string, properties?: TrackProperties): void {
+    this.lastPath = route;
     super.track('screen_view', {
       ...properties,
       __path: route,
diff --git a/packages/sdks/web/src/index.ts b/packages/sdks/web/src/index.ts
index 1446813bc..c15da114a 100644
--- a/packages/sdks/web/src/index.ts
+++ b/packages/sdks/web/src/index.ts
@@ -58,7 +58,7 @@ export type OpenPanelOptions = OpenPanelBaseOptions & {
 
 function toCamelCase(str: string) {
   return str.replace(/([-_][a-z])/gi, ($1) =>
-    $1.toUpperCase().replace('-', '').replace('_', ''),
+    $1.toUpperCase().replace('-', '').replace('_', '')
   );
 }
 
@@ -114,7 +114,9 @@ export class OpenPanel extends OpenPanelBase {
         const sampled = Math.random() < sampleRate;
         if (sampled) {
           this.loadReplayModule().then((mod) => {
-            if (!mod) return;
+            if (!mod) {
+              return;
+            }
             mod.startReplayRecorder(this.options.sessionReplay!, (chunk) => {
               // Replay chunks go through send() and are queued when disabled or waitForProfile
               // until ready() is called (base SDK also queues replay until sessionId is set).
@@ -153,7 +155,10 @@ export class OpenPanel extends OpenPanelBase {
       // dead-code-eliminated in the library build.
       if (typeof __OPENPANEL_REPLAY_URL__ !== 'undefined') {
         const scriptEl = _replayScriptRef;
-        const url = this.options.sessionReplay?.scriptUrl || scriptEl?.src?.replace('.js', '-replay.js') || 'https://openpanel.dev/op1-replay.js';
+        const url =
+          this.options.sessionReplay?.scriptUrl ||
+          scriptEl?.src?.replace('.js', '-replay.js') ||
+          'https://openpanel.dev/op1-replay.js';
 
         // Already loaded (e.g. user included the script manually)
         if ((window as any).__openpanel_replay) {
@@ -287,11 +292,15 @@ export class OpenPanel extends OpenPanelBase {
     });
   }
 
+  track(name: string, properties?: TrackProperties) {
+    return super.track(name, { ...properties, __path: this.lastPath });
+  }
+
   screenView(properties?: TrackProperties): void;
   screenView(path: string, properties?: TrackProperties): void;
   screenView(
     pathOrProperties?: string | TrackProperties,
-    propertiesOrUndefined?: TrackProperties,
+    propertiesOrUndefined?: TrackProperties
   ): void {
     if (this.isServer()) {
       return;
@@ -322,7 +331,7 @@ export class OpenPanel extends OpenPanelBase {
 
   async flushRevenue() {
     const promises = this.pendingRevenues.map((pending) =>
-      super.revenue(pending.amount, pending.properties),
+      super.revenue(pending.amount, pending.properties)
     );
     await Promise.all(promises);
     this.clearRevenue();
@@ -343,7 +352,7 @@ export class OpenPanel extends OpenPanelBase {
       try {
         sessionStorage.setItem(
           'openpanel-pending-revenues',
-          JSON.stringify(this.pendingRevenues),
+          JSON.stringify(this.pendingRevenues)
         );
       } catch {}
     }
diff --git a/packages/trpc/src/routers/project.ts b/packages/trpc/src/routers/project.ts
index 80e650aad..5c515ba5c 100644
--- a/packages/trpc/src/routers/project.ts
+++ b/packages/trpc/src/routers/project.ts
@@ -96,9 +96,7 @@ export const projectRouter = createTRPCRouter({
       });
       await Promise.all([
         getProjectByIdCached.clear(input.id),
-        res.clients.map((client) => {
-          getClientByIdCached.clear(client.id);
-        }),
+        ...res.clients.map((client) => getClientByIdCached.clear(client.id)),
       ]);
       return res;
     }),
diff --git a/packages/trpc/src/routers/realtime.ts b/packages/trpc/src/routers/realtime.ts
index 62229681e..8922f4d4f 100644
--- a/packages/trpc/src/routers/realtime.ts
+++ b/packages/trpc/src/routers/realtime.ts
@@ -1,18 +1,15 @@
-import { z } from 'zod';
-
 import {
-  type EventMeta,
-  TABLE_NAMES,
   ch,
   chQuery,
   clix,
-  db,
   formatClickhouseDate,
-  getEventList,
+  type IClickhouseEvent,
+  TABLE_NAMES,
+  transformEvent,
 } from '@openpanel/db';
-
 import { subMinutes } from 'date-fns';
 import sqlstring from 'sqlstring';
+import { z } from 'zod';
 import { createTRPCRouter, protectedProcedure } from '../trpc';
 
 export const realtimeRouter = createTRPCRouter({
@@ -25,7 +22,7 @@ export const realtimeRouter = createTRPCRouter({
         long: number;
         lat: number;
       }>(
-        `SELECT DISTINCT country, city, longitude as long, latitude as lat FROM ${TABLE_NAMES.events} WHERE project_id = ${sqlstring.escape(input.projectId)} AND created_at >= '${formatClickhouseDate(subMinutes(new Date(), 30))}' ORDER BY created_at DESC`,
+        `SELECT DISTINCT country, city, longitude as long, latitude as lat FROM ${TABLE_NAMES.events} WHERE project_id = ${sqlstring.escape(input.projectId)} AND created_at >= '${formatClickhouseDate(subMinutes(new Date(), 30))}' ORDER BY created_at DESC`
       );
 
       return res;
@@ -33,25 +30,18 @@ export const realtimeRouter = createTRPCRouter({
   activeSessions: protectedProcedure
     .input(z.object({ projectId: z.string() }))
     .query(async ({ input }) => {
-      return getEventList({
-        projectId: input.projectId,
-        take: 30,
-        select: {
-          name: true,
-          path: true,
-          origin: true,
-          referrer: true,
-          referrerName: true,
-          referrerType: true,
-          country: true,
-          device: true,
-          os: true,
-          browser: true,
-          createdAt: true,
-          profile: true,
-          meta: true,
-        },
-      });
+      const rows = await chQuery<IClickhouseEvent>(
+        `SELECT
+          name, session_id, created_at, path, origin, referrer, referrer_name,
+          country, city, region, os, os_version, browser, browser_version,
+          device
+        FROM ${TABLE_NAMES.events}
+        WHERE project_id = ${sqlstring.escape(input.projectId)}
+          AND created_at >= '${formatClickhouseDate(subMinutes(new Date(), 30))}'
+        ORDER BY created_at DESC
+        LIMIT 50`
+      );
+      return rows.map(transformEvent);
     }),
   paths: protectedProcedure
     .input(z.object({ projectId: z.string() }))
@@ -76,7 +66,7 @@ export const realtimeRouter = createTRPCRouter({
         .where(
           'created_at',
           '>=',
-          formatClickhouseDate(subMinutes(new Date(), 30)),
+          formatClickhouseDate(subMinutes(new Date(), 30))
         )
         .groupBy(['path', 'origin'])
         .orderBy('count', 'DESC')
@@ -106,7 +96,7 @@ export const realtimeRouter = createTRPCRouter({
         .where(
           'created_at',
           '>=',
-          formatClickhouseDate(subMinutes(new Date(), 30)),
+          formatClickhouseDate(subMinutes(new Date(), 30))
         )
         .groupBy(['referrer_name'])
         .orderBy('count', 'DESC')
@@ -137,7 +127,7 @@ export const realtimeRouter = createTRPCRouter({
         .where(
           'created_at',
           '>=',
-          formatClickhouseDate(subMinutes(new Date(), 30)),
+          formatClickhouseDate(subMinutes(new Date(), 30))
         )
         .groupBy(['country', 'city'])
         .orderBy('count', 'DESC')
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 8fc079119..68ec0464b 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -16,8 +16,8 @@ catalogs:
       specifier: ^19.2.3
       version: 19.2.3
     groupmq:
-      specifier: 1.1.1-next.2
-      version: 1.1.1-next.2
+      specifier: 2.0.0-next.1
+      version: 2.0.0-next.1
     react:
       specifier: ^19.2.3
       version: 19.2.3
@@ -198,7 +198,7 @@ importers:
         version: 5.0.0
       groupmq:
         specifier: 'catalog:'
-        version: 1.1.1-next.2(ioredis@5.8.2)
+        version: 2.0.0-next.1(ioredis@5.8.2)
       jsonwebtoken:
         specifier: ^9.0.2
         version: 9.0.2
@@ -936,7 +936,7 @@ importers:
         version: 4.18.2
       groupmq:
         specifier: 'catalog:'
-        version: 1.1.1-next.2(ioredis@5.8.2)
+        version: 2.0.0-next.1(ioredis@5.8.2)
       prom-client:
         specifier: ^15.1.3
         version: 15.1.3
@@ -1419,7 +1419,7 @@ importers:
         version: 5.63.0
       groupmq:
         specifier: 'catalog:'
-        version: 1.1.1-next.2(ioredis@5.8.2)
+        version: 2.0.0-next.1(ioredis@5.8.2)
     devDependencies:
       '@openpanel/tsconfig':
         specifier: workspace:*
@@ -13157,11 +13157,11 @@ packages:
 
   glob@7.1.6:
     resolution: {integrity: sha512-LwaxwyZ72Lk7vZINtNNrywX0ZuLyStrdDtabefZKAY5ZGJhVtgdznluResxNmPitE0SAO+O26sWTHeKSI2wMBA==}
-    deprecated: Old versions of glob are not supported, and contain widely publicized security vulnerabilities, which have been fixed in the current version. Please update. Support for old versions may be purchased (at exorbitant rates) by contacting i@izs.me
+    deprecated: Glob versions prior to v9 are no longer supported
 
   glob@7.2.3:
     resolution: {integrity: sha512-nFR0zLpU2YCaRxwoCJvL6UvCH2JFyFVIvwTLsIf21AuHlMskA1hhTdk+LlYJtOlYt9v6dvszD2BGRqBL+iQK9Q==}
-    deprecated: Old versions of glob are not supported, and contain widely publicized security vulnerabilities, which have been fixed in the current version. Please update. Support for old versions may be purchased (at exorbitant rates) by contacting i@izs.me
+    deprecated: Glob versions prior to v9 are no longer supported
 
   glob@9.3.5:
     resolution: {integrity: sha512-e1LleDykUz2Iu+MTYdkSsuWX8lvAjAcs0Xef0lNIu0S2wOAzuTxCJtcd9S3cijlwYF18EsU3rzb8jPVobxDh9Q==}
@@ -13221,8 +13221,8 @@ packages:
     resolution: {integrity: sha512-5gghUc24tP9HRznNpV2+FIoq3xKkj5dTQqf4v0CpdPbFVwFkWoxOM+o+2OC9ZSvjEMTjfmG9QT+gcvggTwW1zw==}
     engines: {node: '>= 10.x'}
 
-  groupmq@1.1.1-next.2:
-    resolution: {integrity: sha512-5gH+P3NfSCjfCLcB2g2TAHCpmQz+rwrQkb+kAyrzB9puZuAHKQVYOUPWKVBRFjY7B9jPRGHrimDO6h9rWKGfMA==}
+  groupmq@2.0.0-next.1:
+    resolution: {integrity: sha512-xcpz29HeXXn0yP/sQTGPPNMLQAZCCrJg3x9kpOAFbtsXki5KVeBsY3mWNBt3Z+YCa9OxwkTFL6tOcrB67z127A==}
     engines: {node: '>=18'}
     peerDependencies:
       ioredis: '>=5'
@@ -34142,7 +34142,7 @@ snapshots:
 
   graphql@15.8.0: {}
 
-  groupmq@1.1.1-next.2(ioredis@5.8.2):
+  groupmq@2.0.0-next.1(ioredis@5.8.2):
     dependencies:
       cron-parser: 4.9.0
       ioredis: 5.8.2
diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml
index 3a8656c9d..8fee52f47 100644
--- a/pnpm-workspace.yaml
+++ b/pnpm-workspace.yaml
@@ -13,4 +13,4 @@ catalog:
   "@types/react-dom": ^19.2.3
   "@types/node": ^24.7.1
   typescript: ^5.9.3
-  groupmq: 1.1.1-next.2
+  groupmq: 2.0.0-next.1