Doing real authorization of API keys now so we can get back the organization ID

I’ve also started to add pulsar integration with the webapp, to get messages from coordinators

README.md

@@ -13,11 +13,10 @@
    ```sh
    cp ./apps/webapp/.env.example ./apps/webapp/.env
    ```
-3. Start postgresql and apache pulsar
+3. Start postgresql
 
    ```bash
    pnpm run docker:db
-   pnpm run docker:pulsar
    ```
 
    > **Note:** The npm script will complete while Docker sets up the container in the background. Ensure that Docker has finished and your container is running before proceeding.
@@ -36,8 +35,29 @@
    ```
    **Running simply `pnpm run build` will build everything, including the NextJS app.**
 7. Run the Remix dev server
+
+```bash
+pnpm run dev --filter=webapp
+```
+
+## Starting and Stopping Pulsar
+
+Both the webapp and coordinate apps rely on Apache Pulsar running on your local machine.
+
+1. Run the pulsar container
+   In a separate terminal window, run the following command:
    ```bash
-   pnpm run dev --filter=webapp
+   ./pulsar/start.sh
+   ```
+2. Wait until pulsar is available
+   In yet another terminal window, run this command and when it's finished pulsar will be ready
+   ```bash
+   until curl http://localhost:8080/admin/v2/brokers/internal-configuration > /dev/null 2>&1 ; do sleep 1; done
+   ```
+3. Stop pulsar
+   In the terminal window where you ran `./pulsar/start.sh`, go ahead and CTRL-C and then run
+   ```bash
+   ./pulsar/stop.sh
    ```
 
 ## Tests, Typechecks, Lint, Install packages...

apps/coordinator/src/env.ts

@@ -0,0 +1,13 @@
+import { z } from "zod";
+
+const EnvironmentSchema = z.object({
+  PORT: z.coerce.number().finite().default(8889),
+  PULSAR_URL: z.string().default("pulsar://localhost:6650"),
+  AUTHORIZATION_URL: z
+    .string()
+    .default("http://localhost:3000/api/v1/internal/authorizeKey"),
+});
+
+export type Environment = z.infer<typeof EnvironmentSchema>;
+
+export const env = EnvironmentSchema.parse(process.env);

apps/coordinator/src/index.ts

@@ -1,5 +1,6 @@
 import { createServer } from "node:http";
 import { WebSocketServer } from "ws";
+import { env } from "./env";
 import { TriggerServer } from "./server";
 
 // Create an HTTP server
@@ -57,7 +58,7 @@ server.on("upgrade", async (req, socket, head) => {
 });
 
 // Listen on port from env
-const port = process.env.PORT ?? 8089;
+const port = env.PORT;
 server.listen(port, () => {
   console.log(`Listening on port ${port}`);
 });

apps/coordinator/src/pulsarClient.ts

@@ -1,5 +1,6 @@
 import Pulsar from "pulsar-client";
+import { env } from "./env";
 
 export const pulsarClient = new Pulsar.Client({
-  serviceUrl: process.env.PULSAR_URL ?? "pulsar://localhost:6650",
+  serviceUrl: env.PULSAR_URL,
 });

apps/coordinator/src/server.ts

@@ -9,6 +9,7 @@ import {
   Logger,
 } from "internal-bridge";
 import { WorkflowMessageBroker } from "./messageBroker";
+import { env } from "./env";
 
 export class TriggerServer {
   #connection?: TriggerServerConnection;
@@ -213,6 +214,7 @@ export const sleep = (ms: number) =>
 type AuthorizationSuccess = {
   authorized: true;
   organizationId: string;
+  env: string;
 };
 
 type AuthorizationFailure = {
@@ -229,9 +231,38 @@ async function authorizeApiKey(
     return { authorized: false, reason: "Missing API key" };
   }
 
-  if (apiKey === "trigger_123") {
-    return { authorized: true, organizationId: "123" };
+  return performAuthorizationRequest(apiKey);
+}
+
+async function performAuthorizationRequest(
+  apiKey: string
+): Promise<AuthorizationResponse> {
+  const response = await fetch(env.AUTHORIZATION_URL, {
+    method: "GET",
+    headers: {
+      Accept: "application/json",
+      Authorization: `Bearer ${apiKey}`,
+    },
+  });
+
+  if (response.ok) {
+    const body = await response.json();
+
+    return {
+      authorized: true,
+      organizationId: body.organizationId,
+      env: body.env,
+    };
   }
 
-  return { authorized: false, reason: "Invalid API key" };
+  if (response.status === 401) {
+    const errorBody = await response.json();
+
+    return { authorized: false, reason: errorBody.error };
+  }
+
+  return {
+    authorized: false,
+    reason: `[${response.status}] Something went wrong: ${response.statusText}`,
+  };
 }

apps/webapp/app/entry.server.tsx

@@ -2,6 +2,7 @@ import type { EntryContext } from "@remix-run/node";
 import { RemixServer } from "@remix-run/react";
 import { renderToString } from "react-dom/server";
 import * as Sentry from "@sentry/remix";
+import * as MessageBroker from "~/services/messageBroker.server";
 import { prisma } from "./db.server";
 
 export default function handleRequest(
@@ -33,3 +34,5 @@ if (process.env.NODE_ENV === "production" && process.env.SENTRY_DSN) {
 
   console.log("🚦 Sentry initialized");
 }
+
+MessageBroker.init();

apps/webapp/app/env.server.ts

@@ -25,6 +25,7 @@ const EnvironmentSchema = z.object({
   FLY_REGION: z.string().optional(),
   SESSION_SECRET: z.string(),
   PIZZLY_HOST: z.string(),
+  PULSAR_URL: z.string().default("pulsar://localhost:6650"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/models/runtimeEnvironment.server.ts

@@ -0,0 +1,17 @@
+import type { RuntimeEnvironment } from ".prisma/client";
+import { prisma } from "~/db.server";
+
+export type { RuntimeEnvironment };
+
+export async function findEnvironmentByApiKey(apiKey: string) {
+  const environment = await prisma.runtimeEnvironment.findUnique({
+    where: {
+      apiKey,
+    },
+    include: {
+      organization: true,
+    },
+  });
+
+  return environment;
+}

apps/webapp/app/routes/api/v1/internal/authorizeKey.ts

@@ -0,0 +1,31 @@
+import type { LoaderArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { findEnvironmentByApiKey } from "~/models/runtimeEnvironment.server";
+
+const AuthorizationHeaderSchema = z.string().regex(/^Bearer .+$/);
+
+// Get the API key from the request headers,
+// Then lookup the organization ID from the API key in the RuntimeEnvironment
+export async function loader({ request }: LoaderArgs) {
+  const rawAuthorization = request.headers.get("Authorization");
+
+  const authorization = AuthorizationHeaderSchema.safeParse(rawAuthorization);
+
+  if (!authorization.success) {
+    return json({ error: "Missing or invalid API key" }, { status: 401 });
+  }
+
+  const apiKey = authorization.data.replace(/^Bearer /, "");
+
+  const environment = await findEnvironmentByApiKey(apiKey);
+
+  if (!environment) {
+    return json({ error: "Invalid API Key" }, { status: 401 });
+  }
+
+  return json({
+    organizationId: environment.organizationId,
+    env: environment.slug,
+  });
+}

apps/webapp/app/services/messageBroker.server.ts

@@ -0,0 +1,31 @@
+import type { PulsarMessage, PulsarConsumer } from "./pulsarClient.server";
+import { pulsarClient } from "./pulsarClient.server";
+
+export async function init() {
+  const workflowsMetaProducer = await pulsarClient.subscribe({
+    topic: "workflows-meta",
+    subscription: "webapp",
+    subscriptionType: "Shared",
+    ackTimeoutMs: 30000,
+    listener: async (msg, consumer) => {
+      await receiveMetadata(msg, consumer);
+    },
+  });
+
+  console.log("📡 Message Broker initialized");
+
+  process.on("beforeExit", () => {
+    workflowsMetaProducer.close();
+
+    console.log("📡 Message Broker closed");
+  });
+}
+
+async function receiveMetadata(msg: PulsarMessage, consumer: PulsarConsumer) {
+  const data = JSON.parse(msg.getData().toString());
+  const properties = msg.getProperties();
+
+  console.log("Received metadata", data, properties);
+
+  await consumer.acknowledge(msg);
+}

apps/webapp/app/services/pulsarClient.server.ts

@@ -0,0 +1,34 @@
+import Pulsar from "pulsar-client";
+import type {
+  Producer as PulsarProducer,
+  Consumer as PulsarConsumer,
+  Message as PulsarMessage,
+} from "pulsar-client";
+
+import { env } from "~/env.server";
+
+let pulsarClient: Pulsar.Client;
+
+declare global {
+  var __pulsarClient__: typeof pulsarClient;
+}
+
+if (env.NODE_ENV === "production") {
+  pulsarClient = getClient();
+} else {
+  if (!global.__pulsarClient__) {
+    global.__pulsarClient__ = getClient();
+  }
+  pulsarClient = global.__pulsarClient__;
+}
+
+function getClient() {
+  const client = new Pulsar.Client({
+    serviceUrl: env.PULSAR_URL,
+  });
+
+  return client;
+}
+
+export { pulsarClient };
+export type { PulsarProducer, PulsarConsumer, PulsarMessage };

examples/smoke-test/src/index.ts

@@ -8,7 +8,7 @@ const userCreatedEvent = z.object({
 const workflow = new Workflow({
   id: "my-workflow",
   name: "My workflow",
-  apiKey: "trigger_123",
+  apiKey: "trigger_dev_zC25mKNn6c0q",
   endpoint: "ws://localhost:8889/ws",
   logLevel: "debug",
   trigger: onEvent({ name: "user.created", schema: userCreatedEvent }),