display event stream history in user menu

OliverDudgeon · OliverDudgeon · commit fc99a1cb745b · 2025-07-22T19:53:15.000+01:00
diff --git a/src/components/eventStream/EventList.tsx b/src/components/eventStream/EventList.tsx
@@ -0,0 +1,55 @@
+import { Box, List, ListItem, Typography } from "@mui/material";
+import dayjs from "dayjs";
+import utc from "dayjs/plugin/utc";
+
+import { DATE_FORMAT, TIME_FORMAT } from "../../constants/datetimes";
+import { useEventStream } from "../../state/eventStream";
+import { EventMessage } from "../eventMessages/EventMessage";
+
+dayjs.extend(utc);
+
+/**
+ * Displays a list of events sorted by timestamp (newest first)
+ */
+export const EventList = () => {
+  const { events } = useEventStream();
+
+  if (events.length === 0) {
+    return (
+      <Box sx={{ p: 2, textAlign: "center" }}>
+        <Typography color="text.secondary" variant="body2">
+          No events to display
+        </Typography>
+      </Box>
+    );
+  }
+
+  return (
+    <Box sx={{ maxHeight: 400, overflow: "auto" }}>
+      <Typography sx={{ p: 1, fontWeight: "bold" }} variant="subtitle2">
+        Events ({events.length})
+      </Typography>
+
+      <List dense>
+        {events.map((event) => {
+          const eventTime = dayjs.utc(event.timestamp);
+
+          return (
+            <ListItem key={`${event.timestamp}-${event.ordinal}`} sx={{ mb: 1, borderRadius: 1 }}>
+              <Box sx={{ width: "100%" }}>
+                <Typography
+                  color="text.secondary"
+                  sx={{ display: "block", mb: 0.5 }}
+                  variant="caption"
+                >
+                  {eventTime.local().format(TIME_FORMAT)} • {eventTime.local().format(DATE_FORMAT)}
+                </Typography>
+                <EventMessage message={event} />
+              </Box>
+            </ListItem>
+          );
+        })}
+      </List>
+    </Box>
+  );
+};
diff --git a/src/components/eventStream/EventStream.tsx b/src/components/eventStream/EventStream.tsx
@@ -6,36 +6,70 @@ import {
   useGetEventStream,
 } from "@squonk/account-server-client/event-stream";
 
+import dayjs from "dayjs";
+import utc from "dayjs/plugin/utc";
 import { useAtom } from "jotai";
 import { useSnackbar } from "notistack";
 
 import { useASAuthorizationStatus } from "../../hooks/useIsAuthorized";
-import { getMessageFromEvent, protoBlobToText } from "../../protobuf/protobuf";
-import { eventStreamEnabledAtom } from "../../state/eventStream";
+import { getMessageFromEvent } from "../../protobuf/protobuf";
+import { eventStreamEnabledAtom, useEventStream } from "../../state/eventStream";
 import { useUnreadEventCount } from "../../state/notifications";
 import { EventMessage } from "../eventMessages/EventMessage";
 import { useIsEventStreamInstalled } from "./useIsEventStreamInstalled";
 
+dayjs.extend(utc);
+
+/**
+ * Builds WebSocket URL
+ */
+const buildWebSocketUrl = (location: string): string => {
+  const url = new URL(location);
+  url.protocol = "wss:";
+  return url.toString();
+};
+
+/**
+ * Manages WebSocket connection for event stream and displays toast notifications
+ */
 export const EventStream = () => {
   const isEventStreamInstalled = useIsEventStreamInstalled();
   const [location, setLocation] = useState<string | null>(null);
   const { enqueueSnackbar } = useSnackbar();
   const { incrementCount } = useUnreadEventCount();
   const asRole = useASAuthorizationStatus();
+  const { addEvent, isEventNewerThanSession, initializeSession } = useEventStream();
 
   const { data, error: streamError } = useGetEventStream({
     query: { select: (data) => data.location, enabled: !!asRole && isEventStreamInstalled },
   });
+
   const { mutate: createEventStream } = useCreateEventStream({
-    mutation: {
-      onSuccess: (eventStreamResponse) => {
-        setLocation(eventStreamResponse.location);
-      },
-    },
+    mutation: { onSuccess: (eventStreamResponse) => setLocation(eventStreamResponse.location) },
   });
+
   const [eventStreamEnabled] = useAtom(eventStreamEnabledAtom);
 
-  // Define callbacks *before* useWebSocket hook
+  const handleWebSocketMessage = useCallback(
+    (event: MessageEvent) => {
+      const message = getMessageFromEvent(JSON.parse(event.data));
+
+      if (
+        message &&
+        addEvent(message) && // Only show toast for events newer than session start
+        isEventNewerThanSession(message)
+      ) {
+        enqueueSnackbar(<EventMessage message={message} />, {
+          variant: "default",
+          anchorOrigin: { horizontal: "right", vertical: "bottom" },
+          autoHideDuration: 10_000,
+        });
+        incrementCount();
+      }
+    },
+    [enqueueSnackbar, incrementCount, addEvent, isEventNewerThanSession],
+  );
+
   const handleWebSocketOpen = useCallback(() => {
     enqueueSnackbar("Connected to event stream", {
       variant: "success",
@@ -45,19 +79,14 @@ export const EventStream = () => {
 
   const handleWebSocketClose = useCallback(
     (event: CloseEvent) => {
-      console.log(event);
-      if (event.wasClean) {
-        enqueueSnackbar("Disconnected from event stream", {
-          variant: "info",
-          anchorOrigin: { horizontal: "right", vertical: "bottom" },
-        });
-      } else {
-        console.warn("EventStream: WebSocket closed unexpectedly.");
-        enqueueSnackbar("Event stream disconnected unexpectedly. Attempting to reconnect...", {
-          variant: "warning",
-          anchorOrigin: { horizontal: "right", vertical: "bottom" },
-        });
-      }
+      const message = event.wasClean
+        ? "Disconnected from event stream"
+        : "Event stream disconnected unexpectedly. Attempting to reconnect...";
+
+      enqueueSnackbar(message, {
+        variant: event.wasClean ? "info" : "warning",
+        anchorOrigin: { horizontal: "right", vertical: "bottom" },
+      });
     },
     [enqueueSnackbar],
   );
@@ -69,51 +98,8 @@ export const EventStream = () => {
     });
   }, [enqueueSnackbar]);
 
-  const handleWebSocketMessage = useCallback(
-    (event: MessageEvent) => {
-      console.log("message");
-      if (event.data instanceof Blob) {
-        protoBlobToText(event.data)
-          .then((textData) => {
-            const message = getMessageFromEvent(textData);
-            if (message) {
-              incrementCount();
-              enqueueSnackbar(<EventMessage message={message} />, {
-                variant: "default",
-                anchorOrigin: { horizontal: "right", vertical: "bottom" },
-                autoHideDuration: 10_000,
-              });
-            } else {
-              console.warn(
-                "Received event data could not be parsed into a known message type:",
-                textData,
-              );
-            }
-          })
-          .catch((error) => {
-            console.error("Error processing protobuf message:", error);
-            enqueueSnackbar("Error processing incoming event", {
-              variant: "error",
-              anchorOrigin: { horizontal: "right", vertical: "bottom" },
-            });
-          });
-      } else {
-        console.warn("Received non-Blob WebSocket message:", event.data);
-      }
-    },
-    [enqueueSnackbar, incrementCount],
-  );
-
-  let wsUrl = null;
-  if (eventStreamEnabled && asRole && location) {
-    const url = new URL(location);
-    url.protocol = "wss:";
-    url.search = new URLSearchParams({
-      // stream_from_timestamp: encodeURIComponent("2025-07-1T12:00:00Z"),
-      stream_from_ordinal: "1",
-    }).toString();
-    wsUrl = url.toString();
-  }
+  // Build WebSocket URL
+  const wsUrl = eventStreamEnabled && asRole && location ? buildWebSocketUrl(location) : null;
 
   useWebSocket(wsUrl, {
     onOpen: handleWebSocketOpen,
@@ -126,7 +112,6 @@ export const EventStream = () => {
     reconnectInterval: 3000,
   });
 
-  // Effects can now safely use the hook results or return early based on auth
   useEffect(() => {
     if (asRole && data) {
       setLocation(data);
@@ -139,5 +124,10 @@ export const EventStream = () => {
     }
   }, [asRole, streamError, createEventStream]);
 
+  // Initialize session on client side only
+  useEffect(() => {
+    initializeSession();
+  }, [initializeSession]);
+
   return null;
 };
diff --git a/src/components/eventStream/EventStreamMessages.tsx b/src/components/eventStream/EventStreamMessages.tsx
@@ -0,0 +1,21 @@
+import { Box, Divider, Typography } from "@mui/material";
+
+import { EventList } from "./EventList";
+import { EventStreamToggle } from "./EventStreamToggle";
+
+/**
+ * Main event stream interface in the user menu popover
+ */
+export const EventStreamMessages = () => (
+  <Box sx={{ minWidth: 300 }}>
+    <Typography sx={{ mb: 1 }} variant="h6">
+      Event Stream
+    </Typography>
+
+    <EventStreamToggle />
+
+    <Divider sx={{ my: 2 }} />
+
+    <EventList />
+  </Box>
+);
diff --git a/src/components/instances/JobDetails/JobInputSection/useGetJobInputs.ts b/src/components/instances/JobDetails/JobInputSection/useGetJobInputs.ts
@@ -27,8 +27,6 @@ export const useGetJobInputs = (instance: InstanceGetResponse | InstanceSummary)
     { query: { enabled: inputsEnabled, retry: instance.job_id === TEST_JOB_ID ? 1 : 3 } },
   );
 
-  console.log(instance);
-
   // Parse application specification
   const applicationSpecification: ApplicationSpecification = instance.application_specification
     ? JSON.parse(instance.application_specification)
diff --git a/src/layouts/navigation/UserMenuContent.tsx b/src/layouts/navigation/UserMenuContent.tsx
@@ -5,30 +5,10 @@ import { AuthButton } from "../../components/auth/AuthButton";
 import { CenterLoader } from "../../components/CenterLoader";
 import { Chips } from "../../components/Chips";
 import { ColourSchemeSelection } from "../../components/ColourSchemeSelection";
-import { EventStreamToggle } from "../../components/eventStream/EventStreamToggle";
+import { EventStreamMessages } from "../../components/eventStream/EventStreamMessages";
 import { useASAuthorizationStatus, useDMAuthorizationStatus } from "../../hooks/useIsAuthorized";
 import { useKeycloakUser } from "../../hooks/useKeycloakUser";
 
-/**
- * Content of the user menu
- */
-export const UserMenuContent = () => {
-  const theme = useTheme();
-  const biggerThanMd = useMediaQuery(theme.breakpoints.up("md"));
-  // Removed eventStreamEnabledAtom usage, now handled in EventStreamToggle
-
-  return (
-    <Box sx={{ textAlign: biggerThanMd ? "center" : undefined }}>
-      <Typography gutterBottom variant="h3">
-        Account
-      </Typography>
-      <UserMenuContentInner />
-      <ColourSchemeSelection />
-      <EventStreamToggle />
-    </Box>
-  );
-};
-
 const UserMenuContentInner = () => {
   const asRole = useASAuthorizationStatus();
   const dmRole = useDMAuthorizationStatus();
@@ -74,3 +54,23 @@ const UserMenuContentInner = () => {
 
   return <AuthButton mode="login" />;
 };
+
+/**
+ * Content of the user menu
+ */
+export const UserMenuContent = () => {
+  const theme = useTheme();
+  const biggerThanMd = useMediaQuery(theme.breakpoints.up("md"));
+  // Removed eventStreamEnabledAtom usage, now handled in EventStreamToggle
+
+  return (
+    <Box sx={{ textAlign: biggerThanMd ? "center" : undefined }}>
+      <Typography gutterBottom variant="h3">
+        Account
+      </Typography>
+      <UserMenuContentInner />
+      <ColourSchemeSelection />
+      <EventStreamMessages />
+    </Box>
+  );
+};
diff --git a/src/protobuf/protobuf.ts b/src/protobuf/protobuf.ts
@@ -82,19 +82,24 @@ export const storageType = getPrefixedMessageNameFromSchema(
 ) as StorageTypeName;
 // --- End Runtime Constants ---
 
-type ProcessingMessagePayload = {
+interface MessageBase {
+  timestamp: string;
+  ordinal: number;
+}
+
+interface ProcessingMessagePayload extends MessageBase {
   type: ProcessingTypeName;
   name: string;
   coins: string;
   product: string;
-};
+}
 
-type StorageMessagePayload = {
+interface StorageMessagePayload extends MessageBase {
   type: StorageTypeName;
   name: string;
   bytes: string;
   reason: StorageReasonEnum;
-};
+}
 
 // Discriminated union type representing the possible charge message payloads
 // derived from Protobuf messages (Processing or Storage).
@@ -103,6 +108,8 @@ export type ChargeMessage = ProcessingMessagePayload | StorageMessagePayload;
 interface EventStreamMessage {
   message_type: string;
   message_body: any;
+  ess_timestamp: string;
+  ess_ordinal: number;
 }
 
 /**
@@ -133,6 +140,8 @@ export const getMessageFromEvent = (event: EventStreamMessage): ChargeMessage |
     case processingType: {
       const parsed = fromJson(MerchantProcessingChargeMessageSchema, event.message_body);
       return {
+        timestamp: parsed.timestamp,
+        ordinal: event.ess_ordinal,
         type: processingType,
         name: parsed.name,
         coins: parsed.coins,
@@ -141,7 +150,14 @@ export const getMessageFromEvent = (event: EventStreamMessage): ChargeMessage |
     }
     case storageType: {
       const parsed = fromJson(MerchantStorageChargeMessageSchema, event.message_body);
-      return { type: storageType, name: parsed.name, bytes: parsed.bytes, reason: parsed.reason };
+      return {
+        timestamp: parsed.timestamp,
+        ordinal: event.ess_ordinal,
+        type: storageType,
+        name: parsed.name,
+        bytes: parsed.bytes,
+        reason: parsed.reason,
+      };
     }
     default:
       return null;
diff --git a/src/state/eventStream.ts b/src/state/eventStream.ts
