Auto-reload the queue page every 10 seconds

Author: matt-aitken

apps/webapp/app/env.server.ts

@@ -568,7 +568,6 @@ const EnvironmentSchema = z.object({
   /** How long should the presence ttl last */
   DEV_PRESENCE_TTL_MS: z.coerce.number().int().default(30_000),
   DEV_PRESENCE_POLL_INTERVAL_MS: z.coerce.number().int().default(5_000),
-  DEV_PRESENCE_RECONNECT_THRESHOLD_MS: z.coerce.number().int().default(2_000),
   /** How many ms to wait until dequeuing again, if there was a run last time */
   DEV_DEQUEUE_INTERVAL_WITH_RUN: z.coerce.number().int().default(250),
   /** How many ms to wait until dequeuing again, if there was no run last time */
@@ -660,6 +659,9 @@ const EnvironmentSchema = z.object({
 
   TASK_EVENT_PARTITIONING_ENABLED: z.string().default("0"),
   TASK_EVENT_PARTITIONED_WINDOW_IN_SECONDS: z.coerce.number().int().default(60), // 1 minute
+
+  QUEUE_SSE_AUTORELOAD_INTERVAL_MS: z.coerce.number().int().default(10_000),
+  QUEUE_SSE_AUTORELOAD_TIMEOUT_MS: z.coerce.number().int().default(60_000),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx

@@ -6,7 +6,7 @@ import {
   PlayIcon,
   RectangleStackIcon,
 } from "@heroicons/react/20/solid";
-import { Await, Form, useNavigation, type MetaFunction } from "@remix-run/react";
+import { Await, Form, useNavigation, useRevalidator, type MetaFunction } from "@remix-run/react";
 import { type ActionFunctionArgs, type LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { Suspense, useEffect, useState } from "react";
 import { TypedAwait, typeddefer, useTypedLoaderData } from "remix-typedjson";
@@ -58,6 +58,8 @@ import { Callout } from "~/components/primitives/Callout";
 import upgradeForQueuesPath from "~/assets/images/queues-dashboard.png";
 import { PauseQueueService } from "~/v3/services/pauseQueue.server";
 import { Badge } from "~/components/primitives/Badge";
+import { useEventSource } from "~/hooks/useEventSource";
+import { useProject } from "~/hooks/useProject";
 
 const SearchParamsSchema = z.object({
   page: z.coerce.number().min(1).default(1),
@@ -198,9 +200,26 @@ export default function Page() {
   const { environment, queues, success, pagination, code } = useTypedLoaderData<typeof loader>();
 
   const organization = useOrganization();
+  const project = useProject();
   const env = useEnvironment();
   const plan = useCurrentPlan();
 
+  const revalidation = useRevalidator();
+
+  const streamedEvents = useEventSource(
+    `/resources/orgs/${organization.slug}/projects/${project.slug}/env/${env.slug}/queues/stream`,
+    {
+      event: "update",
+    }
+  );
+
+  useEffect(() => {
+    if (streamedEvents) {
+      console.log("streamedEvents", streamedEvents);
+      revalidation.revalidate();
+    }
+  }, [streamedEvents]);
+
   return (
     <PageContainer>
       <NavBar>

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues.stream.tsx

@@ -0,0 +1,55 @@
+import { $replica } from "~/db.server";
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { requireUserId } from "~/services/session.server";
+import { EnvironmentParamSchema } from "~/utils/pathBuilder";
+import { createSSELoader } from "~/utils/sse";
+
+export const loader = createSSELoader({
+  timeout: env.QUEUE_SSE_AUTORELOAD_TIMEOUT_MS,
+  interval: env.QUEUE_SSE_AUTORELOAD_INTERVAL_MS,
+  debug: true,
+  handler: async ({ request, params }) => {
+    const userId = await requireUserId(request);
+    const { projectParam, envParam } = EnvironmentParamSchema.parse(params);
+
+    const environment = await $replica.runtimeEnvironment.findFirst({
+      where: {
+        slug: envParam,
+        type: "DEVELOPMENT",
+        orgMember: {
+          userId,
+        },
+        project: {
+          slug: projectParam,
+        },
+      },
+    });
+
+    if (!environment) {
+      throw new Response("Not Found", { status: 404 });
+    }
+
+    return {
+      beforeStream: async () => {
+        logger.debug("Start queue page SSE session", {
+          environmentId: environment.id,
+        });
+      },
+      initStream: async ({ send }) => {
+        send({ event: "time", data: new Date().toISOString() });
+      },
+      iterator: async ({ send }) => {
+        send({
+          event: "update",
+          data: new Date().toISOString(),
+        });
+      },
+      cleanup: async () => {
+        logger.debug("End queue page SSE session", {
+          environmentId: environment.id,
+        });
+      },
+    };
+  },
+});

references/hello-world/src/trigger/queues.ts

@@ -39,3 +39,14 @@ export const queuesTester = task({
     logger.log("Resumed queue", { resumedQueue });
   },
 });
+
+export const otherQueueTask = task({
+  id: "other-queue-task",
+  queue: {
+    name: "my-custom-queue",
+    concurrencyLimit: 1,
+  },
+  run: async (payload: any, { ctx }) => {
+    logger.log("Other queue task", { payload });
+  },
+});