fix(realtime): prevent silent Postgres connection drop on notification listener (fixes #2619)

services/realtime-proxy/src/listener.rs

@@ -20,7 +20,7 @@ pub async fn run_postgres_listener(config: Config, broadcaster: Arc<Broadcaster>
 
                 // Spawn connection handler task
                 let stream_task = tokio::spawn(async move {
-                    if let Err(e) = futures_util::future::poll_fn(|cx| connection.poll_message(cx)).await {
+                    if let Err(e) = connection.await {
                         error!("Postgres connection error: {}", e);
                     }
                 });

src/hooks/useCursorEventsQuery.test.ts

@@ -1,11 +1,30 @@
-import { describe, it, expect, vi, beforeEach } from "vitest";
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
 import { fetchGraphQL, EVENTS_CONNECTION_QUERY } from "./useCursorEventsQuery";
 
-global.fetch = vi.fn();
+const originalFetch = globalThis.fetch;
+
+// ── Mock OpenTelemetry so tests don't need a real tracer ────────────
+vi.mock("@opentelemetry/api", () => {
+  const mockSpan = {
+    setStatus: vi.fn(),
+    recordException: vi.fn(),
+    end: vi.fn(),
+  };
+  return {
+    trace: {
+      getTracer: () => ({ startSpan: () => mockSpan }),
+    },
+    SpanStatusCode: { ERROR: 2 },
+  };
+});
 
 describe("useCursorEventsQuery", () => {
   beforeEach(() => {
-    vi.clearAllMocks();
+    globalThis.fetch = vi.fn();
+  });
+
+  afterEach(() => {
+    globalThis.fetch = originalFetch;
   });
 
   it("fetchGraphQL posts query to /api/graphql and returns data", async () => {
@@ -28,13 +47,14 @@ describe("useCursorEventsQuery", () => {
       },
     };
 
-    (global.fetch as unknown as ReturnType<typeof vi.fn>).mockResolvedValue({
+    (globalThis.fetch as ReturnType<typeof vi.fn>).mockResolvedValue({
+      ok: true,
       json: vi.fn().mockResolvedValue({ data: mockData }),
     });
 
     const result = await fetchGraphQL(EVENTS_CONNECTION_QUERY, { first: 1, after: undefined });
 
-    expect(global.fetch).toHaveBeenCalledWith(
+    expect(globalThis.fetch).toHaveBeenCalledWith(
       "/api/graphql",
       expect.objectContaining({
         method: "POST",
@@ -47,13 +67,46 @@ describe("useCursorEventsQuery", () => {
     expect(result).toEqual(mockData);
   });
 
-  it("fetchGraphQL throws error when graphql endpoint returns errors", async () => {
-    (global.fetch as unknown as ReturnType<typeof vi.fn>).mockResolvedValue({
+  it("fetchGraphQL throws error when graphql endpoint returns errors with no data", async () => {
+    (globalThis.fetch as ReturnType<typeof vi.fn>).mockResolvedValue({
+      ok: true,
       json: vi.fn().mockResolvedValue({
         errors: [{ message: "GraphQL syntax error" }],
       }),
     });
 
     await expect(fetchGraphQL(EVENTS_CONNECTION_QUERY)).rejects.toThrow("GraphQL syntax error");
   });
+
+  it("fetchGraphQL returns partial data when both data and errors are present", async () => {
+    const partialData = {
+      events: {
+        edges: [],
+        nodes: [],
+        pageInfo: {
+          hasNextPage: false,
+          hasPreviousPage: false,
+          startCursor: null,
+          endCursor: null,
+        },
+        totalCount: 0,
+      },
+    };
+
+    (globalThis.fetch as ReturnType<typeof vi.fn>).mockResolvedValue({
+      ok: true,
+      json: vi.fn().mockResolvedValue({
+        data: partialData,
+        errors: [
+          {
+            message: "Organizer resolver timeout",
+            path: ["events", "edges", 0, "node", "organizer"],
+          },
+        ],
+      }),
+    });
+
+    const result = await fetchGraphQL(EVENTS_CONNECTION_QUERY, { first: 10 });
+    expect(result).toEqual(partialData);
+  });
 });

src/hooks/useCursorEventsQuery.ts

@@ -93,24 +93,7 @@ export const EVENTS_CONNECTION_QUERY = /* GraphQL */ `
   }
 `;
 
-export async function fetchGraphQL<TData, TVariables>(
-  query: string,
-  variables?: TVariables,
-): Promise<TData> {
-  const res = await fetch("/api/graphql", {
-    method: "POST",
-    headers: {
-      "Content-Type": "application/json",
-    },
-    body: JSON.stringify({ query, variables }),
-  });
-
-  const json = await res.json();
-  if (json.errors && json.errors.length > 0) {
-    throw new Error(json.errors[0].message);
-  }
-  return json.data;
-}
+export { fetchGraphQL } from "@/lib/graphql-client";
 
 /**
  * Hook to consume the GraphQL Relay-style cursor-paginated events connection API (`events(first: $first, after: $after)`).

src/lib/graphql-client.test.ts

@@ -0,0 +1,149 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { fetchGraphQL, GraphQLPartialError, isPartialNull } from "./graphql-client";
+
+// ── Mock OpenTelemetry so tests don't need a real tracer ────────────
+vi.mock("@opentelemetry/api", () => {
+  const mockSpan = {
+    setStatus: vi.fn(),
+    recordException: vi.fn(),
+    end: vi.fn(),
+  };
+  return {
+    trace: {
+      getTracer: () => ({ startSpan: () => mockSpan }),
+    },
+    SpanStatusCode: { ERROR: 2 },
+  };
+});
+
+// ── Test suite ──────────────────────────────────────────────────────
+describe("fetchGraphQL", () => {
+  const originalFetch = globalThis.fetch;
+
+  beforeEach(() => {
+    // Reset fetch mock before each test
+    globalThis.fetch = vi.fn();
+  });
+
+  afterEach(() => {
+    globalThis.fetch = originalFetch;
+  });
+
+  it("returns data on a clean 200 response with no errors", async () => {
+    const mockData = { user: { id: "1", name: "Alice" } };
+
+    (globalThis.fetch as ReturnType<typeof vi.fn>).mockResolvedValue({
+      ok: true,
+      json: () => Promise.resolve({ data: mockData }),
+    });
+
+    const result = await fetchGraphQL("query { user { id name } }");
+    expect(result).toEqual(mockData);
+  });
+
+  it("throws Error when response has errors but no data (complete failure)", async () => {
+    (globalThis.fetch as ReturnType<typeof vi.fn>).mockResolvedValue({
+      ok: true,
+      json: () =>
+        Promise.resolve({
+          errors: [{ message: "Database unavailable" }],
+        }),
+    });
+
+    await expect(fetchGraphQL("query { user { id } }")).rejects.toThrow("Database unavailable");
+  });
+
+  it("returns partial data when both data and errors are present", async () => {
+    const partialData = { user: { id: "1", name: "Alice", recentPosts: null } };
+
+    (globalThis.fetch as ReturnType<typeof vi.fn>).mockResolvedValue({
+      ok: true,
+      json: () =>
+        Promise.resolve({
+          data: partialData,
+          errors: [{ message: "Posts resolver timeout", path: ["user", "recentPosts"] }],
+        }),
+    });
+
+    const result = await fetchGraphQL("query { user { id name recentPosts { title } } }");
+    expect(result).toEqual(partialData);
+    expect((result as typeof partialData).user.recentPosts).toBeNull();
+  });
+
+  it("throws on non-OK HTTP status", async () => {
+    (globalThis.fetch as ReturnType<typeof vi.fn>).mockResolvedValue({
+      ok: false,
+      status: 500,
+      statusText: "Internal Server Error",
+    });
+
+    await expect(fetchGraphQL("query { user { id } }")).rejects.toThrow(
+      "GraphQL request failed: 500 Internal Server Error",
+    );
+  });
+
+  it("throws when response has neither data nor errors", async () => {
+    (globalThis.fetch as ReturnType<typeof vi.fn>).mockResolvedValue({
+      ok: true,
+      json: () => Promise.resolve({}),
+    });
+
+    await expect(fetchGraphQL("query { user { id } }")).rejects.toThrow(
+      "GraphQL response contained neither data nor errors",
+    );
+  });
+
+  it("sends request to custom endpoint when provided", async () => {
+    const mockData = { ok: true };
+
+    (globalThis.fetch as ReturnType<typeof vi.fn>).mockResolvedValue({
+      ok: true,
+      json: () => Promise.resolve({ data: mockData }),
+    });
+
+    await fetchGraphQL("query { ok }", undefined, { endpoint: "/custom/graphql" });
+
+    expect(globalThis.fetch).toHaveBeenCalledWith(
+      "/custom/graphql",
+      expect.objectContaining({
+        method: "POST",
+      }),
+    );
+  });
+});
+
+describe("GraphQLPartialError", () => {
+  it("carries data and errors", () => {
+    const data = { user: { id: "1" } };
+    const errors = [{ message: "Timeout on posts" }];
+    const err = new GraphQLPartialError(errors, data);
+
+    expect(err).toBeInstanceOf(Error);
+    expect(err.name).toBe("GraphQLPartialError");
+    expect(err.message).toBe("Timeout on posts");
+    expect(err.data).toEqual(data);
+    expect(err.graphQLErrors).toEqual(errors);
+  });
+});
+
+describe("isPartialNull", () => {
+  it("returns true for null", () => {
+    expect(isPartialNull(null)).toBe(true);
+  });
+
+  it("returns true for undefined", () => {
+    expect(isPartialNull(undefined)).toBe(true);
+  });
+
+  it("returns false for empty array", () => {
+    expect(isPartialNull([])).toBe(false);
+  });
+
+  it("returns false for 0", () => {
+    expect(isPartialNull(0)).toBe(false);
+  });
+
+  it("returns false for empty string", () => {
+    expect(isPartialNull("")).toBe(false);
+  });
+});