Merge pull request #1 from kustom-dev/kustom/implement-events-processor

Implement example Cloudflare worker for fetching and processing WorkOS events

.github/workflows/test.yaml

@@ -0,0 +1,50 @@
+name: test
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    types: [ opened, synchronize, reopened ]
+
+jobs:
+  run-integration-tests:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - uses: pnpm/action-setup@v4
+        name: Install pnpm
+        with:
+          run_install: false
+
+      - name: Install Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: 20
+          cache: "pnpm"
+
+      - name: Get pnpm store directory
+        shell: bash
+        run: |
+          echo "STORE_PATH=$(pnpm store path --silent)" >> $GITHUB_ENV
+
+      - uses: actions/cache@v4
+        name: Setup pnpm cache
+        with:
+          path: ${{ env.STORE_PATH }}
+          key: ${{ runner.os }}-pnpm-store-${{ hashFiles('**/pnpm-lock.yaml') }}
+          restore-keys: |
+            ${{ runner.os }}-pnpm-store-
+
+      - name: Install dependencies
+        run: pnpm install
+
+      - name: Run integration tests
+        run: pnpm run test
+        env:
+          SKIP_TEARDOWN_MIGRATE: false
+          WORKOS_API_KEY: ${{ secrets.WORKOS_API_KEY }}
+          WORKOS_CLIENT_ID: ${{ secrets.WORKOS_CLIENT_ID }}

.gitignore

@@ -53,9 +53,6 @@ build/Release
 node_modules/
 jspm_packages/
 
-# pnpm
-pnpm-lock.yaml
-
 # Snowpack dependency directory (https://snowpack.dev/)
 
 web_modules/

README.md

@@ -0,0 +1,67 @@
+# @kustom/workos-events-consumer
+
+## What does this Cloudflare Worker do?
+[WorkOS](https://workos.com/) is an auth service provider with a variety of products. It has an [Events API](https://workos.com/docs/events/data-syncing/events-api) which allows customers to sync WorkOS data within their application. WorkOS also provides webhooks but the Events API is the recommended approach to syncing data.
+
+This repository contains a Cloudflare worker which runs on a schedule (every 60 seconds) and persists WorkOS data to a [Cloudflare D1 database](https://developers.cloudflare.com/d1). You can still use this code with another SQL database by using [Hyperdrive](https://developers.cloudflare.com/hyperdrive/). As of now, this repository does not contain a Hyperdrive example.
+
+## Getting Started
+- Have `npm`, `yarn` or `pnpm` installed. Commands in this repo use `pnpm`
+- Install dependencies via `pnpm i`
+- Initialize the dev DB with `pnpm run dev:init-db`
+- Start jaeger containers for viewing telemetry with `docker compose up -d`
+- Go to http://localhost:16686 for the dev environment and http://localhost:16687 for the test environment to view traces
+- Copy `sample.dev.vars` to `.dev.vars`
+- Add your WorkOS credentials to the `.dev.vars` file
+- Run with `pnpm run dev`
+- In another terminal tab run `http://localhost:8787/__scheduled`
+- The default behavior for the development environment is to request events in the last 60 seconds from the WorkOS Events API
+- You can modify this behavior by adding `EVENTS_RANGE_START` and/or `EVENTS_RANGE_END` to `.dev.vars` to fetch events within a given time range. Both env vars need to be a Unix Timestamp in Milliseconds
+- See the [Testing](#testing) section for how to generate and test events within a specific time range
+
+### Known limitations
+- This example worker does not take in event replay into account, processing the same event again will result in an error because the `event_id` and payload will already be in the database. This can be viewed as a feature or a bug depending on how you look at it
+- Running newer events is considered safe but please do your own testing
+
+## Local Database & Migrations
+- Create a new migration
+```
+pnpm run migrations:create kustom_app_dev create_user_org_cursor_tables --env dev
+```
+
+- Apply migrations
+```
+pnpm run dev:init-db
+```
+
+- Reset development DB
+```
+pnpm run dev:destroy-db
+```
+
+## Testing
+The current recommended approach for testing Cloudflare Workers is the [Vitest Integration](https://developers.cloudflare.com/workers/testing/vitest-integration/). It is in beta and I have not had a great experience with it, so far. The Vitest Integration is pinned to vitest version `1.5.0` at the time of writing (Sep 2024).
+
+The tests in this repo DO NOT use the Vitest Integration. They run on the latest version of Vitest and Miniflare. The setup probably seems tedious but since I've invested the time and effort to get it working, I'm rolling with it until the Vitest Integration is stable.
+
+I didn't want to create new events every time the test suite runs. To get around that, I create a set of events that are appropriate for my use case, then set the `EVENTS_RANGE_START` and `EVENTS_RANGE_END` environment variables for the `test` environment. This ensures that I'm always processing the same events but still calling out to the real WorkOS API.
+
+Over time, when new events are added, they can be re-processed and the time interval for test events can be updated.
+
+- To run the test suite, copy `sample.envrc` to `.envrc` if you use [direnv](https://direnv.net/) or ensure those env vars are loaded
+- The test suite uses Vitest and Node's `process.env` to read environment variables
+- Uncomment the test at the top of `cursor.test.ts` and run the suite with `pnpm run test`
+- The tests will fail since the IDs for each entity will be different compared to the ones I've used
+- You can replace the IDs or change the `emitTestingEvents()` function to match your use case
+- Once you've matched the IDs, the tests should pass
+- If you want to examine your test D1 database, it's in `.wrangler/state/test/v3/d1/miniflare-D1DatabaseObject`
+
+### References
+
+The implementation in this repo was informed by [this blog post](https://workos.com/blog/why-you-should-rethink-your-webhook-strategy) from WorkOS.
+
+Thanks for providing an alternative to the webhook strategy and writing a detailed post about it.
+
+### Questions/Comments
+
+If you have questions or would like to chat about this implementation, feel free to reach out via email: `[email protected]`

compose.yaml

@@ -0,0 +1,21 @@
+services:
+  jaeger-dev:
+    image: jaegertracing/all-in-one:latest
+    ports:
+      - "16686:16686"
+      - "4318:4318"
+    environment:
+      - COLLECTOR_OTLP_HTTP_CORS_ALLOWED_HEADERS=*
+      - COLLECTOR_OTLP_HTTP_CORS_ALLOWED_ORIGINS=*
+      - COLLECTOR_OTLP_ENABLED=true
+      - LOG_LEVEL=debug
+  jaeger-test:
+    image: jaegertracing/all-in-one:latest
+    ports:
+      - "16687:16686"
+      - "4319:4318"
+    environment:
+      - COLLECTOR_OTLP_HTTP_CORS_ALLOWED_HEADERS=*
+      - COLLECTOR_OTLP_HTTP_CORS_ALLOWED_ORIGINS=*
+      - COLLECTOR_OTLP_ENABLED=true
+      - LOG_LEVEL=debug

migrations/0001_create_user_org_cursor_tables.sql

@@ -0,0 +1,52 @@
+-- Migration number: 0001 	 2024-09-06T22:40:10.825Z
+-- WorkOS Events cursor
+CREATE TABLE
+  IF NOT EXISTS events_cursor (
+    event_id TEXT PRIMARY KEY NOT NULL,
+    event_payload TEXT NOT NULL,
+    created_at INTEGER NOT NULL,
+    processed_at INTEGER NOT NULL
+  );
+
+CREATE INDEX IF NOT EXISTS events_cursor_created_at_idx ON events_cursor (created_at);
+
+-- WorkOS User
+CREATE TABLE
+  IF NOT EXISTS user (
+    id TEXT PRIMARY KEY NOT NULL,
+    email VARCHAR(254) NOT NULL,
+    first_name TEXT,
+    last_name TEXT,
+    email_verified BOOLEAN NOT NULL,
+    profile_picture_url TEXT,
+    created_at INTEGER NOT NULL,
+    updated_at INTEGER NOT NULL
+  );
+
+CREATE INDEX IF NOT EXISTS user_email_idx ON user (email);
+
+-- WorkOS Organization
+CREATE TABLE
+  IF NOT EXISTS organization (
+    id TEXT PRIMARY KEY NOT NULL,
+    name TEXT NOT NULL,
+    created_at INTEGER NOT NULL,
+    updated_at INTEGER NOT NULL
+  );
+
+CREATE INDEX IF NOT EXISTS organization_name_idx ON organization (name);
+
+-- WorkOS Organization Membership
+CREATE TABLE
+  IF NOT EXISTS organization_membership (
+    id TEXT NOT NULL,
+    user_id TEXT NOT NULL,
+    organization_id TEXT NOT NULL,
+    role_slug TEXT NOT NULL,
+    status TEXT NOT NULL,
+    created_at INTEGER NOT NULL,
+    updated_at INTEGER NOT NULL,
+    FOREIGN KEY (user_id) REFERENCES user (id),
+    FOREIGN KEY (organization_id) REFERENCES organization (id),
+    PRIMARY KEY (id, user_id, organization_id)
+  );

package.json

@@ -1,39 +1,53 @@
 {
-  "name": "workos-events-consumer",
-  "version": "0.0.0",
+  "name": "@kustom/workos-events-consumer",
+  "version": "0.0.1",
   "private": true,
   "type": "module",
   "scripts": {
     "deploy": "wrangler deploy",
-    "dev": "wrangler dev",
-    "start": "wrangler dev",
+    "dev": "wrangler dev --env dev --test-scheduled",
+    "dev:destroy-db": "rm -rf .wrangler/state/v3/d1",
+    "dev:init-db": "pnpm run migrations:apply kustom_app_dev --local --env dev",
+    "migrations:create": "wrangler d1 migrations create",
+    "migrations:apply": "wrangler d1 migrations apply",
+    "start": "wrangler dev --env dev --test-scheduled",
     "test": "vitest run --config vitest.config.mts",
+    "test:init-db": "echo y | pnpm run migrations:apply kustom_app_test --local --env test --persist-to .wrangler/state/test",
     "cf-typegen": "wrangler types"
   },
   "devDependencies": {
     "@cloudflare/workers-types": "^4.20240903.0",
     "@eslint/js": "^9.10.0",
-    "@opentelemetry/api": "^1.9.0",
-    "@opentelemetry/context-zone": "^1.26.0",
-    "@opentelemetry/core": "^1.26.0",
-    "@opentelemetry/exporter-trace-otlp-http": "^0.53.0",
-    "@opentelemetry/instrumentation": "^0.53.0",
-    "@opentelemetry/otlp-exporter-base": "^0.53.0",
-    "@opentelemetry/otlp-transformer": "^0.53.0",
-    "@opentelemetry/resources": "^1.26.0",
-    "@opentelemetry/sdk-metrics": "^1.26.0",
-    "@opentelemetry/sdk-trace-base": "^1.26.0",
-    "@opentelemetry/sdk-trace-web": "^1.26.0",
+    "@types/node": "^22.5.4",
     "eslint": "^9.10.0",
     "eslint-config-prettier": "^9.1.0",
     "eslint-plugin-import": "^2.30.0",
     "eslint-plugin-prettier": "^5.2.1",
     "eslint-plugin-simple-import-sort": "^12.1.1",
     "globals": "^15.9.0",
+    "miniflare": "^3.20240821.1",
     "prettier": "^3.3.3",
     "typescript": "^5.5.4",
     "typescript-eslint": "^8.4.0",
     "vitest": "^2.0.5",
     "wrangler": "^3.75.0"
-  }
-}
+  },
+  "dependencies": {
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/context-zone": "^1.26.0",
+    "@opentelemetry/core": "^1.26.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.53.0",
+    "@opentelemetry/instrumentation": "^0.53.0",
+    "@opentelemetry/otlp-exporter-base": "^0.53.0",
+    "@opentelemetry/otlp-transformer": "^0.53.0",
+    "@opentelemetry/resources": "^1.26.0",
+    "@opentelemetry/sdk-metrics": "^1.26.0",
+    "@opentelemetry/sdk-trace-base": "^1.26.0",
+    "@opentelemetry/sdk-trace-web": "^1.26.0",
+    "@workos-inc/node": "^7.26.0"
+  },
+  "engines": {
+    "node": ">=20.0.0"
+  },
+  "packageManager": "[email protected]+sha512.60c18acd138bff695d339be6ad13f7e936eea6745660d4cc4a776d5247c540d0edee1a563695c183a66eb917ef88f2b4feb1fc25f32a7adcadc7aaf3438e99c1"
+}