New Integration whole steps

[Zack-Rider]

How to Add a New Integration

Step by step. Reuse everything already built.

---

The Most Important Thing to Understand First

Gmail and Slack are already built. They share a pipeline:

Your new integration
        ↓
   [Adapter]          ← YOU BUILD THIS (normalize raw data → our format)
        ↓
   [Router]           ← YOU BUILD THIS (OAuth callback + webhook/poll endpoint)
        ↓
──────────────────── everything below is ALREADY BUILT, don't touch ────────────────────
        ↓
IntegrationMessageService   integrations/message-ingestion.ts
        ↓
MessageRouter               services/message-router.ts
        ↓
BullMQ Queue                services/queues/message-queue.ts
        ↓
Worker                      services/workers/message-worker.ts
        ↓
TriggerResolver             services/trigger-resolver.ts
        ↓
WorkflowEngine

You only ever write 2 files per integration. Everything else is shared.

---

The 5 Steps (Same for Every Integration)

---

Step 1 — Add the type to the database enum

Open server/prisma/schema.prisma and add your new type to IntegrationType:

enum IntegrationType {
  GMAIL
  SLACK
  WHATSAPP
  GOOGLE_DRIVE    // ← add
  NOTION          // ← add
  HUBSPOT         // ← add
  DISCORD         // ← add
  SAP             // ← add
  MS_DYNAMICS     // ← add
  ORACLE          // ← add
  MS_EXCEL        // ← add
  CSV             // ← add
  DATABASE        // ← add
  TABLEAU         // ← add
}

Then run:

cd server && bunx prisma migrate dev --name add-new-integration-types

Why: The IntegrationType enum is used in the Integration table (which org connected what) and in the IntegrationMessage table (where each message came from). It's also used by TriggerResolver to match messages to workflow trigger nodes.

Who does this: Zapier, Airbyte, Fivetran — they all have a central registry of provider types. Adding to an enum is the correct approach, not a new table per provider.

---

Step 2 — Add the trigger node mapping

Open services/trigger-resolver.ts at line 88 and add your new type:

const triggerNodeTypes: Record<IntegrationType, string[]> = {
  GMAIL:        ["GmailTrigger", "GMAIL_TRIGGER", "Gmail Trigger"],
  SLACK:        ["SlackTrigger", "SLACK_TRIGGER"],
  WHATSAPP:     ["WhatsAppTrigger", "WHATSAPP_TRIGGER"],
  GOOGLE_DRIVE: ["GoogleDriveTrigger", "GOOGLE_DRIVE_TRIGGER"],   // ← add
  NOTION:       ["NotionTrigger", "NOTION_TRIGGER"],               // ← add
  HUBSPOT:      ["HubSpotTrigger", "HUBSPOT_TRIGGER"],             // ← add
  // ... etc
};

Why: When a message arrives from your integration, the worker looks for a workflow that has a matching trigger node. If you skip this step, messages arrive but no workflow ever fires.

---

Step 3 — Create the Adapter

This is the file that takes the raw data from the third-party platform and converts it to NormalizedMessage.

Look at how Gmail does it (integrations/gmail/gmail.ts:132) and copy the pattern.

The interface you must implement — integrations/base/types.ts:20:

interface IntegrationAdapter {
  type: IntegrationType;                   // which provider this handles
  verifyRequest?(args): void;             // optional — only if they send a signature header
  normalize(payload, ctx): Promise<NormalizedMessage[]>;  // REQUIRED — translate their data
}

The shape you produce — integrations/base/types.ts:7:

interface NormalizedMessage {
  sourceId: string;          // unique ID from the source (email id, slack ts, deal id...)
  conversationId?: string;   // thread/channel/ticket ID — groups related messages
  sender: string;            // who sent it (email address, slack userId, contact name...)
  recipient?: string;        // who received it
  content: string;           // the actual text content you want workflows to see
  sourceDate?: Date;         // when it happened in the source system
  metadata?: Record<string, unknown>;  // anything extra (subject, channel, labels...)
  rawData?: unknown;         // the original response — always keep this
}

Rule: Always put everything in rawData. If your normalize() has a bug, you can re-process without re-fetching.

---

Step 4 — Create the Router

This file has two things:

1. The OAuth callback route (where the user gets redirected back after connecting)
2. The data ingestion route (where the third-party pushes data, or where you poll)

Copy the pattern from integrations/slack/slack-router.ts (for OAuth) or integrations/gmail/gmail-router.ts (for OAuth + webhook).

The critical lines you must always write in the ingestion handler:

// 1. Find the integration (which org does this message belong to?)
const integration = await prisma.integration.findFirst({
  where: { type: "YOUR_TYPE", externalAccountId: someId, isActive: true },
  select: { id: true, orgId: true, type: true, config: true },
});

// 2. Normalize
const adapter = IntegrationRegistry.get(integration.type);
const messages = await adapter.normalize(rawPayload, { orgId: integration.orgId });
if (!messages.length) return c.json({ ok: true });

// 3. Ingest (dedup + save)
const messageService = new IntegrationMessageService(prisma);
const result = await messageService.ingest({
  integrationId: integration.id,
  orgId: integration.orgId,
  sourceType: integration.type,
  messages,
});

// 4. Queue (hand off to worker)
const router = new MessageRouter(prisma);
const insertedIds: string[] = (result as any).insertedSourceIds ?? [];
for (const sourceId of insertedIds) {
  const stored = await prisma.integrationMessage.findFirst({
    where: { integrationId: integration.id, sourceId },
    select: { id: true },
  });
  if (stored) {
    await router.route({ messageId: stored.id, orgId: integration.orgId,
      integrationId: integration.id, sourceType: integration.type,
      content: messages.find(m => m.sourceId === sourceId)?.content ?? "" }, c);
  }
}

Steps 3 and 4 are identical for every single integration. Copy-paste them.

---

Step 5 — Register the Adapter

Open integrations/index.ts and add one line:

import { IntegrationRegistry } from "./base/registry";
import { SlackAdapter }        from "./slack/slack";
import { GmailAdapter }        from "./gmail/gmail";
import { NotionAdapter }       from "./notion/notion";      // ← add
import { HubSpotAdapter }      from "./hubspot/hubspot";    // ← add

IntegrationRegistry.register(new SlackAdapter());
IntegrationRegistry.register(new GmailAdapter());
IntegrationRegistry.register(new NotionAdapter());      // ← add
IntegrationRegistry.register(new HubSpotAdapter());     // ← add

That's it. The worker, queue, dedup, and trigger resolution are all automatic.

---

Integration-by-Integration Guide

There are 4 types of integrations in your list. Each type needs a slightly different router. The adapter is always the same pattern.

---

Type A — OAuth + Webhook (data is pushed to you in real time)

Use for: Slack ✅, Google Drive, HubSpot, Notion, Discord

The third-party calls your endpoint when something changes. Fast, no polling needed.

Example: HubSpot

Step 1: User clicks "Connect HubSpot"
Step 2: Redirect to HubSpot OAuth → they approve
Step 3: HubSpot calls GET /api/webhooks/hubspot?code=...
Step 4: You exchange code → get access_token
Step 5: You register a webhook subscription on HubSpot
Step 6: HubSpot calls POST /api/webhooks/hubspot/events on new contacts/deals

Router file: integrations/hubspot/hubspot-router.ts

// OAuth callback
hubspotRouter.get("/api/webhooks/hubspot", async (c) => {
  const code = url.searchParams.get("code");
  // 1. exchange code for token
  // 2. register webhook: POST https://api.hubspot.com/webhooks/v3/{appId}/subscriptions
  // 3. prisma.integration.upsert({ type: "HUBSPOT", ... })
  // 4. redirect back
});

// Webhook receiver
hubspotRouter.post("/api/webhooks/hubspot/events", async (c) => {
  // run the 4-step pattern from Step 4 above
});

Adapter file: integrations/hubspot/hubspot.ts

export class HubSpotAdapter implements IntegrationAdapter {
  type: IntegrationType = "HUBSPOT";

  async normalize(payload: unknown, ctx: IntegrationContext): Promise<NormalizedMessage[]> {
    const events = Array.isArray(payload) ? payload : [payload] as any[];
    return events.map(event => ({
      sourceId: String(event.objectId),
      sender: String(event.portalId),
      content: `${event.subscriptionType}: ${JSON.stringify(event.propertyValue ?? "")}`,
      sourceDate: new Date(event.occurredAt),
      metadata: { objectType: event.objectType, subscriptionType: event.subscriptionType },
      rawData: event,
    }));
  }
}

Companies that use this: Zapier (receives webhooks from 8,500+ apps), HubSpot itself uses webhook subscriptions for their own marketplace apps.

Reference: https://developers.hubspot.com/docs/api/webhooks

---

Type B — OAuth + Polling (you ask them for new data on a schedule)

Use for: Google Drive (file changes), Notion (page updates), MS Dynamics

Some APIs don't support webhooks or their webhooks are unreliable. You poll every N minutes.

Example: Google Drive

Step 1: User connects Google Drive via OAuth (same as Gmail)
Step 2: You save the token to Integration table
Step 3: A cron job (or BullMQ repeatable job) runs every 15 minutes
Step 4: You call GET /drive/v3/changes?pageToken=... to fetch what changed
Step 5: Normalize → ingest → queue (same as always)

Router file: integrations/google-drive/google-drive-router.ts

// OAuth callback — reuse almost exactly the same code as gmail-router.ts:24
googleDriveRouter.get("/api/webhooks/google-drive", async (c) => {
  // exchange code, get token, upsert Integration with type: "GOOGLE_DRIVE"
  // save initial pageToken from: GET https://www.googleapis.com/drive/v3/changes/startPageToken
});

// Poll endpoint — called by your scheduler, not by Google
googleDriveRouter.post("/api/internal/google-drive/poll", async (c) => {
  const integrations = await prisma.integration.findMany({
    where: { type: "GOOGLE_DRIVE", isActive: true }
  });
  for (const integration of integrations) {
    await pollGoogleDriveChanges(integration, prisma, c);
  }
});

Adapter file: integrations/google-drive/google-drive.ts

export class GoogleDriveAdapter implements IntegrationAdapter {
  type: IntegrationType = "GOOGLE_DRIVE";

  async normalize(payload: unknown, ctx: IntegrationContext): Promise<NormalizedMessage[]> {
    const changes = payload as GoogleDriveChange[];
    return changes.map(change => ({
      sourceId: change.fileId,
      sender: change.file?.lastModifyingUser?.emailAddress ?? "unknown",
      content: `File changed: ${change.file?.name ?? change.fileId}`,
      sourceDate: new Date(change.time),
      metadata: { fileName: change.file?.name, mimeType: change.file?.mimeType,
                  webViewLink: change.file?.webViewLink },
      rawData: change,
    }));
  }
}

Companies that use this: Fivetran polls every connector on a schedule (5min to 24hr). Airbyte uses cursor-based incremental sync with updated_at fields.

Reference: https://developers.google.com/drive/api/guides/push

---

Type C — API Key / Basic Auth (enterprise systems, no OAuth)

Use for: SAP, MS Dynamics 365, Oracle, Tableau

These don't use OAuth. The org gives you an API key or username/password and you call their API directly. You poll on a schedule — they never push to you.

Step 1: User opens Settings → pastes API key / server URL
Step 2: You validate the credentials (test API call)
Step 3: You save to Integration table (config: { apiKey, serverUrl })
Step 4: Scheduler polls every hour
Step 5: Normalize → ingest → queue

Router file: integrations/sap/sap-router.ts

// No OAuth — just a settings save endpoint
sapRouter.post("/api/integrations/sap/connect", async (c) => {
  const { apiKey, serverUrl, orgId } = await c.req.json();

  // Validate credentials
  const test = await fetch(`${serverUrl}/sap/opu/odata/sap/...`, {
    headers: { Authorization: `Basic ${Buffer.from(apiKey).toString("base64")}` }
  });
  if (!test.ok) return c.json({ ok: false, error: "Invalid credentials" }, 400);

  await prisma.integration.upsert({
    where: { orgId_type: { orgId, type: "SAP" } },
    create: { orgId, type: "SAP", isActive: true,
              config: { apiKey: encrypt(apiKey), serverUrl } },
    update: { config: { apiKey: encrypt(apiKey), serverUrl }, isActive: true },
    select: { id: true },
  });

  return c.json({ ok: true });
});

Adapter file: integrations/sap/sap.ts

export class SAPAdapter implements IntegrationAdapter {
  type: IntegrationType = "SAP";

  async normalize(payload: unknown, ctx: IntegrationContext): Promise<NormalizedMessage[]> {
    const records = payload as SAPRecord[];
    return records.map(record => ({
      sourceId: record.ObjectKey,
      sender: record.CreatedBy ?? "SAP",
      content: `${record.ObjectType}: ${record.Description}`,
      sourceDate: new Date(record.CreatedAt),
      metadata: { objectType: record.ObjectType, plant: record.Plant },
      rawData: record,
    }));
  }
}

Companies that use this: MuleSoft specializes in enterprise system connectors (SAP, Oracle, Salesforce). Boomi connects enterprise ERPs via API keys.

Reference: https://www.mulesoft.com/resources/api/sap-integration

---

Type D — File Upload (user uploads a file, you process it)

Use for: CSV, MS Excel

No API, no OAuth. The org drags and drops a file. You parse it and normalize each row as a message.

Step 1: User uploads CSV/Excel via file picker in the UI
Step 2: File lands in S3 (already built — see upload router)
Step 3: You parse the file server-side
Step 4: Each row becomes a NormalizedMessage
Step 5: Ingest → queue → workflow (same as always)

Router file: integrations/csv/csv-router.ts

csvRouter.post("/api/integrations/csv/upload", async (c) => {
  const formData = await c.req.formData();
  const file = formData.get("file") as File;
  const orgId = formData.get("orgId") as string;

  const text = await file.text();
  const rows = parseCSV(text);  // use 'csv-parse' or similar

  // Each row becomes a message
  const messages: NormalizedMessage[] = rows.map((row, i) => ({
    sourceId: `${file.name}-row-${i}`,
    sender: "csv-upload",
    content: JSON.stringify(row),        // workflow sees the row as JSON string
    metadata: { fileName: file.name, rowIndex: i, columns: Object.keys(row) },
    rawData: row,
  }));

  const messageService = new IntegrationMessageService(prisma);
  // ... same 4-step pattern
});

Why this approach: Segment and Airbyte handle CSV as just another source type — each row is a record, normalized to the same schema as any other integration.

---

File Structure (What to Create)

server/src/integrations/
├── base/
│   ├── registry.ts          ← ALREADY EXISTS — don't touch
│   └── types.ts             ← ALREADY EXISTS — don't touch
├── gmail/                   ← ALREADY EXISTS — reference, don't touch
├── slack/                   ← ALREADY EXISTS — reference, don't touch
│
├── google-drive/            ← YOU CREATE
│   ├── google-drive.ts      ← Adapter (normalize file changes)
│   └── google-drive-router.ts ← OAuth callback + poll endpoint
│
├── notion/                  ← YOU CREATE
│   ├── notion.ts
│   └── notion-router.ts
│
├── hubspot/                 ← YOU CREATE
│   ├── hubspot.ts
│   └── hubspot-router.ts
│
├── discord/                 ← YOU CREATE
│   ├── discord.ts
│   └── discord-router.ts
│
├── whatsapp/                ← YOU CREATE
│   ├── whatsapp.ts
│   └── whatsapp-router.ts
│
├── sap/                     ← YOU CREATE
│   ├── sap.ts
│   └── sap-router.ts
│
├── ms-dynamics/             ← YOU CREATE
│   ├── ms-dynamics.ts
│   └── ms-dynamics-router.ts
│
├── oracle/                  ← YOU CREATE
│   ├── oracle.ts
│   └── oracle-router.ts
│
├── tableau/                 ← YOU CREATE
│   ├── tableau.ts
│   └── tableau-router.ts
│
├── csv/                     ← YOU CREATE
│   ├── csv.ts
│   └── csv-router.ts
│
├── ms-excel/                ← YOU CREATE
│   ├── ms-excel.ts
│   └── ms-excel-router.ts
│
├── message-ingestion.ts     ← ALREADY EXISTS — don't touch
└── index.ts                 ← ADD your adapter registrations here

---

Build Order (What to Do First)

Priority	Integration	Type	Why first
1	HubSpot	OAuth + Webhook	Most requested CRM, clean API, great docs
2	Google Drive	OAuth + Poll	Token refresh already built in Gmail, 80% reusable
3	Notion	OAuth + Poll	Popular with teams already on Coyax
4	WhatsApp	Webhook (Meta API)	Trigger type already in TriggerResolver
5	Discord	OAuth + Webhook	Similar to Slack — copy Slack adapter
6	CSV / Excel	File upload	No OAuth needed, fastest to build
7	Tableau	API key + Poll	Read-only data export
8	SAP / MS Dynamics / Oracle	API key + Poll	Enterprise, need customer env to test

---

Checklist for Every New Integration

[ ] 1. Add IntegrationType enum value in schema.prisma
[ ] 2. Run: bunx prisma migrate dev --name add-{name}-type
[ ] 3. Add trigger node mapping in trigger-resolver.ts:88
[ ] 4. Create integrations/{name}/{name}.ts — implement IntegrationAdapter
[ ] 5. Create integrations/{name}/{name}-router.ts — OAuth + data ingestion
[ ] 6. Register in integrations/index.ts
[ ] 7. Register router in server/src/index.ts (add app.route("/", yourRouter))
[ ] 8. Add env vars to server/src/env.ts and .env
[ ] 9. Test: connect org → send a message → check integrationMessage table

---

What You Can Copy Directly

What you need	Copy from	Lines
OAuth callback structure	gmail-router.ts	24–200
Webhook ingestion + 4-step pattern	gmail-router.ts	204–385
Signature verification	slack.ts	22–48
Normalize structure	gmail.ts	72–126
Register adapter	integrations/index.ts	1–10
Token storage (upsert pattern)	slack-router.ts	138–171

---

What Never Changes (Don't Touch These)

File	Why untouched
integrations/base/types.ts	The interface all adapters implement
integrations/base/registry.ts	The adapter lookup map
integrations/message-ingestion.ts	Dedup + save logic — same for all
services/message-router.ts	Queue routing — same for all
services/queues/message-queue.ts	BullMQ queue — same for all
services/workers/message-worker.ts	Worker loop — same for all

These files run for Gmail and Slack today. Your new integration automatically uses them the moment you call messageService.ingest() and router.route().