fix:buffers

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

@@ -1,10 +1,7 @@
 import type { WebSocket } from '@fastify/websocket';
 import { eventBuffer } from '@openpanel/db';
 import { setSuperJson } from '@openpanel/json';
-import {
-  psubscribeToPublishedEvent,
-  subscribeToPublishedEvent,
-} from '@openpanel/redis';
+import { subscribeToPublishedEvent } from '@openpanel/redis';
 import { getProjectAccess } from '@openpanel/trpc';
 import { getOrganizationAccess } from '@openpanel/trpc/src/access';
 import type { FastifyRequest } from 'fastify';
@@ -39,19 +36,8 @@ export function wsVisitors(
     }
   );
 
-  const punsubscribe = psubscribeToPublishedEvent(
-    '__keyevent@0__:expired',
-    (key) => {
-      const [, , projectId] = key.split(':');
-      if (projectId === params.projectId) {
-        sendCount();
-      }
-    }
-  );
-
   socket.on('close', () => {
     unsubscribe();
-    punsubscribe();
   });
 }
 

apps/start/src/components/overview/live-counter.tsx

@@ -1,61 +1,25 @@
-import { TooltipComplete } from '@/components/tooltip-complete';
-import { useDebounceState } from '@/hooks/use-debounce-state';
-import useWS from '@/hooks/use-ws';
-import { useTRPC } from '@/integrations/trpc/react';
-import { cn } from '@/utils/cn';
-import { useQuery, useQueryClient } from '@tanstack/react-query';
-import { useEffect, useRef } from 'react';
+import { useQueryClient } from '@tanstack/react-query';
+import { useCallback } from 'react';
 import { toast } from 'sonner';
 import { AnimatedNumber } from '../animated-number';
+import { TooltipComplete } from '@/components/tooltip-complete';
+import { useLiveCounter } from '@/hooks/use-live-counter';
+import { cn } from '@/utils/cn';
 
 export interface LiveCounterProps {
   projectId: string;
   shareId?: string;
 }
 
-const FIFTEEN_SECONDS = 1000 * 30;
-
 export function LiveCounter({ projectId, shareId }: LiveCounterProps) {
-  const trpc = useTRPC();
   const client = useQueryClient();
-  const counter = useDebounceState(0, 1000);
-  const lastRefresh = useRef(Date.now());
-  const query = useQuery(
-    trpc.overview.liveVisitors.queryOptions({
-      projectId,
-      shareId,
-    }),
-  );
-
-  useEffect(() => {
-    if (query.data) {
-      counter.set(query.data);
-    }
-  }, [query.data]);
-
-  useWS<number>(
-    `/live/visitors/${projectId}`,
-    (value) => {
-      if (!Number.isNaN(value)) {
-        counter.set(value);
-        if (Date.now() - lastRefresh.current > FIFTEEN_SECONDS) {
-          lastRefresh.current = Date.now();
-          if (!document.hidden) {
-            toast('Refreshed data');
-            client.refetchQueries({
-              type: 'active',
-            });
-          }
-        }
-      }
-    },
-    {
-      debounce: {
-        delay: 1000,
-        maxWait: 5000,
-      },
-    },
-  );
+  const onRefresh = useCallback(() => {
+    toast('Refreshed data');
+    client.refetchQueries({
+      type: 'active',
+    });
+  }, [client]);
+  const counter = useLiveCounter({ projectId, shareId, onRefresh });
 
   return (
     <TooltipComplete
@@ -66,13 +30,13 @@ export function LiveCounter({ projectId, shareId }: LiveCounterProps) {
           <div
             className={cn(
               'h-3 w-3 animate-ping rounded-full bg-emerald-500 opacity-100 transition-all',
-              counter.debounced === 0 && 'bg-destructive opacity-0',
+              counter.debounced === 0 && 'bg-destructive opacity-0'
             )}
           />
           <div
             className={cn(
-              'absolute left-0 top-0 h-3 w-3 rounded-full bg-emerald-500 transition-all',
-              counter.debounced === 0 && 'bg-destructive',
+              'absolute top-0 left-0 h-3 w-3 rounded-full bg-emerald-500 transition-all',
+              counter.debounced === 0 && 'bg-destructive'
             )}
           />
         </div>

apps/start/src/hooks/use-live-counter.ts

@@ -0,0 +1,81 @@
+import { useQuery, useQueryClient } from '@tanstack/react-query';
+import { useEffect, useRef } from 'react';
+import { useDebounceState } from './use-debounce-state';
+import useWS from './use-ws';
+import { useTRPC } from '@/integrations/trpc/react';
+
+const FIFTEEN_SECONDS = 1000 * 15;
+/** Refetch from API when WS-only updates may be stale (e.g. visitors left). */
+const FALLBACK_STALE_MS = 1000 * 60;
+
+export function useLiveCounter({
+  projectId,
+  shareId,
+  onRefresh,
+}: {
+  projectId: string;
+  shareId?: string;
+  onRefresh?: () => void;
+}) {
+  const trpc = useTRPC();
+  const queryClient = useQueryClient();
+  const counter = useDebounceState(0, 1000);
+  const lastRefresh = useRef(Date.now());
+  const query = useQuery(
+    trpc.overview.liveVisitors.queryOptions({
+      projectId,
+      shareId: shareId ?? undefined,
+    })
+  );
+
+  useEffect(() => {
+    if (query.data) {
+      counter.set(query.data);
+    }
+  }, [query.data]);
+
+  useWS<number>(
+    `/live/visitors/${projectId}`,
+    (value) => {
+      if (!Number.isNaN(value)) {
+        counter.set(value);
+        if (Date.now() - lastRefresh.current > FIFTEEN_SECONDS) {
+          lastRefresh.current = Date.now();
+          if (!document.hidden) {
+            onRefresh?.();
+          }
+        }
+      }
+    },
+    {
+      debounce: {
+        delay: 1000,
+        maxWait: 5000,
+      },
+    }
+  );
+
+  useEffect(() => {
+    const id = setInterval(async () => {
+      if (Date.now() - lastRefresh.current < FALLBACK_STALE_MS) {
+        return;
+      }
+      const data = await queryClient.fetchQuery(
+        trpc.overview.liveVisitors.queryOptions(
+          {
+            projectId,
+            shareId: shareId ?? undefined,
+          },
+          // Default query staleTime is 5m; bypass cache so this reconciliation always hits the API.
+          { staleTime: 0 }
+        )
+      );
+      counter.set(data);
+      lastRefresh.current = Date.now();
+    }, FALLBACK_STALE_MS);
+
+    return () => clearInterval(id);
+  }, [projectId, shareId, trpc, queryClient, counter.set]);
+
+  return counter;
+}

apps/start/src/routes/widget/counter.tsx

@@ -1,12 +1,11 @@
+import { useQuery, useQueryClient } from '@tanstack/react-query';
+import { createFileRoute } from '@tanstack/react-router';
+import { z } from 'zod';
 import { AnimatedNumber } from '@/components/animated-number';
 import { Ping } from '@/components/ping';
-import { useNumber } from '@/hooks/use-numer-formatter';
 import useWS from '@/hooks/use-ws';
 import { useTRPC } from '@/integrations/trpc/react';
 import type { RouterOutputs } from '@/trpc/client';
-import { useQuery, useQueryClient } from '@tanstack/react-query';
-import { createFileRoute } from '@tanstack/react-router';
-import { z } from 'zod';
 
 const widgetSearchSchema = z.object({
   shareId: z.string(),
@@ -20,33 +19,33 @@ export const Route = createFileRoute('/widget/counter')({
 });
 
 function RouteComponent() {
-  const { shareId, limit, color } = Route.useSearch();
+  const { shareId } = Route.useSearch();
   const trpc = useTRPC();
 
   // Fetch widget data
   const { data, isLoading } = useQuery(
-    trpc.widget.counter.queryOptions({ shareId }),
+    trpc.widget.counter.queryOptions({ shareId })
   );
 
   if (isLoading) {
     return (
-      <div className="flex items-center gap-2 px-2 h-8">
+      <div className="flex h-8 items-center gap-2 px-2">
         <Ping />
-        <AnimatedNumber value={0} suffix=" unique visitors" />
+        <AnimatedNumber suffix=" unique visitors" value={0} />
       </div>
     );
   }
 
   if (!data) {
     return (
-      <div className="flex items-center gap-2 px-2 h-8">
+      <div className="flex h-8 items-center gap-2 px-2">
         <Ping className="bg-orange-500" />
-        <AnimatedNumber value={0} suffix=" unique visitors" />
+        <AnimatedNumber suffix=" unique visitors" value={0} />
       </div>
     );
   }
 
-  return <CounterWidget shareId={shareId} data={data} />;
+  return <CounterWidget data={data} shareId={shareId} />;
 }
 
 interface RealtimeWidgetProps {
@@ -57,30 +56,29 @@ interface RealtimeWidgetProps {
 function CounterWidget({ shareId, data }: RealtimeWidgetProps) {
   const trpc = useTRPC();
   const queryClient = useQueryClient();
-  const number = useNumber();
 
   // WebSocket subscription for real-time updates
   useWS<number>(
     `/live/visitors/${data.projectId}`,
-    (res) => {
+    () => {
       if (!document.hidden) {
         queryClient.refetchQueries(
-          trpc.widget.counter.queryFilter({ shareId }),
+          trpc.widget.counter.queryFilter({ shareId })
         );
       }
     },
     {
       debounce: {
         delay: 1000,
-        maxWait: 60000,
+        maxWait: 60_000,
       },
-    },
+    }
   );
 
   return (
-    <div className="flex items-center gap-2 px-2 h-8">
+    <div className="flex h-8 items-center gap-2 px-2">
       <Ping />
-      <AnimatedNumber value={data.counter} suffix=" unique visitors" />
+      <AnimatedNumber suffix=" unique visitors" value={data.counter} />
     </div>
   );
 }