fix: races in incoming calls

backend/Taskfile.yml

@@ -32,7 +32,7 @@ tasks:
   compose-up:
     desc: Start the development databases and related services (local Go server for faster dev)
     cmds:
-      - docker compose -f dev-compose.yml up -d
+      - docker compose -f docker-files/docker-compose.yml up -d
 
   build:
     cmds:
@@ -51,6 +51,14 @@ tasks:
     cmds:
       - docker run -p 1926:1926 --env-file env-files/.env.local --network host --name hopp-backend hopp-backend
 
+  fullstack-up:
+    desc: Build and run the fullstack setup with Redis and backend
+    cmds:
+      # - task: build-image
+      - docker compose -f docker-files/fullstack-compose.yml up -d
+    env:
+      ENV_STACK: local
+
   install-reload-tools:
     desc: Install required development tools (maybe you will need to install manually)
     cmds:

backend/internal/handlers/websocketHandlers.go

@@ -38,6 +38,11 @@ func CreateWSHandler(server *common.ServerState) echo.HandlerFunc {
 		}
 		defer ws.Close()
 
+		// Kill the connections after 2 heartbeats (30 seconds for old apps)
+		// TODO: Modify after most users are upgraded to >1.0.15
+		const wsReadTimeout = 65 * time.Second
+		_ = ws.SetReadDeadline(time.Now().Add(wsReadTimeout))
+
 		// Get user from context
 		email, err := server.JwtIssuer.GetUserEmail(c)
 		if err != nil {
@@ -107,7 +112,7 @@ func CreateWSHandler(server *common.ServerState) echo.HandlerFunc {
 					if websocket.IsCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure, websocket.CloseNoStatusReceived) {
 						c.Logger().Debug("WebSocket connection closed normally")
 					} else {
-						c.Logger().Error("WebSocket read error: ", err)
+						c.Logger().Errorf("WebSocket read error: %v (user: %s)", err, user.ID)
 					}
 					done <- struct{}{}
 					return
@@ -143,7 +148,9 @@ func CreateWSHandler(server *common.ServerState) echo.HandlerFunc {
 					endCall(c, server, user.ID, *parsedMessage.CallEnd)
 				case parsedMessage.Ping != nil:
 					// Handle ping message
-					c.Logger().Debug("Received ping")
+					// Reset the read deadline
+					_ = ws.SetReadDeadline(time.Now().Add(wsReadTimeout))
+					c.Logger().Debugf("Received ping from user: %s", user.ID)
 					pong := messages.NewPongMessage()
 					pongJSON, err := json.Marshal(pong)
 					if err != nil {
@@ -345,7 +352,7 @@ func initiateCall(ctx echo.Context, s *common.ServerState, ws *websocket.Conn, r
 
 	// User is online ping the callee
 	// Publish a message to the callee channel
-	msg := messages.NewIncomingCallMessage(callerId)
+	msg := messages.NewIncomingCallMessage(callerId, time.Now().Unix())
 	msgJSON, err := json.Marshal(msg)
 	if err != nil {
 		ctx.Logger().Error(err)

backend/internal/messages/messages.go

@@ -81,7 +81,8 @@ type CallEndMessage struct {
 
 // IncomingCallPayload represents the payload for an incoming call by another user
 type IncomingCallPayload struct {
-	CallerID string `json:"caller_id" validate:"required"`
+	CallerID    string `json:"caller_id" validate:"required"`
+	InitiatedAt *int64 `json:"initiated_at,omitempty"`
 }
 
 // IncomingCallMessage is a complete call request message
@@ -297,11 +298,12 @@ func NewCallEndMessage(callID string) CallEndMessage {
 	}
 }
 
-func NewIncomingCallMessage(callerID string) IncomingCallMessage {
+func NewIncomingCallMessage(callerID string, initiatedAt int64) IncomingCallMessage {
 	return IncomingCallMessage{
 		Type: MessageTypeIncomingCall,
 		Payload: IncomingCallPayload{
-			CallerID: callerID,
+			CallerID:    callerID,
+			InitiatedAt: &initiatedAt,
 		},
 	}
 }

tauri/src-tauri/tauri.conf.json

@@ -29,7 +29,7 @@
   },
   "productName": "hopp",
   "mainBinaryName": "hopp",
-  "version": "1.0.14",
+  "version": "1.0.15",
   "identifier": "com.hopp.app",
   "plugins": {
     "updater": {

tauri/src/components/ui/call-banner.tsx

@@ -29,7 +29,7 @@ export const CallBanner = ({ callerId, toastId }: { callerId: string; toastId: s
     };
   }
 
-  const { setCallTokens } = useStore();
+  const { setCallTokens, setIncomingCallCallerId } = useStore();
 
   const handleReject = useMemo(
     () =>
@@ -43,11 +43,12 @@ export const CallBanner = ({ callerId, toastId }: { callerId: string; toastId: s
             },
           });
           toast.dismiss(toastId);
+          setIncomingCallCallerId(null);
         },
         ACTION_THROTTLE_MS,
         { leading: true, trailing: false },
       ),
-    [callerId, toastId],
+    [callerId, toastId, setIncomingCallCallerId],
   );
 
   const handleAnswer = useMemo(
@@ -89,6 +90,7 @@ export const CallBanner = ({ callerId, toastId }: { callerId: string; toastId: s
               }
 
               toast.dismiss(toastId);
+              setIncomingCallCallerId(null);
             }
           };
 
@@ -116,7 +118,7 @@ export const CallBanner = ({ callerId, toastId }: { callerId: string; toastId: s
         ACTION_THROTTLE_MS,
         { leading: true, trailing: false },
       ),
-    [callerId, toastId, setCallTokens, handleReject],
+    [callerId, toastId, setCallTokens, setIncomingCallCallerId, handleReject],
   );
 
   useEffect(() => {

tauri/src/components/ui/participant-row-wo-livekit.tsx

@@ -56,11 +56,14 @@ export const ParticipantRow = (props: { user: components["schemas"]["BaseUser"]
   const isCalling = useStore((state) => state.calling === props.user.id);
   const { setCalling, setCallTokens } = useStore((state) => state);
   const inACall = useStore((state) => state.callTokens !== null);
+  const hasIncomingCall = useStore((state) => state.incomingCallCallerId !== null);
 
   const callbackIdRef = useRef<string>(`call-response-${props.user.id}`);
   const callResolvedRef = useRef(false);
 
   const callUser = useCallback(() => {
+    if (hasIncomingCall) return;
+
     posthog.capture("user_call_request", {
       user_id: props.user.id,
       user_name: props.user.first_name,
@@ -97,7 +100,7 @@ export const ParticipantRow = (props: { user: components["schemas"]["BaseUser"]
         callee_id: props.user.id,
       },
     } as TCallRequestMessage);
-  }, [props.user]);
+  }, [props.user, hasIncomingCall]);
 
   // Add a useEffect to listen for call responses
   // that will unsubscribe from the socket when the component unmounts
@@ -182,6 +185,24 @@ export const ParticipantRow = (props: { user: components["schemas"]["BaseUser"]
       }
     });
 
+    const timeoutId =
+      isCalling ?
+        setTimeout(
+          () => {
+            callResolvedRef.current = true;
+            sounds.ringing.stop();
+            setCalling(null);
+            socketService.send({
+              type: "call_end",
+              payload: { participant_id: props.user.id },
+            });
+            toast.error(`${props.user.first_name} didn't answer`, { duration: 1500 });
+          },
+          // 5 secs more from auto-reject from callee's timeout
+          65_000,
+        )
+      : undefined;
+
     return () => {
       if (!isCalling) return;
       console.debug("Unsubscribing from call response for user:", props.user.id);
@@ -196,6 +217,7 @@ export const ParticipantRow = (props: { user: components["schemas"]["BaseUser"]
           payload: { participant_id: props.user.id },
         });
       }
+      if (timeoutId) clearTimeout(timeoutId);
     };
   }, [isCalling]);
 
@@ -230,7 +252,7 @@ export const ParticipantRow = (props: { user: components["schemas"]["BaseUser"]
               callUser();
             }
           }}
-          disabled={inACall}
+          disabled={inACall || hasIncomingCall}
           className={clsx(
             "px-2 w-auto h-7 flex flex-row items-center gap-1",
             !isCalling && "text-slate-600",

tauri/src/payloads.ts

@@ -213,7 +213,10 @@ export const PCallEndMessage = z.object({
 
 export const PIncomingCallMessage = z.object({
   type: z.literal("incoming_call"),
-  payload: z.object({ caller_id: z.string() }),
+  payload: z.object({
+    caller_id: z.string(),
+    initiated_at: z.number().optional(),
+  }),
 });
 
 export const PAcceptCallMessage = z.object({

tauri/src/services/socket.ts

@@ -105,7 +105,7 @@ class SocketService {
           },
         });
       }
-    }, 30_000); // Send ping every 30 seconds
+    }, 15_000); // Send ping every 15 seconds
   }
 
   private stopHeartbeat() {

tauri/src/store/store.ts

@@ -43,6 +43,8 @@ type State = {
   teammates: components["schemas"]["BaseUser"][] | null;
   // The targeted user id (callee)
   calling: string | null;
+  // Caller id while incoming call banner is visible
+  incomingCallCallerId: string | null;
   // Call tokens for LiveKit
   callTokens: CallState | null;
   customServerUrl: string | null;
@@ -60,6 +62,7 @@ type Actions = {
   getStoredToken: () => Promise<string | null>;
   reset: () => void;
   setCalling: (calling: string | null) => void;
+  setIncomingCallCallerId: (callerId: string | null) => void;
   setCallTokens: (tokens: CallState | null) => void;
   // TODO(@konsalex): Rename `xxCallToken` as its not
   // representative anymore or the actual state it holds.
@@ -77,6 +80,7 @@ const initialState: State = {
   user: null,
   teammates: null,
   calling: null,
+  incomingCallCallerId: null,
   callTokens: null,
   customServerUrl: null,
   livekitUrl: null,
@@ -118,6 +122,10 @@ const useStore = create<State & Actions>()(
       set((state) => {
         state.calling = calling;
       }),
+    setIncomingCallCallerId: (callerId) =>
+      set((state) => {
+        state.incomingCallCallerId = callerId;
+      }),
     setCallTokens: (tokens) =>
       set((state) => {
         state.callTokens = tokens ? { ...tokens, micLevel: tokens.micLevel ?? 0 } : null;

tauri/src/windows/main-window/app.tsx

@@ -1,4 +1,5 @@
-import { useEffect, useRef, useState } from "react";
+import { useEffect, useRef } from "react";
+import { differenceInSeconds, fromUnixTime } from "date-fns";
 import { Button } from "@/components/ui/button";
 import { onOpenUrl } from "@tauri-apps/plugin-deep-link";
 import { checkForUpdates } from "@/update.ts";
@@ -40,14 +41,14 @@ function App() {
     setTeammates,
     setAuthToken,
     setLivekitUrl,
+    setIncomingCallCallerId,
   } = useStore();
 
   const coreProcessCrashedRef = useRef(false);
   useDisableNativeContextMenu();
 
   const { useQuery } = useAPI();
 
-  const [incomingCallerId, setIncomingCallerId] = useState<string | null>(null);
   const sentryMetadataRef = useRef<boolean>(false);
 
   const { error: userError } = useQuery("get", "/api/auth/user", undefined, {
@@ -168,17 +169,19 @@ function App() {
   }, []);
 
   const handleReject = (isInCall?: boolean) => {
-    if (!incomingCallerId && !isInCall) return;
+    const { incomingCallCallerId } = useStore.getState();
+    if (!incomingCallCallerId && !isInCall) return;
     if (!isInCall) {
       sounds.incomingCall.stop();
     }
     socketService.send({
       type: "call_reject",
       payload: {
-        caller_id: incomingCallerId,
+        caller_id: incomingCallCallerId,
         reject_reason: "rejected",
       },
     } as TRejectCallMessage);
+    setIncomingCallCallerId(null);
   };
 
   const handleInCallRejection = (callerID: string) => {
@@ -199,6 +202,18 @@ function App() {
       if (data.type !== "incoming_call") {
         return;
       }
+
+      const MAX_CALL_AGE_S = 60;
+      const incomingMsg = data as TIncomingCallMessage;
+      const initiatedAt = incomingMsg.payload.initiated_at;
+      if (initiatedAt != null) {
+        const ageSeconds = differenceInSeconds(new Date(), fromUnixTime(initiatedAt));
+        if (ageSeconds > MAX_CALL_AGE_S) {
+          console.warn(`Dropping stale incoming call (age: ${ageSeconds}s)`);
+          return;
+        }
+      }
+
       // Check that there is no ongoing call
       // If there is, reject the call
       const { callTokens } = useStore.getState();
@@ -222,7 +237,7 @@ function App() {
       }
 
       if (data.type === "incoming_call") {
-        setIncomingCallerId(data.payload.caller_id);
+        setIncomingCallCallerId(data.payload.caller_id);
 
         /* Reject call if update in progress */
         if (updateInProgress) {
@@ -250,6 +265,7 @@ function App() {
       if (data.type === "call_end") {
         // Get call info before clearing tokens
         toast.dismiss("call-banner");
+        setIncomingCallCallerId(null);
 
         const { callTokens: currentCallTokens, user } = useStore.getState();
         const participantId = currentCallTokens?.participant || "";