Submatter/apps/web/app/api/fleet/events/route.ts

import { desc, eq, inArray } from "drizzle-orm";

import { db } from "@tasks/database";
import { convoyEvents, workspaceMembers, workspaces } from "@tasks/database/schema";
import { auth } from "@/lib/auth";

export const runtime = "nodejs";
export const dynamic = "force-dynamic";

async function getUserWorkspaceIds(userId: string): Promise<string[]> {
  const memberRows = await db
    .select({ workspaceId: workspaceMembers.workspaceId })
    .from(workspaceMembers)
    .where(eq(workspaceMembers.userId, userId));
  const ownedRows = await db
    .select({ id: workspaces.id })
    .from(workspaces)
    .where(eq(workspaces.ownerUserId, userId));
  const ids = new Set<string>([...memberRows.map((r) => r.workspaceId), ...ownedRows.map((r) => r.id)]);
  return [...ids];
}

export async function GET() {
  const session = await auth();
  if (!session?.user?.id) {
    return new Response("Unauthorized", { status: 401 });
  }
  const userId = session.user.id;
  const workspaceIds = await getUserWorkspaceIds(userId);

  if (workspaceIds.length === 0) {
    // No workspaces: return empty SSE that stays open with keepalive
    const stream = new ReadableStream({
      start(controller) {
        const encoder = new TextEncoder();
        controller.enqueue(encoder.encode(`: no workspaces\n\n`));
        const iv = setInterval(() => {
          try {
            controller.enqueue(encoder.encode(`: keepalive\n\n`));
          } catch {
            clearInterval(iv);
          }
        }, 15000);
      },
    });
    return new Response(stream, {
      headers: {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache, no-transform",
        Connection: "keep-alive",
      },
    });
  }

  const encoder = new TextEncoder();
  let closed = false;
  let pollInterval: ReturnType<typeof setInterval> | null = null;

  const fetchRecent = async () =>
    db
      .select()
      .from(convoyEvents)
      .where(inArray(convoyEvents.workspaceId, workspaceIds))
      .orderBy(desc(convoyEvents.occurredAt))
      .limit(30);

  const stream = new ReadableStream({
    async start(controller) {
      const send = (data: unknown) => {
        if (closed) return;
        try {
          controller.enqueue(encoder.encode(`data: ${JSON.stringify(data)}\n\n`));
        } catch {
          closed = true;
        }
      };

      const sendComment = (comment: string) => {
        if (closed) return;
        try {
          controller.enqueue(encoder.encode(`: ${comment}\n\n`));
        } catch {
          closed = true;
        }
      };

      let initial: Awaited<ReturnType<typeof fetchRecent>> = [];
      try {
        initial = await fetchRecent();
      } catch {
        initial = [];
      }

      for (const ev of [...initial].reverse()) {
        send({
          id: ev.id,
          workspaceId: ev.workspaceId,
          convoyId: ev.convoyId,
          briefId: ev.briefId,
          role: ev.role,
          actor: ev.actor,
          eventType: ev.eventType,
          harnessAlias: ev.harnessAlias,
          occurredAt: ev.occurredAt.toISOString(),
          metadata: ev.metadata,
          classification: ev.classification,
          durationS: ev.durationS,
          outcome: ev.outcome,
        });
      }

      const seen = new Set(initial.map((e) => e.id));

      const poll = async () => {
        if (closed) return;
        try {
          const rows = await fetchRecent();
          const fresh = rows.filter((r) => !seen.has(r.id));
          for (const ev of [...fresh].reverse()) {
            seen.add(ev.id);
            send({
              id: ev.id,
              workspaceId: ev.workspaceId,
              convoyId: ev.convoyId,
              briefId: ev.briefId,
              role: ev.role,
              actor: ev.actor,
              eventType: ev.eventType,
              harnessAlias: ev.harnessAlias,
              occurredAt: ev.occurredAt.toISOString(),
              metadata: ev.metadata,
              classification: ev.classification,
              durationS: ev.durationS,
              outcome: ev.outcome,
            });
          }
          if (seen.size > 400) {
            const toKeep = new Set(rows.map((r) => r.id));
            for (const id of seen) {
              if (!toKeep.has(id) && seen.size > 200) seen.delete(id);
            }
          }
        } catch {
          // ignore poll errors
        }
        if (!closed) sendComment("keepalive");
      };

      pollInterval = setInterval(poll, 5000);
    },
    cancel() {
      closed = true;
      if (pollInterval) clearInterval(pollInterval);
    },
  });

  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache, no-transform",
      Connection: "keep-alive",
      "X-Accel-Buffering": "no",
    },
  });
}