Add Pub/Sub publisher for event notifications

k8s/base/configmap.yaml

@@ -5,5 +5,6 @@ metadata:
 data:
   K8S: "1"
   NODE_ENV: "production"
+  GCP_PROJECT_ID: "ethans-services"
   SENTRY_ORG: "forecasting"
   SENTRY_PROJECT: "forecasting-app"

k8s/prod/configmap-env.yaml

@@ -7,4 +7,5 @@ data:
   APP_BASE_URL: "https://forecasting.ethanswan.com"
   IDP_BASE_URL: "http://identity.identity-prod.svc.cluster.local"
   NEXT_PUBLIC_IDP_BASE_URL: "https://identity.ethanswan.com"
+  PUBSUB_TOPIC: "forecasting-events-prod"
   SENTRY_DSN: "https://42fb7fde7d5842831f2324ed33c7f50f@o4509062063587328.ingest.us.sentry.io/4509062066012160"

k8s/staging/configmap-env.yaml

@@ -7,4 +7,5 @@ data:
   APP_BASE_URL: "https://forecasting-staging.tailc06f30.ts.net"
   IDP_BASE_URL: "http://identity.identity-staging.svc.cluster.local"
   NEXT_PUBLIC_IDP_BASE_URL: "https://identity-staging.tailc06f30.ts.net"
+  PUBSUB_TOPIC: "forecasting-events-staging"
   SENTRY_DSN: "https://42fb7fde7d5842831f2324ed33c7f50f@o4509062063587328.ingest.us.sentry.io/4509062066012160"

lib/db_actions/competition-members.ts

@@ -8,6 +8,7 @@ import {
 import { getUserFromCookies } from "../get-user";
 import { revalidatePath } from "next/cache";
 import { logger } from "@/lib/logger";
+import { publishEvent } from "@/lib/pubsub/client";
 import {
   ServerActionResult,
   success,
@@ -514,7 +515,7 @@ export async function addCompetitionMemberById({
       // Verify the target user exists and is active
       const userToAdd = await trx
         .selectFrom("users")
-        .select("id")
+        .select(["id", "name", "email"])
         .where("id", "=", userId)
         .where("deactivated_at", "is", null)
         .executeTakeFirst();
@@ -551,20 +552,50 @@ export async function addCompetitionMemberById({
         .returningAll()
         .executeTakeFirstOrThrow();
 
-      return success(inserted);
+      // Look up competition for notification
+      const competition = await trx
+        .selectFrom("competitions")
+        .select(["name", "is_private"])
+        .where("id", "=", competitionId)
+        .executeTakeFirst();
+
+      return success({ inserted, userToAdd: userToAdd!, competition });
     });
 
     if (result.success) {
+      const { inserted, userToAdd, competition } = result.data;
       const duration = Date.now() - startTime;
       logger.info("Competition member added successfully", {
         operation: "addCompetitionMemberById",
         table: "competition_members",
         competitionId,
-        addedUserId: result.data.user_id,
+        addedUserId: inserted.user_id,
         role,
         duration,
       });
       revalidatePath(`/competitions/${competitionId}`);
+
+      // Notify the added user (fire-and-forget, only for private competitions)
+      if (competition?.is_private) {
+        publishEvent({
+          event_type: "competition.member_added",
+          source: "forecasting",
+          timestamp: new Date().toISOString(),
+          notify: [{ email: userToAdd.email, name: userToAdd.name }],
+          notify_link: `${process.env.APP_BASE_URL}/competitions/${competitionId}`,
+          data: {
+            competition_name: competition.name,
+            competition_id: competitionId,
+          },
+        }).catch((err) => {
+          logger.error("Failed to publish competition.member_added event", err as Error, {
+            competitionId,
+            userId,
+          });
+        });
+      }
+
+      return success(inserted);
     }
 
     return result;

lib/pubsub/client.test.ts

@@ -0,0 +1,87 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+
+vi.mock("server-only", () => ({}));
+
+const mockPublishMessage = vi.fn().mockResolvedValue("msg-123");
+const mockTopic = vi.fn().mockReturnValue({ publishMessage: mockPublishMessage });
+
+vi.mock("@google-cloud/pubsub", () => ({
+  PubSub: class {
+    topic = mockTopic;
+  },
+}));
+
+describe("publishEvent", () => {
+  beforeEach(() => {
+    vi.resetModules();
+    vi.clearAllMocks();
+    mockTopic.mockReturnValue({ publishMessage: mockPublishMessage });
+    mockPublishMessage.mockResolvedValue("msg-123");
+    vi.stubEnv("GCP_PROJECT_ID", "test-project");
+    vi.stubEnv("PUBSUB_TOPIC", "test-topic");
+  });
+
+  it("publishes a BaseEvent to the configured topic", async () => {
+    const { publishEvent } = await import("./client");
+
+    const event = {
+      event_type: "test.notification",
+      source: "forecasting",
+      timestamp: "2026-04-07T00:00:00Z",
+      notify: [{ email: "test@example.com", name: "Test" }],
+      data: { message: "hello" },
+    };
+
+    const messageId = await publishEvent(event);
+
+    expect(messageId).toBe("msg-123");
+    expect(mockTopic).toHaveBeenCalledWith("test-topic");
+    expect(mockPublishMessage).toHaveBeenCalledWith({ json: event });
+  });
+
+  it("returns the message ID from Pub/Sub", async () => {
+    mockPublishMessage.mockResolvedValue("msg-456");
+    const { publishEvent } = await import("./client");
+
+    const event = {
+      event_type: "some.event",
+      source: "forecasting",
+      timestamp: "2026-04-07T00:00:00Z",
+
+      data: {},
+    };
+
+    const messageId = await publishEvent(event);
+    expect(messageId).toBe("msg-456");
+  });
+
+  it("throws when GCP_PROJECT_ID is missing", async () => {
+    vi.stubEnv("GCP_PROJECT_ID", "");
+    const { publishEvent } = await import("./client");
+
+    const event = {
+      event_type: "test.notification",
+      source: "forecasting",
+      timestamp: "2026-04-07T00:00:00Z",
+
+      data: {},
+    };
+
+    await expect(publishEvent(event)).rejects.toThrow("GCP_PROJECT_ID");
+  });
+
+  it("throws when PUBSUB_TOPIC is missing", async () => {
+    vi.stubEnv("PUBSUB_TOPIC", "");
+    const { publishEvent } = await import("./client");
+
+    const event = {
+      event_type: "test.notification",
+      source: "forecasting",
+      timestamp: "2026-04-07T00:00:00Z",
+
+      data: {},
+    };
+
+    await expect(publishEvent(event)).rejects.toThrow("PUBSUB_TOPIC");
+  });
+});

lib/pubsub/client.ts

@@ -0,0 +1,40 @@
+import "server-only";
+import { PubSub } from "@google-cloud/pubsub";
+
+function requiredEnv(name: string): string {
+  const value = process.env[name];
+  if (!value) throw new Error(`Missing required environment variable: ${name}`);
+  return value;
+}
+
+export interface NotifyTarget {
+  email: string;
+  name: string;
+}
+
+export interface BaseEvent {
+  event_type: string;
+  source: string;
+  timestamp: string;
+  notify?: NotifyTarget[];
+  notify_link?: string;
+  data: Record<string, unknown>;
+}
+
+let pubsubClient: PubSub | null = null;
+
+function getClient(): PubSub {
+  if (!pubsubClient) {
+    pubsubClient = new PubSub({ projectId: requiredEnv("GCP_PROJECT_ID") });
+  }
+  return pubsubClient;
+}
+
+export async function publishEvent(event: BaseEvent): Promise<string> {
+  const topic = getClient().topic(requiredEnv("PUBSUB_TOPIC"));
+  const messageId = await topic.publishMessage({
+    json: event,
+  });
+  console.log(`Published event ${event.event_type} (message ${messageId})`);
+  return messageId;
+}