WIP new internal job system

2023-05-04 21:18:19 +01:00
parent 3e87e832c7
commit c77dfae653
32 changed files with 986 additions and 1007 deletions
@@ -3,26 +3,16 @@ import {
  IssuesEvent,
  IssuesOpenedEvent,
 } from "@octokit/webhooks-types";
-import type { Connection, EventFilter } from "@trigger.dev/sdk";
-import { ExternalSourceEventTrigger, Trigger } from "@trigger.dev/sdk/triggers";
+import {
+  Connection,
+  EventFilter,
+  ExternalSourceEventTrigger,
+} from "@trigger.dev/sdk";
 import { Octokit } from "octokit";
 import { clientFactory } from "./clientFactory";
 import { metadata } from "./metadata";
 import { repositoryWebhookSource } from "./sources";
-import {
-  createIssue,
-  createIssueComment,
-  createIssueCommentWithReaction,
-  getRepo,
-} from "./tasks";
-import { ClientOptions } from "./types";
-
-const tasks = {
-  createIssue,
-  createIssueComment,
-  getRepo,
-  createIssueCommentWithReaction,
-};
+import { tasks } from "./tasks";

 export type GitHubConnectionOptions =
  | {
@@ -33,6 +23,17 @@ export type GitHubConnectionOptions =
    };

 export const github = (options: GitHubConnectionOptions) => {
+  const connection = createConnectionFromOptions(options);
+
+  return {
+    ...connection,
+    triggers: createTriggers(connection),
+  };
+};
+
+function createConnectionFromOptions(
+  options: GitHubConnectionOptions
+): Connection<Octokit, typeof tasks> {
  if ("token" in options) {
    const client = new Octokit({
      auth: options.token,
@@ -43,8 +44,7 @@ export const github = (options: GitHubConnectionOptions) => {
      tasks,
      usesLocalAuth: true,
      client,
-      triggers: createTriggers({ usesLocalAuth: true, octokit: client }),
-    } satisfies Connection<Octokit, typeof tasks>;
+    };
  }

  return {
@@ -53,26 +53,20 @@ export const github = (options: GitHubConnectionOptions) => {
    tasks,
    usesLocalAuth: false,
    clientFactory,
-    triggers: createTriggers(
-      { usesLocalAuth: false, clientFactory },
-      options.id
-    ),
-  } satisfies Connection<Octokit, typeof tasks>;
-};
-0;
-function createTriggers(client: ClientOptions, id?: string) {
+  };
+}
+
+function createTriggers(connection: Connection<Octokit, typeof tasks>) {
  return {
    onIssue: buildRepoWebhookTrigger<IssuesEvent>(
      "On Issue",
      "issues",
-      client,
-      id
+      connection
    ),
    onIssueOpened: buildRepoWebhookTrigger<IssuesOpenedEvent>(
      "On Issue Opened",
      "issues",
-      client,
-      id,
+      connection,
      {
        action: ["opened"],
      }
@@ -80,21 +74,19 @@ function createTriggers(client: ClientOptions, id?: string) {
    onIssueComment: buildRepoWebhookTrigger<IssueCommentEvent>(
      "On Issue Comment",
      "issue_comment",
-      client,
-      id
+      connection
    ),
  };
 }

-function buildRepoWebhookTrigger<TEventType>(
+function buildRepoWebhookTrigger<TEvent>(
  title: string,
  event: string,
-  client: ClientOptions,
-  id?: string,
+  connection: Connection<Octokit, typeof tasks>,
  filter?: EventFilter
-): (params: { repo: string }) => ExternalSourceEventTrigger<TEventType> {
+) {
  return (params: { repo: string }) =>
-    new ExternalSourceEventTrigger<TEventType>({
+    new ExternalSourceEventTrigger({
      title,
      elements: [
        {
@@ -111,8 +103,8 @@ function buildRepoWebhookTrigger<TEventType>(
          repo: params.repo,
          events: [event],
        },
-        client,
-        id
+        connection,
+        (payload) => payload as TEvent
      ),
      eventRule: {
        event,
@@ -1,7 +1,7 @@
 import { Webhooks } from "@octokit/webhooks";
-import { ExternalSource } from "@trigger.dev/sdk/externalSource";
-import { metadata } from "./metadata";
-import { ClientOptions } from "./types";
+import { Connection, ExternalSource } from "@trigger.dev/sdk";
+import { Octokit } from "octokit";
+import { tasks } from "./tasks";

 type WebhookData = {
  id: number;
@@ -21,37 +21,26 @@ function webhookData(data: any): data is WebhookData {
  );
 }

-export function repositoryWebhookSource(
+export function repositoryWebhookSource<TEventType>(
  params: {
    repo: string;
    events: string[];
    secret?: string;
  },
-  client: ClientOptions,
-  id?: string
+  connection: Connection<Octokit, typeof tasks>,
+  parsePayload: (payload: any) => TEventType
 ) {
  // Create a stable key for this source so we only register it once
  const key = `github.repo.${params.repo}.webhook`;

-  return new ExternalSource("http", metadata, {
-    id,
-    usesLocalAuth: client.usesLocalAuth,
-    key,
-    register: async (triggerClient, auth) => {
-      if (!auth) {
-        throw new Error("No auth provided");
-      }
-
-      const octokit = client.usesLocalAuth
-        ? client.octokit
-        : client.clientFactory(auth);
-
-      const httpSource = await triggerClient.registerHttpSource({
+  return new ExternalSource("http", {
+    parsePayload,
+    connection,
+    register: async (io, ctx) => {
+      const httpSource = await io.registerHttpSource("register-http-source", {
        key,
      });

-      const [owner, repo] = params.repo.split("/");
-
      if (
        httpSource.active &&
        webhookData(httpSource.data) &&
@@ -70,20 +59,19 @@ export function repositoryWebhookSource(

        if (missingEvents.length > 0) {
          // We need to update the webhook to add the new events and then return
-          const { data: newWebhookData } =
-            await octokit.rest.repos.updateWebhook({
-              owner,
-              repo,
-              hook_id: existingData.id,
-              config: {
-                content_type: "json",
-                url: httpSource.url,
-                secret: httpSource.secret,
-              },
-              add_events: missingEvents,
-            });
+          const newWebhookData = await io.client.updateWebhook(
+            "update-webhook",
+            {
+              repo: params.repo,
+              hookId: existingData.id,
+              url: httpSource.url,
+              secret: httpSource.secret,
+              addEvents: missingEvents,
+            }
+          );

-          await triggerClient.updateHttpSource(httpSource.id, {
+          await io.updateHttpSource("update-http-source", {
+            id: httpSource.id,
            data: newWebhookData,
          });
        }
@@ -91,9 +79,8 @@ export function repositoryWebhookSource(
        return;
      }

-      const { data: webhooks } = await octokit.rest.repos.listWebhooks({
-        owner,
-        repo,
+      const webhooks = await io.client.listWebhooks("list-webhooks", {
+        repo: params.repo,
      });

      const existingWebhook = webhooks.find(
@@ -103,18 +90,15 @@ export function repositoryWebhookSource(
      const secret = params.secret || Math.random().toString(36).slice(2);

      if (existingWebhook && existingWebhook.active) {
-        await octokit.rest.repos.updateWebhook({
-          owner,
-          repo,
-          hook_id: existingWebhook.id,
-          config: {
-            content_type: "json",
-            url: httpSource.url,
-            secret,
-          },
+        await io.client.updateWebhook("update-webhook", {
+          repo: params.repo,
+          hookId: existingWebhook.id,
+          url: httpSource.url,
+          secret,
        });

-        await triggerClient.updateHttpSource(httpSource.id, {
+        await io.updateHttpSource("update-http-source", {
+          id: httpSource.id,
          secret,
          data: existingWebhook,
          active: true,
@@ -123,42 +107,31 @@ export function repositoryWebhookSource(
        return;
      }

-      // Generate secret
-
-      if (!owner || !repo) {
-        throw new Error(
-          'Invalid repo, should be in format "owner/repo". For example: "triggerdotdev/trigger.dev"'
-        );
-      }
-
-      const { data: webhook } = await octokit.rest.repos.createWebhook({
-        owner,
-        repo,
+      const webhook = await io.client.createWebhook("create-webhook", {
+        repo: params.repo,
        events: params.events,
-        config: {
-          url: httpSource.url,
-          content_type: "json",
-          secret,
-        },
+        url: httpSource.url,
+        secret,
      });

-      await triggerClient.updateHttpSource(httpSource.id, {
+      await io.updateHttpSource("update-http-source", {
+        id: httpSource.id,
        secret,
        data: webhook,
        active: true,
      });
    },
-    handler: async (client, source, auth) => {
-      const deliveryId = source.request.headers["x-github-delivery"];
-      const hookId = source.request.headers["x-github-hook-id"];
-      const signature = source.request.headers["x-hub-signature-256"];
+    handler: async (event, io, ctx) => {
+      const deliveryId = event.request.headers["x-github-delivery"];
+      const hookId = event.request.headers["x-github-hook-id"];
+      const signature = event.request.headers["x-hub-signature-256"];

-      if (source.secret && signature) {
+      if (event.secret && signature) {
        const githubWebhooks = new Webhooks({
-          secret: source.secret,
+          secret: event.secret,
        });

-        if (!githubWebhooks.verify(source.request.body, signature)) {
+        if (!githubWebhooks.verify(event.request.body, signature)) {
          return {
            events: [],
            response: {
@@ -171,9 +144,9 @@ export function repositoryWebhookSource(
        }
      }

-      const name = source.request.headers["x-github-event"];
+      const name = event.request.headers["x-github-event"];

-      const context = omit(source.request.headers, [
+      const context = omit(event.request.headers, [
        "x-github-event",
        "x-github-delivery",
        "x-hub-signature-256",
@@ -185,7 +158,7 @@ export function repositoryWebhookSource(
        "x-forwarded-proto",
      ]);

-      const payload = parseBody(source.request.body);
+      const payload = parseBody(event.request.body);

      return {
        events: [
@@ -238,3 +238,135 @@ export const createIssueCommentWithReaction = authenticatedTask({
    };
  },
 });
+
+export const updateWebhook = authenticatedTask({
+  run: async (
+    params: {
+      repo: string;
+      hookId: number;
+      url: string;
+      secret: string;
+      addEvents?: string[];
+    },
+    client: InstanceType<typeof Octokit>,
+    task
+  ) => {
+    const [owner, repo] = params.repo.split("/");
+
+    return client.rest.repos
+      .updateWebhook({
+        owner,
+        repo,
+        hook_id: params.hookId,
+        config: {
+          content_type: "json",
+          url: params.url,
+          secret: params.secret,
+        },
+        add_events: params.addEvents,
+      })
+      .then((response) => response.data);
+  },
+  init: (params) => {
+    return {
+      name: "Update Webhook",
+      params,
+      elements: [
+        {
+          label: "Repo",
+          text: params.repo,
+        },
+        {
+          label: "Hook ID",
+          text: String(params.hookId),
+        },
+      ],
+    };
+  },
+});
+
+export const createWebhook = authenticatedTask({
+  run: async (
+    params: {
+      repo: string;
+      url: string;
+      secret: string;
+      events: string[];
+    },
+    client: InstanceType<typeof Octokit>,
+    task
+  ) => {
+    const [owner, repo] = params.repo.split("/");
+
+    return client.rest.repos
+      .createWebhook({
+        owner,
+        repo,
+        config: {
+          content_type: "json",
+          url: params.url,
+          secret: params.secret,
+        },
+        events: params.events,
+      })
+      .then((response) => response.data);
+  },
+  init: (params) => {
+    return {
+      name: "Create Webhook",
+      params,
+      elements: [
+        {
+          label: "Repo",
+          text: params.repo,
+        },
+        {
+          label: "Events",
+          text: params.events.join(", "),
+        },
+      ],
+    };
+  },
+});
+
+export const listWebhooks = authenticatedTask({
+  run: async (
+    params: {
+      repo: string;
+    },
+    client: InstanceType<typeof Octokit>,
+    task
+  ) => {
+    const [owner, repo] = params.repo.split("/");
+
+    return client.rest.repos
+      .listWebhooks({
+        owner,
+        repo,
+      })
+      .then((response) => response.data);
+  },
+  init: (params) => {
+    return {
+      name: "List Webhooks",
+      params,
+      elements: [
+        {
+          label: "Repo",
+          text: params.repo,
+        },
+      ],
+    };
+  },
+});
+
+export const tasks = {
+  createIssue,
+  createIssueComment,
+  getRepo,
+  createIssueCommentWithReaction,
+  addIssueCommentReaction,
+  updateWebhook,
+  createWebhook,
+  listWebhooks,
+};
@@ -1,12 +0,0 @@
-import { ClientFactory } from "@trigger.dev/sdk";
-import { Octokit } from "octokit";
-
-export type ClientOptions =
-  | {
-      usesLocalAuth: true;
-      octokit: Octokit;
-    }
-  | {
-      usesLocalAuth: false;
-      clientFactory: ClientFactory<Octokit>;
-    };