run create immediately (#23)

Co-authored-by: Shawn Erquhart <shawn@erquh.art>

Author: ianmacartney

.cursor/rules/convex_rules.mdc

@@ -212,7 +212,6 @@ export const exampleQuery = query({
 - Always use `as const` for string literals in discriminated union types.
 - When using the `Array` type, make sure to always define your arrays as `const array: Array<T> = [...];`
 - When using the `Record` type, make sure to always define your records as `const record: Record<KeyType, ValueType> = {...};`
-- Always add `@types/node` to your `package.json` when using any Node.js built-in modules.
 
 ## Full text search guidelines
 - A query for "10 messages in channel '#general' that best match the query 'hello hi' in their body" would look like:

example/convex/_generated/api.d.ts

@@ -51,20 +51,6 @@ export declare const internal: FilterApi<
 export declare const components: {
   workOSAuthKit: {
     lib: {
-      enqueueWebhookEvent: FunctionReference<
-        "mutation",
-        "internal",
-        {
-          apiKey: string;
-          event: string;
-          eventId: string;
-          eventTypes?: Array<string>;
-          logLevel?: "DEBUG";
-          onEventHandle?: string;
-          updatedAt?: string;
-        },
-        any
-      >;
       getAuthUser: FunctionReference<
         "query",
         "internal",
@@ -84,6 +70,24 @@ export declare const components: {
           updatedAt: string;
         } | null
       >;
+      onWebhookEvent: FunctionReference<
+        "mutation",
+        "internal",
+        {
+          apiKey: string;
+          event: {
+            context?: Record<string, any>;
+            createdAt: string;
+            data: Record<string, any>;
+            event: string;
+            id: string;
+          };
+          eventTypes?: Array<string>;
+          logLevel?: "DEBUG";
+          onEventHandle?: string;
+        },
+        null
+      >;
     };
   };
 };

example/convex/tsconfig.json

@@ -1,33 +1,24 @@
 {
   /* This TypeScript project config describes the environment that
    * Convex functions run in and is used to typecheck them.
-   * You can modify it, but some settings required to use Convex.
+   * You can modify it, but some settings are required to use Convex.
    */
   "compilerOptions": {
     /* These settings are not required by Convex and can be modified. */
     "allowJs": true,
     "strict": true,
+    "moduleResolution": "Bundler",
+    "jsx": "react-jsx",
     "skipLibCheck": true,
-    "verbatimModuleSyntax": true,
+    "allowSyntheticDefaultImports": true,
 
     /* These compiler options are required by Convex */
     "target": "ESNext",
     "lib": ["ES2021", "dom", "ESNext.Array"],
     "forceConsistentCasingInFileNames": true,
-    "allowSyntheticDefaultImports": true,
     "module": "ESNext",
-    "moduleResolution": "Bundler",
     "isolatedModules": true,
     "noEmit": true
-
-    /* This should only be used in this example. Real apps should not attempt
-     * to compile TypeScript because differences between tsconfig.json files can
-     * cause the code to be compiled differently.
-     */
-    // Un-comment this to get instant types between your component and example.
-    // However, if you're willing to wait for a build before the types update,
-    // it's better to leave this commented out to catch build errors faster.
-    // "customConditions": ["@convex-dev/component-source"]
   },
   "include": ["./**/*"],
   "exclude": ["./_generated"]

src/client/index.ts

@@ -18,7 +18,9 @@ import {
 } from "@workos-inc/node";
 import type { SetRequired } from "type-fest";
 import { v } from "convex/values";
+import { parse } from "convex-helpers/validators";
 import type { ComponentApi } from "../component/_generated/component.js";
+import { vEvent } from "../component/lib.js";
 
 type WorkOSResponsePayload =
   | AuthenticationActionResponseData
@@ -226,15 +228,12 @@ export class AuthKit<DataModel extends GenericDataModel> {
         if (this.config.logLevel === "DEBUG") {
           console.log("received event", event);
         }
-        await ctx.runMutation(this.component.lib.enqueueWebhookEvent, {
+        await ctx.runMutation(this.component.lib.onWebhookEvent, {
           apiKey: this.config.apiKey,
-          eventId: event.id,
-          event: event.event,
+          event: parse(vEvent, event),
           onEventHandle: this.config.authFunctions?.authKitEvent
             ? await createFunctionHandle(this.config.authFunctions.authKitEvent)
             : undefined,
-          updatedAt:
-            "updated_at" in event ? (event.updated_at as string) : undefined,
           eventTypes: this.config.additionalEventTypes,
           logLevel: this.config.logLevel,
         });

src/component/_generated/component.ts

@@ -24,21 +24,6 @@ import type { FunctionReference } from "convex/server";
 export type ComponentApi<Name extends string | undefined = string | undefined> =
   {
     lib: {
-      enqueueWebhookEvent: FunctionReference<
-        "mutation",
-        "internal",
-        {
-          apiKey: string;
-          event: string;
-          eventId: string;
-          eventTypes?: Array<string>;
-          logLevel?: "DEBUG";
-          onEventHandle?: string;
-          updatedAt?: string;
-        },
-        any,
-        Name
-      >;
       getAuthUser: FunctionReference<
         "query",
         "internal",
@@ -59,5 +44,24 @@ export type ComponentApi<Name extends string | undefined = string | undefined> =
         } | null,
         Name
       >;
+      onWebhookEvent: FunctionReference<
+        "mutation",
+        "internal",
+        {
+          apiKey: string;
+          event: {
+            context?: Record<string, any>;
+            createdAt: string;
+            data: Record<string, any>;
+            event: string;
+            id: string;
+          };
+          eventTypes?: Array<string>;
+          logLevel?: "DEBUG";
+          onEventHandle?: string;
+        },
+        null,
+        Name
+      >;
     };
   };

src/component/lib.ts

@@ -1,11 +1,12 @@
-import { v } from "convex/values";
+import { type Infer, v } from "convex/values";
 import {
   internalAction,
   internalMutation,
   internalQuery,
   mutation,
   query,
 } from "./_generated/server.js";
+import type { MutationCtx } from "./_generated/server.js";
 import { components, internal } from "./_generated/api.js";
 import { omit, withoutSystemFields } from "convex-helpers";
 import { WorkOS, type Event as WorkOSEvent } from "@workos-inc/node";
@@ -18,32 +19,117 @@ const eventWorkpool = new Workpool(components.eventWorkpool, {
   maxParallelism: 1,
 });
 
-const vEvent = v.object({
+export const vEvent = v.object({
   id: v.string(),
   createdAt: v.string(),
   event: v.string(),
   data: v.record(v.string(), v.any()),
   context: v.optional(v.record(v.string(), v.any())),
 });
 
-export const enqueueWebhookEvent = mutation({
+async function processEventHandler(
+  ctx: MutationCtx,
+  args: {
+    event: Infer<typeof vEvent>;
+    logLevel?: "DEBUG";
+    onEventHandle?: string;
+  }
+) {
+  if (args.logLevel === "DEBUG") {
+    console.log("processing event", args.event);
+  }
+  const dbEvent = await ctx.db
+    .query("events")
+    .withIndex("eventId", (q) => q.eq("eventId", args.event.id))
+    .unique();
+  if (dbEvent) {
+    console.log("event already processed", args.event.id);
+    return;
+  }
+  await ctx.db.insert("events", {
+    eventId: args.event.id,
+    event: args.event.event,
+    updatedAt: args.event.data.updatedAt as string | undefined,
+  });
+  const event = args.event as WorkOSEvent;
+  switch (event.event) {
+    case "user.created": {
+      const data = omit(event.data, ["object"]);
+      const existingUser = await ctx.db
+        .query("users")
+        .withIndex("id", (q) => q.eq("id", data.id))
+        .unique();
+      if (existingUser) {
+        console.warn("user already exists", data.id);
+        break;
+      }
+      await ctx.db.insert("users", data);
+      break;
+    }
+    case "user.updated": {
+      const data = omit(event.data, ["object"]);
+      const user = await ctx.db
+        .query("users")
+        .withIndex("id", (q) => q.eq("id", data.id))
+        .unique();
+      if (!user) {
+        console.error("user not found", data.id);
+        break;
+      }
+      if (user.updatedAt >= data.updatedAt) {
+        console.warn(`user already updated for event ${event.id}, skipping`);
+        break;
+      }
+      await ctx.db.patch(user._id, data);
+      break;
+    }
+    case "user.deleted": {
+      const data = omit(event.data, ["object"]);
+      const user = await ctx.db
+        .query("users")
+        .withIndex("id", (q) => q.eq("id", data.id))
+        .unique();
+      if (!user) {
+        console.warn("user not found", data.id);
+        break;
+      }
+      await ctx.db.delete(user._id);
+      break;
+    }
+  }
+  if (args.onEventHandle) {
+    await ctx.runMutation(args.onEventHandle as FunctionHandle<"mutation">, {
+      event: args.event.event,
+      data: args.event.data,
+    });
+  }
+}
+
+export const onWebhookEvent = mutation({
   args: {
     apiKey: v.string(),
-    eventId: v.string(),
-    event: v.string(),
-    updatedAt: v.optional(v.string()),
+    event: vEvent,
     onEventHandle: v.optional(v.string()),
     eventTypes: v.optional(v.array(v.string())),
     logLevel: v.optional(v.literal("DEBUG")),
   },
+  returns: v.null(),
   handler: async (ctx, args) => {
-    await eventWorkpool.cancelAll(ctx);
-    await eventWorkpool.enqueueAction(ctx, internal.lib.updateEvents, {
-      apiKey: args.apiKey,
-      onEventHandle: args.onEventHandle,
-      eventTypes: args.eventTypes,
-      logLevel: args.logLevel,
-    });
+    const isCreateEvent = args.event.event.endsWith(".created");
+
+    if (isCreateEvent) {
+      // Process create events immediately
+      await processEventHandler(ctx, args);
+    } else {
+      // Enqueue update/delete events to workpool
+      await eventWorkpool.enqueueAction(ctx, internal.lib.updateEvents, {
+        apiKey: args.apiKey,
+        onEventHandle: args.onEventHandle,
+        eventTypes: args.eventTypes,
+        logLevel: args.logLevel,
+      });
+    }
+    return null;
   },
 });
 
@@ -67,6 +153,9 @@ export const updateEvents = internalAction({
     logLevel: v.optional(v.literal("DEBUG")),
   },
   handler: async (ctx, args) => {
+    // Cancel other pending workpool jobs since this run will
+    // process all available events from the WorkOS API.
+    await eventWorkpool.cancelAll(ctx);
     const workos = new WorkOS(args.apiKey);
     const cursor = await ctx.runQuery(internal.lib.getCursor);
     let nextCursor = cursor ?? undefined;
@@ -107,74 +196,7 @@ export const processEvent = internalMutation({
     onEventHandle: v.optional(v.string()),
   },
   handler: async (ctx, args) => {
-    if (args.logLevel === "DEBUG") {
-      console.log("processing event", args.event);
-    }
-    const dbEvent = await ctx.db
-      .query("events")
-      .withIndex("eventId", (q) => q.eq("eventId", args.event.id))
-      .unique();
-    if (dbEvent) {
-      console.log("event already processed", args.event.id);
-      return;
-    }
-    await ctx.db.insert("events", {
-      eventId: args.event.id,
-      event: args.event.event,
-      updatedAt: args.event.data.updatedAt,
-    });
-    const event = args.event as WorkOSEvent;
-    switch (event.event) {
-      case "user.created": {
-        const data = omit(event.data, ["object"]);
-        const existingUser = await ctx.db
-          .query("users")
-          .withIndex("id", (q) => q.eq("id", data.id))
-          .unique();
-        if (existingUser) {
-          console.warn("user already exists", data.id);
-          break;
-        }
-        await ctx.db.insert("users", data);
-        break;
-      }
-      case "user.updated": {
-        const data = omit(event.data, ["object"]);
-        const user = await ctx.db
-          .query("users")
-          .withIndex("id", (q) => q.eq("id", data.id))
-          .unique();
-        if (!user) {
-          console.error("user not found", data.id);
-          break;
-        }
-        if (user.updatedAt >= data.updatedAt) {
-          console.warn(`user already updated for event ${event.id}, skipping`);
-          break;
-        }
-        await ctx.db.patch(user._id, data);
-        break;
-      }
-      case "user.deleted": {
-        const data = omit(event.data, ["object"]);
-        const user = await ctx.db
-          .query("users")
-          .withIndex("id", (q) => q.eq("id", data.id))
-          .unique();
-        if (!user) {
-          console.warn("user not found", data.id);
-          break;
-        }
-        await ctx.db.delete(user._id);
-        break;
-      }
-    }
-    if (args.onEventHandle) {
-      await ctx.runMutation(args.onEventHandle as FunctionHandle<"mutation">, {
-        event: args.event.event,
-        data: args.event.data,
-      });
-    }
+    await processEventHandler(ctx, args);
   },
 });