feat(webhooks): add webhook workflows

.eslintrc.json

@@ -16,6 +16,7 @@
   },
   "plugins": ["@typescript-eslint", "import", "unused-imports", "prettier", "eslint-plugin-jest"],
   "rules": {
+    "no-void": "off",
     "@typescript-eslint/no-shadow": "off",
     "@typescript-eslint/no-throw-literal": "off",
     "no-use-before-define": "off",

.github/workflows/tests.yml

@@ -34,7 +34,7 @@ jobs:
 
   e2e:
     runs-on: ubuntu-latest
-    timeout-minutes: 30
+    timeout-minutes: 10
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -83,7 +83,7 @@ jobs:
         run: |
           export DATABASE_URL="postgresql://hatchet:hatchet@127.0.0.1:5431/hatchet"
           go run github.com/steebchen/prisma-client-go migrate deploy
-          task generate-all
+          go run github.com/steebchen/prisma-client-go generate
           task generate-certs
           task generate-local-encryption-keys
 
@@ -112,6 +112,8 @@ jobs:
           SERVER_AUTH_SET_EMAIL_VERIFIED=true
           EOF
 
+          export SEED_DEVELOPMENT=true
+
           go run ./cmd/hatchet-admin quickstart
 
           go run ./cmd/hatchet-engine --config ./generated/ &

examples/webhooks.e2e.ts

@@ -0,0 +1,96 @@
+import { createServer } from 'node:http';
+import { Workflow, Worker } from '../src';
+import sleep from '../src/util/sleep';
+import Hatchet from '../src/sdk';
+
+const port = 8369;
+
+describe('webhooks', () => {
+  let hatchet: Hatchet;
+  let worker: Worker;
+
+  beforeEach(async () => {
+    hatchet = Hatchet.init();
+    worker = await hatchet.worker('simple-webhook-workflow');
+  });
+
+  afterEach(async () => {
+    await worker.stop();
+    await sleep(2000);
+  });
+
+  it('should pass a webhook workflow', async () => {
+    let invoked = 0;
+
+    const workflow: Workflow = {
+      id: 'simple-webhook-workflow',
+      description: 'test',
+      on: {
+        event: 'user:create-webhook',
+      },
+      steps: [
+        {
+          name: 'step1',
+          run: async (ctx) => {
+            invoked += 1;
+            return { message: `${ctx.workflowName()} results!` };
+          },
+        },
+        {
+          name: 'step2',
+          parents: ['step1'],
+          run: (ctx) => {
+            invoked += 1;
+            return { message: `${ctx.workflowName()} results!` };
+          },
+        },
+      ],
+    };
+
+    console.log('registering workflow...');
+    await worker.registerWorkflow(workflow);
+
+    const secret = 'secret';
+
+    console.log('registering webhook...');
+    await worker.registerWebhook({
+      secret,
+      url: `http://localhost:${port}/webhook`,
+      workflows: ['simple-webhook-workflow'],
+    });
+
+    console.log('starting worker...');
+
+    const handler = hatchet.webhooks(workflow);
+
+    const server = createServer(handler.httpHandler({ secret }));
+
+    await new Promise((resolve) => {
+      server.listen(port, () => {
+        resolve('');
+      });
+    });
+
+    console.log('server started.');
+    console.log('waiting for webhook to be registered...');
+
+    // wait for engine to pick up the webhook worker
+    await sleep(30000 + 3000);
+
+    console.log('webhook wait time complete.');
+
+    console.log('pushing event...');
+
+    await hatchet.event.push('user:create-webhook', {
+      test: 'test',
+    });
+
+    await sleep(10000);
+
+    console.log('invoked', invoked);
+
+    expect(invoked).toEqual(2);
+
+    await worker.stop();
+  }, 60000);
+});

package.json

@@ -19,7 +19,7 @@
     "prepare": "npm run build",
     "tsc:build": "tsc && resolve-tspaths",
     "test:unit": "jest --testMatch='**/*.test.ts'",
-    "test:e2e": "jest --testMatch='**/*.e2e.ts'",
+    "test:e2e": "jest --testMatch='**/*.e2e.ts' --detectOpenHandles",
     "test:unit:watch": "jest --testMatch='**/*.test.ts' --watch",
     "generate": "pnpm run '/generate-.*/'",
     "generate-api": "npx --yes swagger-cli bundle ./hatchet/api-contracts/openapi/openapi.yaml --outfile openapi.yaml --type yaml && npx swagger-typescript-api -p openapi.yaml -o src/clients/rest/generated -n hatchet.ts --modular --axios",
@@ -100,4 +100,4 @@
     "yaml": "^2.3.4",
     "zod": "^3.22.4"
   }
-}
+}

src/clients/admin/admin-client.ts

@@ -11,7 +11,7 @@ import { Logger } from '@hatchet/util/logger';
 import { retrier } from '@hatchet/util/retrier';
 
 import { Api } from '../rest';
-import { WorkflowRunStatus } from '../rest/generated/data-contracts';
+import { WebhookWorkerCreateRequest, WorkflowRunStatus } from '../rest/generated/data-contracts';
 
 type WorkflowMetricsQuery = {
   workflowId?: string;
@@ -90,6 +90,10 @@ export class AdminClient {
     }
   }
 
+  async webhook_create(data: WebhookWorkerCreateRequest) {
+    return this.api.webhookCreate(this.tenantId, data);
+  }
+
   /**
    * Run a new instance of a workflow with the given input. This will create a new workflow run and return the ID of the
    * new run.

src/clients/dispatcher/action-listener.ts

@@ -1,11 +1,16 @@
-import { DispatcherClient as PbDispatcherClient, AssignedAction } from '@hatchet/protoc/dispatcher';
+import {
+  DispatcherClient as PbDispatcherClient,
+  AssignedAction,
+  actionTypeFromJSON,
+} from '@hatchet/protoc/dispatcher';
 
 import { Status } from 'nice-grpc';
 import { ClientConfig } from '@clients/hatchet-client/client-config';
 import sleep from '@util/sleep';
 import HatchetError from '@util/errors/hatchet-error';
 import { Logger } from '@hatchet/util/logger';
 
+import { z } from 'zod';
 import { DispatcherClient } from './dispatcher-client';
 import { Heartbeat } from './heartbeat/heartbeat-controller';
 
@@ -18,20 +23,22 @@ enum ListenStrategy {
   LISTEN_STRATEGY_V2 = 2,
 }
 
-export interface Action {
-  tenantId: string;
-  jobId: string;
-  jobName: string;
-  jobRunId: string;
-  stepId: string;
-  stepRunId: string;
-  actionId: string;
-  actionType: number;
-  actionPayload: string;
-  workflowRunId: string;
-  getGroupKeyRunId: string;
-  stepName: string;
-}
+export const ActionObject = z.object({
+  tenantId: z.string(),
+  jobId: z.string(),
+  jobName: z.string(),
+  jobRunId: z.string(),
+  stepId: z.string(),
+  stepRunId: z.string(),
+  actionId: z.string(),
+  actionType: z.preprocess((s) => actionTypeFromJSON(s), z.number().optional()),
+  actionPayload: z.string(),
+  workflowRunId: z.string(),
+  getGroupKeyRunId: z.string().optional(),
+  stepName: z.string(),
+});
+
+export type Action = z.infer<typeof ActionObject>;
 
 export class ActionListener {
   config: ClientConfig;

src/clients/hatchet-client/hatchet-client.ts

@@ -162,4 +162,12 @@ export class HatchetClient {
 
     return worker;
   }
+
+  webhooks(workflow: Workflow) {
+    const worker = new Worker(this, {
+      name: workflow.id,
+    });
+
+    return worker.getHandler(workflow);
+  }
 }

src/clients/rest/generated/Api.ts

@@ -68,6 +68,9 @@ import {
   UserLoginRequest,
   UserRegisterRequest,
   UserTenantMembershipsList,
+  WebhookWorker,
+  WebhookWorkerCreateRequest,
+  WebhookWorkerListResponse,
   Worker,
   WorkerList,
   Workflow,
@@ -1636,4 +1639,36 @@ export class Api<SecurityDataType = unknown> extends HttpClient<SecurityDataType
       format: 'json',
       ...params,
     });
+  /**
+   * @description Lists all webhooks
+   *
+   * @name WebhookList
+   * @summary List webhooks
+   * @request GET:/api/v1/webhook-workers/{tenant}
+   * @secure
+   */
+  webhookList = (tenant: string, params: RequestParams = {}) =>
+    this.request<WebhookWorkerListResponse, APIErrors>({
+      path: `/api/v1/webhook-workers/${tenant}`,
+      method: 'GET',
+      secure: true,
+      format: 'json',
+      ...params,
+    });
+  /**
+   * @description Creates a webhook
+   *
+   * @name WebhookCreate
+   * @summary Create a webhook
+   * @request POST:/api/v1/webhook-workers/{tenant}/create
+   */
+  webhookCreate = (tenant: string, data: WebhookWorkerCreateRequest, params: RequestParams = {}) =>
+    this.request<WebhookWorker, APIErrors>({
+      path: `/api/v1/webhook-workers/${tenant}/create`,
+      method: 'POST',
+      body: data,
+      type: ContentType.Json,
+      format: 'json',
+      ...params,
+    });
 }

src/clients/rest/generated/data-contracts.ts

@@ -1014,3 +1014,32 @@ export interface WorkflowMetrics {
   /** The total number of concurrency group keys. */
   groupKeyCount?: number;
 }
+
+export interface WebhookWorker {
+  metadata: APIResourceMeta;
+  /** The webhook url. */
+  url: string;
+  /** The secret key for validation. */
+  secret: string;
+}
+
+export interface WebhookWorkerCreateRequest {
+  /** The webhook url. */
+  url: string;
+  /** The workflow IDs or names to register for this webhook worker. */
+  workflows: string[];
+  /**
+   * The secret key for validation. If not provided, a random secret will be generated.
+   * @minLength 32
+   */
+  secret?: string;
+}
+
+export interface WebhookWorkerCreateResponse {
+  worker?: WebhookWorker;
+}
+
+export interface WebhookWorkerListResponse {
+  pagination?: PaginationResponse;
+  rows?: WebhookWorker[];
+}