Airtable integration with webhook and integration changes (#399)

* CLI create-integration command now accepts an Open AI api key

* Create integration docs separated into multiple pages

* Initial Airtable integration commit, with OpenAI generated code

* OAuth page coming soon

* Export DisplayProperty from the SDK

* TSConfig made to match GitHub’s with paths

* getRecords

* Removed duplicate Stripe job from the catalog

* Renamed Airtable apiKey option to token

* First Airtable job

* Export Collaborator and Attachment field types

* A typesafe example that uses runTask

* WIP on new integration tasks… not working yet

* Attempt with class

* Revert "Attempt with class"

This reverts commit 93a4833.

* WIP changing how tasks work

* Mock of async local storage

* Moved client creation from constructor

* New approach with a clone method on TriggerIntegration

* Added runTask to Airtable which is used by integration tasks

* Added the Airtable icon and connection when using runTask

* base().table() is working

* runTask options moved to the 3rd param, made optional with optional name

* Added some generic arguments

* Added generic type to table

* Removed old comment

* We don’t need to repeat the icon

* getRecords and getRecord now returning the right data and types

* Creating records

* Update records

* Delete records

* The internal properties of integrations are now hidden by the TypeScript types

* Sprinkled a Prettify in there

* Improved the types

* Added Airtable to the integration catalog

* Early work on Airtable webhook registration

* More progress with webhooks

* connectionKey needs to be cloned for webhooks to work

* connectionKey needs to be cloned for webhooks to work

* It was unclear that the ActivateSourceService was using a graphileJob id

* ActivateSourceService optionally takes a jobId, if missing it generate a unique id

* When retrying trigger registration, don’t pass an id so it is generated

* Removed Airtable webhooks tasks from the job-catalog example

* Added TriggerSourceOption, removed TriggerSourceEvent

* WIP with new ExternalSource options

* ExternalSourceTrigger setup

* DynamicTrigger changed to options, will need some more work

* filter gets options passed to it

* SourceMetadata v2 renamed to SourceMetadataV2, kept original

* Started versioning the backend

* Moved param order on io.getEvent and io.cancelEvent

* The runTask stuff that allows unknown to work is back

* Indexing for v1 and v2, with version on “activateSource” schema

* Added todos, to deal with Airtable SDK calls inside the webhook handler

* “deliverHttpSourceRequest” queueName changed to the source id so they process in order

* ActivateSource changes to deal with old and new data formats

* Update existing TriggerSources to v2

* Fix for dynamic.ts typescript errors, need to revisit this later

* UpdateSourceService v1 and v2, with new v2 API endpoint

* Removed unused imports

* More progress on v1 and v2

* Airtable webhooks are now triggering a job

* Moved webhooks to a new file

* You can do API calls in the webhook handler now, Airtable webhook data is being processed

* Airtable events coming through

* Defined the Airtable table payload type

* TriggerSource metadata is being stored and used

* Removed some logs

* Added filtering and don’t allow any webhooks that use automated sources

* Resend switched to new integration

* Moved Resend test jobs to the catalog, and tested it worked

* WIP on Slack, there are compile errors

* Created a generic type that strips out indexes

* Slack updated to use new integration

* SendGrid migrated over

* Integration runTask is now allowing regular types

* Changed io.runTask types so it only allows Json-able types

* OpenAI models tasks working

* Added Airtable changes to runTask

* Removed the index signature crap from the Slack integration

* Don’t need to cast the callback result

* Updated Resend

* Re-ordered runTask params

* WIP on openai

* onAccountUpdated is Connect only

* Removed RunTaskResult

* Handle Resend errors, the official SDK doesn’t expose them properly at the moment

* Removed OmitIndexSignature

* OpenAI converted to new integration, with backwards compatible functions

* Put the openai catalog back to what it was originally

* Export a standard retry with backoff, to be used

* Use the standard exponential backoff in the integrations

* Retry options moved earlier so they can be overriden by a task

* GitHub tasks migrated

* Added sources, fixed one bundling issue

* Added GitHub jobs to catalog

* Remove duplicate options

* Deduplicate events

* Removed duplicate Job

* Switched Plain over

* Set the Plain icon

* Converted Stripe over

* Supabase adapted

* Typeform working

* Added dynamic-schedule to catalog

* Added background-fetch job catalog

* Created dynamic-triggers catalog file

* Fixed old general file with runTask param order

* Dynamic triggers working

* SendGrid updated to use the same tsconfig as other integrations

* Removed Airtable webhook, until we have batch support

* Added OAuth airtable auth example

* Created beta changeset tag

* Beta changesets for most packages

---------

Co-authored-by: Eric Allam <eallam@icloud.com>

Author: matt-aitken

.changeset/healthy-yaks-chew.md

@@ -0,0 +1,17 @@
+---
+"@trigger.dev/airtable": minor
+"@trigger.dev/sendgrid": minor
+"@trigger.dev/supabase": minor
+"@trigger.dev/typeform": minor
+"@trigger.dev/sdk": minor
+"@trigger.dev/github": minor
+"@trigger.dev/openai": minor
+"@trigger.dev/resend": minor
+"@trigger.dev/stripe": minor
+"@trigger.dev/plain": minor
+"@trigger.dev/slack": minor
+"@trigger.dev/core": minor
+"@trigger.dev/cli": minor
+---
+
+Integrations are now simpler and support authentication during webhook registration

.changeset/pre.json

@@ -0,0 +1,29 @@
+{
+  "mode": "pre",
+  "tag": "beta",
+  "initialVersions": {
+    "webapp": "1.0.0",
+    "@trigger.dev/airtable": "2.0.10",
+    "@trigger.dev/github": "2.0.14",
+    "@trigger.dev/openai": "2.0.14",
+    "@trigger.dev/plain": "2.0.14",
+    "@trigger.dev/resend": "2.0.14",
+    "@trigger.dev/sendgrid": "2.0.14",
+    "@trigger.dev/slack": "2.0.14",
+    "@trigger.dev/stripe": "2.0.14",
+    "@trigger.dev/supabase": "2.0.14",
+    "@trigger.dev/typeform": "2.0.14",
+    "@trigger.dev/astro": "2.0.14",
+    "@trigger.dev/cli": "2.0.14",
+    "@trigger.dev/core": "2.0.14",
+    "@trigger.dev/database": "0.0.0",
+    "emails": "1.0.0",
+    "@trigger.dev/eslint-plugin": "2.0.14",
+    "@trigger.dev/express": "2.0.14",
+    "@trigger.dev/integration-kit": "2.0.14",
+    "@trigger.dev/nextjs": "2.0.14",
+    "@trigger.dev/react": "2.0.14",
+    "@trigger.dev/sdk": "2.0.14"
+  },
+  "changesets": []
+}

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

@@ -0,0 +1,61 @@
+import { ExternalAccount, Integration, TriggerSource } from "@trigger.dev/database";
+import { ConnectionAuth } from "@trigger.dev/core";
+import { PrismaClientOrTransaction } from "~/db.server";
+import { integrationAuthRepository } from "~/services/externalApis/integrationAuthRepository.server";
+import { logger } from "~/services/logger.server";
+
+type ResolvableTriggerSource = TriggerSource & {
+  integration: Integration;
+  externalAccount: ExternalAccount | null;
+};
+
+export async function resolveSourceConnection(
+  tx: PrismaClientOrTransaction,
+  source: ResolvableTriggerSource
+): Promise<ConnectionAuth | undefined> {
+  if (source.integration.authSource !== "HOSTED") return;
+
+  const connection = await getConnection(tx, source);
+
+  if (!connection) {
+    logger.error(
+      `Integration connection not found for source ${source.id}, integration ${source.integration.id}`
+    );
+    return;
+  }
+
+  const response = await integrationAuthRepository.getCredentials(connection);
+
+  if (!response) {
+    return;
+  }
+
+  return {
+    type: "oauth2",
+    scopes: response.scopes,
+    accessToken: response.accessToken,
+  };
+}
+
+function getConnection(tx: PrismaClientOrTransaction, source: ResolvableTriggerSource) {
+  if (source.externalAccount) {
+    return tx.integrationConnection.findFirst({
+      where: {
+        integrationId: source.integration.id,
+        externalAccountId: source.externalAccount.id,
+      },
+      include: {
+        dataReference: true,
+      },
+    });
+  }
+
+  return tx.integrationConnection.findFirst({
+    where: {
+      integrationId: source.integration.id,
+    },
+    include: {
+      dataReference: true,
+    },
+  });
+}

apps/webapp/app/presenters/TriggerSourcePresenter.server.ts

@@ -62,6 +62,22 @@ export class TriggerSourcePresenter {
             },
           },
         },
+        dynamicTrigger: {
+          select: {
+            id: true,
+            slug: true,
+            sourceRegistrationJob: {
+              select: {
+                job: {
+                  select: {
+                    id: true,
+                    slug: true,
+                  },
+                },
+              },
+            },
+          },
+        },
       },
       where: {
         id: triggerSourceId,
@@ -73,10 +89,15 @@ export class TriggerSourcePresenter {
     }
 
     const runListPresenter = new RunListPresenter(this.#prismaClient);
-    const runList = trigger.sourceRegistrationJob
+    const jobSlug = getJobSlug(
+      trigger.sourceRegistrationJob?.job.slug,
+      trigger.dynamicTrigger?.sourceRegistrationJob?.job.slug
+    );
+
+    const runList = jobSlug
       ? await runListPresenter.call({
           userId,
-          jobSlug: trigger.sourceRegistrationJob.job.slug,
+          jobSlug,
           organizationSlug,
           projectSlug,
           direction,
@@ -95,7 +116,21 @@ export class TriggerSourcePresenter {
         params: trigger.params,
         registrationJob: trigger.sourceRegistrationJob?.job,
         runList,
+        dynamic: trigger.dynamicTrigger
+          ? { id: trigger.dynamicTrigger.id, slug: trigger.dynamicTrigger.slug }
+          : undefined,
       },
     };
   }
 }
+
+function getJobSlug(
+  sourceRegistrationJobSlug: string | undefined,
+  dynamicSourceRegistrationJobSlug: string | undefined
+) {
+  if (sourceRegistrationJobSlug) {
+    return sourceRegistrationJobSlug;
+  }
+
+  return dynamicSourceRegistrationJobSlug;
+}

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam/route.tsx

@@ -43,6 +43,7 @@ import { parse } from "@conform-to/zod";
 import { z } from "zod";
 import { ActivateSourceService } from "~/services/sources/activateSource.server";
 import { redirectWithSuccessMessage } from "~/models/message.server";
+import { nanoid } from "nanoid";
 
 export const loader = async ({ request, params }: LoaderArgs) => {
   const user = await requireUser(request);
@@ -90,7 +91,7 @@ export const action: ActionFunction = async ({ request, params }) => {
   try {
     const service = new ActivateSourceService();
 
-    const result = await service.call(triggerParam, submission.value.jobId);
+    const result = await service.call(triggerParam);
 
     return redirectWithSuccessMessage(
       externalTriggerPath({ slug: organizationSlug }, { slug: projectParam }, { id: triggerParam }),
@@ -167,6 +168,17 @@ export default function Page() {
                 <NamedIcon name={trigger.active ? "active" : "inactive"} className="h-4 w-4" />
               }
             />
+            {trigger.dynamic && (
+              <PageInfoProperty
+                label="Dynamic"
+                value={
+                  <span className="flex items-center gap-0.5">
+                    <NamedIcon name="dynamic" className="h-4 w-4" />
+                    {trigger.dynamic.slug}
+                  </span>
+                }
+              />
+            )}
             <PageInfoProperty
               label="Environment"
               value={<EnvironmentLabel environment={trigger.environment} />}
@@ -206,7 +218,7 @@ export default function Page() {
                   </Button>
                 </Callout>
               </Form>
-            ) : (
+            ) : trigger.dynamic ? null : (
               <Callout variant="error" className="justiy-between mb-4 items-center">
                 This External Trigger hasn't registered successfully. Contact support for help:{" "}
                 {trigger.id}

apps/webapp/app/routes/api.v1.$endpointSlug.sources.$id.ts

@@ -1,10 +1,10 @@
 import type { ActionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import { UpdateTriggerSourceBodySchema } from "@trigger.dev/core";
+import { UpdateTriggerSourceBodyV1Schema } from "@trigger.dev/core";
 import { z } from "zod";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { UpdateSourceService } from "~/services/sources/updateSource.server";
+import { UpdateSourceServiceV1 } from "~/services/sources/updateSourceV1.server";
 
 const ParamsSchema = z.object({
   endpointSlug: z.string(),
@@ -40,13 +40,13 @@ export async function action({ request, params }: ActionArgs) {
   // Now parse the request body
   const anyBody = await request.json();
 
-  const body = UpdateTriggerSourceBodySchema.safeParse(anyBody);
+  const body = UpdateTriggerSourceBodyV1Schema.safeParse(anyBody);
 
   if (!body.success) {
     return json({ error: "Invalid request body" }, { status: 400 });
   }
 
-  const service = new UpdateSourceService();
+  const service = new UpdateSourceServiceV1();
 
   try {
     const source = await service.call({

apps/webapp/app/routes/api.v1.$endpointSlug.triggers.$id.registrations.$key.ts

@@ -1,10 +1,10 @@
 import type { ActionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import { RegisterTriggerBodySchema } from "@trigger.dev/core";
+import { RegisterTriggerBodySchemaV1 } from "@trigger.dev/core";
 import { z } from "zod";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { RegisterTriggerSourceService } from "~/services/triggers/registerTriggerSource.server";
+import { RegisterTriggerSourceServiceV1 } from "~/services/triggers/registerTriggerSourceV1.server";
 
 const ParamsSchema = z.object({
   endpointSlug: z.string(),
@@ -41,13 +41,13 @@ export async function action({ request, params }: ActionArgs) {
   // Now parse the request body
   const anyBody = await request.json();
 
-  const body = RegisterTriggerBodySchema.safeParse(anyBody);
+  const body = RegisterTriggerBodySchemaV1.safeParse(anyBody);
 
   if (!body.success) {
     return json({ error: "Invalid request body" }, { status: 400 });
   }
 
-  const service = new RegisterTriggerSourceService();
+  const service = new RegisterTriggerSourceServiceV1();
 
   try {
     const registration = await service.call({

apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts

@@ -183,7 +183,7 @@ export class RunTaskService {
             },
           },
           parent: taskBody.parentId ? { connect: { id: taskBody.parentId } } : undefined,
-          name: taskBody.name,
+          name: taskBody.name ?? "Task",
           description: taskBody.description,
           status,
           startedAt: new Date(),

apps/webapp/app/routes/api.v2.$endpointSlug.sources.$id.ts

@@ -0,0 +1,72 @@
+import type { ActionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { UpdateTriggerSourceBodyV2Schema } from "@trigger.dev/core";
+import { z } from "zod";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { logger } from "~/services/logger.server";
+import { UpdateSourceServiceV2 } from "~/services/sources/updateSourceV2.server";
+
+const ParamsSchema = z.object({
+  endpointSlug: z.string(),
+  id: z.string(),
+});
+
+export async function action({ request, params }: ActionArgs) {
+  logger.info("Updating source", { url: request.url });
+
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "PUT") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  const parsedParams = ParamsSchema.safeParse(params);
+
+  if (!parsedParams.success) {
+    logger.info("Invalid params", { params });
+
+    return json({ error: "Invalid params" }, { status: 400 });
+  }
+
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    logger.info("Invalid or missing api key", { url: request.url });
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const authenticatedEnv = authenticationResult.environment;
+
+  // Now parse the request body
+  const anyBody = await request.json();
+
+  const body = UpdateTriggerSourceBodyV2Schema.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ error: "Invalid request body" }, { status: 400 });
+  }
+
+  const service = new UpdateSourceServiceV2();
+
+  try {
+    const source = await service.call({
+      environment: authenticatedEnv,
+      payload: body.data,
+      endpointSlug: parsedParams.data.endpointSlug,
+      id: parsedParams.data.id,
+    });
+
+    return json(source);
+  } catch (error) {
+    if (error instanceof Error) {
+      logger.error("Error activating http source", {
+        url: request.url,
+        error: error.message,
+      });
+
+      return json({ error: error.message }, { status: 400 });
+    }
+
+    return json({ error: "Something went wrong" }, { status: 500 });
+  }
+}