Establish webhooks journal example

.env.example

@@ -0,0 +1,9 @@
+# PostgreSQL connection string
+DATABASE_URL=postgresql://user:password@localhost:5432/webhooks_journal_example
+
+# HubSpot Public App credentials
+CLIENT_ID=your-client-id
+CLIENT_SECRET=your-client-secret
+
+# Redis connection (for BullMQ message queue)
+REDIS_URL=redis://localhost:6379

.github/workflows/run-test.yml

@@ -0,0 +1,56 @@
+name: Run Tests
+on:
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:15
+        env:
+          POSTGRES_USER: test
+          POSTGRES_PASSWORD: test
+          POSTGRES_DB: webhooks_journal_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+      redis:
+        image: redis:7
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    env:
+      DATABASE_URL: postgresql://test:test@localhost:5432/webhooks_journal_test
+      REDIS_URL: redis://localhost:6379
+      CLIENT_ID: test-client-id
+      CLIENT_SECRET: test-client-secret
+
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: '22'
+          cache: 'npm'
+
+      - run: npm install
+      - run: npx prisma db push
+      - run: npm test
+
+      - name: Build client
+        run: |
+          cd client
+          npm install
+          npm run build

.gitignore

@@ -6,5 +6,3 @@ coverage/
 .DS_Store
 client/node_modules/
 client/dist/
-package-lock.json
-client/package-lock.json

README.md

@@ -0,0 +1,207 @@
+# HubSpot Webhooks Journal Example
+
+A full-stack example integration demonstrating the [HubSpot Webhooks Journal API](https://developers.hubspot.com/docs). This app subscribes to CRM contact events, processes them through a durable message queue, and displays real-time contact data in a React UI.
+
+## What It Demonstrates
+
+- **Client Credentials OAuth** for authenticating with the Webhooks Journal API
+- **Journal polling** with offset-based cursor management
+- **BullMQ message queue** for durable, parallel event processing
+- **Last-write-wins reconciliation** using `occurredAt` timestamps
+- **Bulk snapshot backfill** for initial data population
+- **Advanced subscription filters** with AND/OR condition logic
+- **Live React UI** reflecting CRM contact state in real time
+
+## Architecture
+
+```
+┌──────────────────────┐     ┌─────────────────┐     ┌──────────────┐
+│   React Frontend     │────▶│  Express API    │────▶│  PostgreSQL  │
+│   (port 3000)        │     │  (port 3002)    │     │              │
+└──────────────────────┘     └────────┬────────┘     └──────────────┘
+                                      │
+                           ┌──────────┴──────────┐
+                           │                     │
+                    ┌──────▼──────┐      ┌───────▼───────┐
+                    │   Journal   │      │    Event      │
+                    │   Poller    │─────▶│   Processor   │
+                    │ (1 req/sec) │ Redis│ (5 workers)   │
+                    └─────────────┘ MQ   └───────────────┘
+```
+
+1. **Journal Poller** — polls the HubSpot Webhooks Journal batch endpoint once per second, enqueues entries into BullMQ
+2. **Event Processor** — 5 concurrent workers download journal files from S3, reconcile events with the local database
+3. **Express API** — serves the React frontend and exposes REST endpoints for contacts, subscriptions, snapshots
+4. **React Frontend** — dark-themed UI showing live contact data, bulk export controls, and test data population
+
+## Prerequisites
+
+- **Node.js** >= 18
+- **PostgreSQL** (running locally)
+- **Redis** (running locally, for BullMQ)
+- A **HubSpot Public App** with OAuth configured (see setup below)
+
+## Setup
+
+### 1. Create a HubSpot Public App
+
+1. Go to your [HubSpot Developer Account](https://developers.hubspot.com/)
+2. Create a new Public App
+3. Under **Auth**, add the redirect URL: `http://localhost:3002/oauth-callback`
+4. Under **Auth > Scopes**, add:
+   - `crm.objects.contacts.read`
+   - `crm.objects.contacts.write`
+5. Copy the **Client ID** and **Client Secret**
+
+### 2. Configure Environment
+
+```bash
+git clone https://github.com/hubspotdev/webhooks-journal-example.git
+cd webhooks-journal-example
+cp .env.example .env
+```
+
+Edit `.env` with your values:
+
+```
+DATABASE_URL=postgresql://user:password@localhost:5432/webhooks_journal_example
+CLIENT_ID=your-client-id
+CLIENT_SECRET=your-client-secret
+REDIS_URL=redis://localhost:6379
+```
+
+### 3. Install Dependencies
+
+```bash
+npm install
+cd client && npm install && cd ..
+```
+
+### 4. Initialize Database
+
+```bash
+createdb webhooks_journal_example  # or use your preferred method
+npm run db-init
+```
+
+### 5. Start the App
+
+In two terminal windows:
+
+```bash
+# Terminal 1: Backend
+npm run dev
+
+# Terminal 2: Frontend
+npm run client:dev
+```
+
+Open `http://localhost:3000` in your browser.
+
+## Usage
+
+1. **Enter Portal ID** — type the HubSpot portal ID you want to sync
+2. **Install App** — click to authorize the app via OAuth (opens HubSpot in a new tab)
+3. **View Live Data** — see contacts appear as the journal poller picks up events
+4. **Create Test Data** (optional) — creates 10 random contacts in HubSpot via the CRM API
+5. **Bulk Export** — triggers a bulk snapshot to backfill all contacts from the portal
+6. **Manage Filters** — open the filter dialog from the subscriptions banner to add property-based filters (up to 5 filters with up to 10 conditions each)
+7. **Clear Data** — purges the local database to start fresh
+
+## Tech Stack
+
+| Component | Technology |
+|-----------|-----------|
+| Backend | Express + TypeScript |
+| Database | PostgreSQL via Prisma |
+| Message Queue | BullMQ (Redis-backed) |
+| Frontend | React 18 + Material UI + Vite |
+| Auth | OAuth 2.0 (authorization code + client credentials) |
+| HubSpot SDK | `@hubspot/api-client` |
+
+## Project Structure
+
+```
+├── src/
+│   ├── app.ts              # Express server entry point
+│   ├── auth.ts             # OAuth + client credentials token management
+│   ├── clients.ts          # Prisma + HubSpot client singletons
+│   ├── swagger.ts          # API documentation config
+│   ├── journal/
+│   │   ├── poller.ts       # Journal polling worker (1 req/sec)
+│   │   ├── processor.ts    # Event processing (download + reconcile)
+│   │   └── queue.ts        # BullMQ queue + worker setup
+│   ├── hubspot/
+│   │   ├── contacts.ts     # Test data creation via CRM API
+│   │   ├── filters.ts      # Subscription filter CRUD
+│   │   ├── subscriptions.ts # Subscription management
+│   │   └── snapshots.ts    # Bulk snapshot API
+│   ├── routes/
+│   │   ├── authRoutes.ts   # OAuth install + callback
+│   │   ├── contactRoutes.ts # Contact CRUD
+│   │   └── journalRoutes.ts # Journal + snapshot controls
+│   └── utils/
+│       ├── error.ts        # Centralized error handler
+│       ├── logger.ts       # Structured logger
+│       ├── shutdown.ts     # Graceful shutdown
+│       └── utils.ts        # Constants and helpers
+├── client/
+│   ├── src/
+│   │   ├── App.tsx         # Root component with screen routing
+│   │   ├── main.tsx        # React entry point
+│   │   ├── theme.ts        # Dark theme configuration
+│   │   ├── components/
+│   │   │   ├── StartScreen.tsx  # Portal ID input + install
+│   │   │   ├── DataView.tsx     # Live contact list + bulk export
+│   │   │   └── FilterDialog.tsx # Subscription filter management UI
+│   │   └── utils/
+│   │       └── api.ts      # API client functions
+│   ├── package.json
+│   └── vite.config.ts      # Dev proxy to backend
+├── prisma/
+│   └── schema.prisma       # Database schema
+├── .env.example
+└── package.json
+```
+
+## How It Works
+
+### Authentication
+
+The app uses **two OAuth flows**:
+
+1. **Authorization Code** — installs the app on a portal and obtains a token for CRM API access (creating test contacts). This token is stored and auto-refreshed.
+2. **Client Credentials** — obtains an app-level token for Webhooks Journal API access (polling, subscriptions, snapshots). This is the token used for all journal operations.
+
+> **Note:** For simplicity, OAuth tokens are stored in plaintext in the database; a production application should encrypt tokens at rest.
+
+### Event Processing Pipeline
+
+1. The **Journal Poller** calls `GET /webhooks-journal/journal/2027-03-beta/batch/{offset}/next/100` once per second
+2. Each journal entry URL is enqueued as a BullMQ job with 3 retry attempts and exponential backoff
+3. **Event Processor** workers (5 concurrent) dequeue jobs, download the S3 file, and process each event:
+   - `CREATE` / `UPDATE` / `RESTORE` / `SNAPSHOT` → upsert contact in database
+   - `DELETE` → remove contact from database
+   - `MERGE` → delete merged contacts, upsert surviving contact
+4. **Last-write-wins**: if an incoming event's `occurredAt` is older than the contact's `lastModifiedDate`, it's skipped
+
+### Subscription Filters
+
+Filters narrow which events a subscription delivers to the journal. They are evaluated server-side before events are written — filtered-out events never appear in the journal.
+
+- A subscription can have **up to 5 filters** — an event passes if **any** filter matches (**OR** across filters)
+- Each filter can have **up to 10 conditions** — all conditions must match for the filter to pass (**AND** within a filter)
+- Filters evaluate against `propertyChanges` in the event. If a property referenced in a condition is not present in the event (e.g., a CREATE event where only `email` was set but the filter checks `firstname`), that condition is skipped and the event passes through
+- Supported operators: `EQ`, `N_EQ`, `LT`, `GT`, `LTE`, `GTE`, `IS_EMPTY`, `IS_NOT_EMPTY`
+- Filters are **not retroactive** — they only apply to future events
+
+### Recommended Sync Pattern: Bulk Snapshot + Incremental Journal
+
+When a new portal installs your app, combine a bulk snapshot with incremental journal polling to get a complete, up-to-date view of the portal's data without gaps:
+
+1. **Create a subscription** for the object types and actions you care about (`CREATE`, `UPDATE`, `DELETE`, `MERGE`). In this example, the app creates these subscriptions automatically after install.
+2. **Trigger a bulk snapshot** by clicking "Bulk Export" in the UI (or calling `POST /webhooks-journal/snapshots/2027-03-beta/crm-bulk`). This exports every contact in the portal as `SNAPSHOT` events that flow into the journal.
+3. **Poll the snapshot status** until it reaches `SUCCESS`. The UI handles this automatically, polling every second. A `404` response immediately after creation is expected — the status record is created asynchronously.
+4. **Continue polling incrementally.** The journal poller is already running, so it picks up the snapshot events as they appear and then seamlessly transitions to processing live `CREATE`/`UPDATE`/`DELETE` events as they arrive — no further bulk operations needed.
+
+This pattern ensures no events are missed: the subscription captures live changes from the moment it's created, and the bulk snapshot backfills everything that existed before. The `occurredAt`-based last-write-wins reconciliation handles any overlap between snapshot data and incremental events that occurred during the export.

__tests__/retry.test.ts

@@ -0,0 +1,99 @@
+import { requestWithRetry, parseRetryAfter, isRetryable } from '../src/utils/retry';
+
+interface FakeAxiosError {
+  isAxiosError: true;
+  message: string;
+  response?: { status: number; headers?: Record<string, string> };
+}
+
+function axiosError(status?: number, headers?: Record<string, string>): FakeAxiosError {
+  return {
+    isAxiosError: true,
+    message: 'boom',
+    response: status === undefined ? undefined : { status, headers },
+  };
+}
+
+const fastOptions = { baseDelayMs: 1, maxDelayMs: 2 };
+
+describe('parseRetryAfter', () => {
+  it('parses a delay expressed in seconds', () => {
+    expect(parseRetryAfter('2')).toBe(2000);
+  });
+
+  it('returns null when the header is missing', () => {
+    expect(parseRetryAfter(undefined)).toBeNull();
+  });
+
+  it('parses an HTTP date into a positive delay', () => {
+    const future = new Date(Date.now() + 5000).toUTCString();
+    const delay = parseRetryAfter(future);
+    expect(delay).not.toBeNull();
+    expect(delay!).toBeGreaterThan(3000);
+  });
+});
+
+describe('isRetryable', () => {
+  it('retries 429 and 5xx responses', () => {
+    expect(isRetryable(axiosError(429))).toBe(true);
+    expect(isRetryable(axiosError(503))).toBe(true);
+  });
+
+  it('retries network errors with no response', () => {
+    expect(isRetryable(axiosError(undefined))).toBe(true);
+  });
+
+  it('does not retry 4xx client errors', () => {
+    expect(isRetryable(axiosError(400))).toBe(false);
+    expect(isRetryable(axiosError(404))).toBe(false);
+  });
+});
+
+describe('requestWithRetry', () => {
+  it('retries transient failures then resolves', async () => {
+    let calls = 0;
+    const fn = jest.fn(async () => {
+      calls += 1;
+      if (calls < 3) throw axiosError(503);
+      return 'ok';
+    });
+
+    await expect(requestWithRetry(fn, { ...fastOptions, maxRetries: 5 })).resolves.toBe('ok');
+    expect(fn).toHaveBeenCalledTimes(3);
+  });
+
+  it('honors the Retry-After header when present', async () => {
+    let calls = 0;
+    const fn = jest.fn(async () => {
+      calls += 1;
+      if (calls < 2) throw axiosError(429, { 'retry-after': '0' });
+      return 'ok';
+    });
+
+    await expect(requestWithRetry(fn, { ...fastOptions, maxRetries: 3 })).resolves.toBe('ok');
+    expect(fn).toHaveBeenCalledTimes(2);
+  });
+
+  it('gives up after maxRetries and rethrows', async () => {
+    const fn = jest.fn(async () => {
+      throw axiosError(503);
+    });
+
+    await expect(requestWithRetry(fn, { ...fastOptions, maxRetries: 2 })).rejects.toMatchObject({
+      response: { status: 503 },
+    });
+    // initial attempt + 2 retries
+    expect(fn).toHaveBeenCalledTimes(3);
+  });
+
+  it('does not retry non-retryable errors', async () => {
+    const fn = jest.fn(async () => {
+      throw axiosError(400);
+    });
+
+    await expect(requestWithRetry(fn, fastOptions)).rejects.toMatchObject({
+      response: { status: 400 },
+    });
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+});