Integration

[Zack-Rider]

Database Integration Plan

---

Context

We are building an integration platform where orgs connect their external data sources to Coyax. Gmail and Slack are already built using an adapter pattern — each source has a normalize() method that converts raw data into a NormalizedMessage, which gets stored in the IntegrationMessage table.

The founder has asked to add a Database integration. The idea: an org provides their PostgreSQL connection details, we connect to their external DB, read their tables, and store those rows in our DB — so their data becomes available inside Coyax.

This is a pull/sync model (not webhook/push). Think Airbyte, Fivetran, Stitch. The org fills a form, not an OAuth screen.

Current state of the codebase:

• IntegrationType enum has: GMAIL, SLACK, WHATSAPP — needs DATABASE added
• Integration model exists with config: Json (for storing credentials encrypted) and @@unique([orgId, type]) (one integration per type per org — fine for MVP)
• IntegrationMessageService.ingest() already handles deduplication via sourceId — we reuse this directly
• IntegrationAdapter interface in base/types.ts is the contract every source must follow
• IntegrationRegistry in base/registry.ts is where adapters are registered at startup

---

Issue

There is no Database integration yet. The problems to solve are:

1. No DATABASE type in the enum — Prisma does not know about it, so nothing can be stored
2. No adapter — no normalize() function that converts a DB row into a NormalizedMessage
3. No backend endpoints — nowhere to save credentials, list tables, or trigger a sync
4. No frontend UI — the integrations page only handles OAuth redirects (Gmail, Slack), not a credentials form
5. Credentials must be encrypted — raw passwords cannot be stored in Integration.config as plaintext

Secondary constraints:

• Must not create new Prisma models — reuse IntegrationMessage for storing imported rows
• Must not create a new queue or worker — sync is synchronous (manual button) for MVP
• Password must survive a round-trip: encrypt on save, decrypt on sync

---

Ask

Add the Database integration end-to-end, touching only what is necessary:

1. Add DATABASE to the enum in schema.zmodel
2. Create DatabaseAdapter following the existing IntegrationAdapter interface
3. Register the adapter in integrations/index.ts
4. Create a Hono router with 3 endpoints: connect, list tables, sync
5. Mount the router in server/src/index.ts
6. Update the frontend integrations page: add DB card + form drawer

---

Why this approach — what the best companies do

Company	What they do	What we copy
Airbyte	Every source is a connector implementing one interface: read() → stream of records. All records go to the same destination table	Our DatabaseAdapter implements IntegrationAdapter. All rows go to IntegrationMessage
Fivetran	Connectors are stateless. Sync state (last run, schema) is stored in the destination DB, not in the connector code	lastSyncAt stored in Integration.config, not a new model
HubSpot	Credentials encrypted at rest. Test connection before saving anything. Never return raw credentials to frontend	encryptConfig() before upsert. POST /connect tests before storing
Notion / Linear	Form-based credential flow, no OAuth redirect needed for API key / DB integrations	Drawer with a form, closes on success — no page redirect
Stripe	Batch reads, never pull an entire table at once	Read 500 rows at a time with OFFSET/LIMIT

---

Why it will not fail

Risk	Prevention
Can't connect to external DB	/connect runs SELECT 1 before saving anything. If it fails, nothing is stored
Duplicate rows on re-sync	sourceId = tableName:primaryKeyValue — ingest() already dedupes via @@unique([integrationId, sourceId]) and skipDuplicates: true
Password stored as plaintext	AES-256-GCM encrypt before upsert, decrypt before connecting
External DB too slow / hangs	connectionTimeoutMillis: 10_000 and query_timeout: 10_000 on the pg client
Syncing millions of rows crashes memory	Batched: 500 rows read → normalize → ingest → repeat. Never loads full table
@@unique([orgId, type]) breaks if org has two DBs	Intentional for MVP. Lift constraint + add name field later when needed
External DB schema changes	rawData column stores the full row as JSON — old records are never broken
Org not found / integration not found	Every endpoint checks before proceeding, returns 404 with a clear message

---

Implementation

1. server/schema.zmodel — add one value

// Around line 121
// BEFORE
enum IntegrationType {
    GMAIL
    WHATSAPP
    SLACK
}

// AFTER
enum IntegrationType {
    GMAIL
    WHATSAPP
    SLACK
    DATABASE
}

Run after:

cd server
bunx prisma migrate dev --name add_database_integration_type
bunx zenstack generate

---

2. New file: server/src/integrations/database/database.ts

Same shape as gmail.ts and slack.ts. Implements IntegrationAdapter.

import type { IntegrationAdapter, IntegrationContext, NormalizedMessage } from "../base/types";

export class DatabaseAdapter implements IntegrationAdapter {
  type = "DATABASE" as const;

  async normalize(
    payload: unknown,
    _ctx: IntegrationContext
  ): Promise<NormalizedMessage[]> {
    const { tableName, primaryKey, row } = payload as {
      tableName: string;
      primaryKey: string;
      row: Record<string, unknown>;
    };

    return [
      {
        sourceId: `${tableName}:${row[primaryKey]}`,   // dedup key — table + PK value
        conversationId: tableName,                     // groups rows by table
        sender: "database",
        content: JSON.stringify(row),                  // full row as JSON string
        metadata: {
          tableName,
          primaryKey,
          primaryKeyValue: row[primaryKey],
          syncedAt: new Date().toISOString(),
        },
        rawData: row,
        sourceDate: new Date(),
      },
    ];
  }
}

---

3. server/src/integrations/index.ts — register (add 2 lines)

// BEFORE
import { IntegrationRegistry } from "./base/registry";
import { SlackAdapter } from "./slack/slack";
import { GmailAdapter } from "./gmail/gmail";

IntegrationRegistry.register(new SlackAdapter());
IntegrationRegistry.register(new GmailAdapter());

// AFTER
import { DatabaseAdapter } from "./database/database";   // ← add

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

---

4. New file: server/src/integrations/database/database-router.ts

Same Hono pattern as gmail-router.ts.

import { Hono } from "hono";
import { z } from "zod";
import { Client } from "pg";
import type { Bindings } from "~server/index";
import { getEnv, getPrisma } from "~server/utils/utils";
import { IntegrationMessageService } from "../message-ingestion";
import { DatabaseAdapter } from "./database";
import { encryptConfig, decryptConfig } from "~server/utils/crypto";

export const databaseIntegrationRouter = new Hono<{ Bindings: Bindings }>();

const DbCredentialsSchema = z.object({
  host: z.string().min(1),
  port: z.coerce.number().int().min(1).max(65535).default(5432),
  database: z.string().min(1),
  user: z.string().min(1),
  password: z.string().min(1),
  ssl: z.boolean().default(false),
  orgName: z.string().min(1),
});

// POST /api/integrations/database/connect
// Test connection → encrypt credentials → upsert Integration row
databaseIntegrationRouter.post("/api/integrations/database/connect", async (c) => {
  const prisma = getPrisma(c);
  const parsed = DbCredentialsSchema.safeParse(await c.req.json());
  if (!parsed.success) return c.json({ ok: false, error: parsed.error.flatten() }, 400);

  const { host, port, database, user, password, ssl, orgName } = parsed.data;

  const org = await prisma.org.findFirst({
    where: { lowercaseName: orgName.toLowerCase() },
    select: { clerkId: true },
  });
  if (!org) return c.json({ ok: false, error: "Org not found" }, 404);

  // Test before saving anything
  const client = new Client({
    host, port, database, user, password,
    ssl: ssl ? { rejectUnauthorized: false } : false,
    connectionTimeoutMillis: 10_000,
    query_timeout: 10_000,
  });

  try {
    await client.connect();
    await client.query("SELECT 1");
    await client.end();
  } catch (err) {
    return c.json({ ok: false, error: `Cannot connect: ${(err as Error).message}` }, 400);
  }

  // Encrypt before storing
  const encryptedConfig = encryptConfig({ host, port, database, user, password, ssl, connectedAt: new Date().toISOString() });

  await prisma.integration.upsert({
    where: { orgId_type: { orgId: org.clerkId, type: "DATABASE" } },
    create: {
      orgId: org.clerkId,
      type: "DATABASE",
      externalAccountId: `${host}:${port}/${database}`,
      config: encryptedConfig,
      isActive: true,
    },
    update: {
      externalAccountId: `${host}:${port}/${database}`,
      config: encryptedConfig,
      isActive: true,
    },
  });

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

// GET /api/integrations/database/tables?orgName=xxx
// List tables from the connected external DB
databaseIntegrationRouter.get("/api/integrations/database/tables", async (c) => {
  const prisma = getPrisma(c);
  const orgName = c.req.query("orgName");
  if (!orgName) return c.json({ ok: false, error: "Missing orgName" }, 400);

  const org = await prisma.org.findFirst({
    where: { lowercaseName: orgName.toLowerCase() },
    select: { clerkId: true },
  });
  if (!org) return c.json({ ok: false, error: "Org not found" }, 404);

  const integration = await prisma.integration.findUnique({
    where: { orgId_type: { orgId: org.clerkId, type: "DATABASE" } },
    select: { config: true },
  });
  if (!integration) return c.json({ ok: false, error: "Not connected" }, 404);

  const config = decryptConfig(integration.config as any);
  const client = new Client({ ...config, ssl: config.ssl ? { rejectUnauthorized: false } : false, connectionTimeoutMillis: 10_000 });

  try {
    await client.connect();
    const result = await client.query(
      `SELECT table_name, (reltuples)::bigint AS row_estimate
       FROM information_schema.tables t
       JOIN pg_class c ON c.relname = t.table_name
       WHERE table_schema = 'public'
       ORDER BY table_name`
    );
    await client.end();
    return c.json({ ok: true, tables: result.rows });
  } catch (err) {
    return c.json({ ok: false, error: (err as Error).message }, 500);
  }
});

// POST /api/integrations/database/sync
// Pull rows from selected tables → store as IntegrationMessages
databaseIntegrationRouter.post("/api/integrations/database/sync", async (c) => {
  const prisma = getPrisma(c);
  const { orgName, tables } = await c.req.json() as { orgName: string; tables: string[] };

  const org = await prisma.org.findFirst({
    where: { lowercaseName: orgName.toLowerCase() },
    select: { clerkId: true },
  });
  if (!org) return c.json({ ok: false, error: "Org not found" }, 404);

  const integration = await prisma.integration.findUnique({
    where: { orgId_type: { orgId: org.clerkId, type: "DATABASE" } },
    select: { id: true, config: true },
  });
  if (!integration) return c.json({ ok: false, error: "Not connected" }, 404);

  const config = decryptConfig(integration.config as any);
  const client = new Client({ ...config, ssl: config.ssl ? { rejectUnauthorized: false } : false, connectionTimeoutMillis: 10_000 });
  await client.connect();

  const adapter = new DatabaseAdapter();
  const ingestor = new IntegrationMessageService(prisma);
  const BATCH_SIZE = 500;
  const summary: Record<string, { inserted: number; deduped: number }> = {};

  for (const tableName of tables) {
    // Find primary key
    const pkResult = await client.query<{ column_name: string }>(
      `SELECT kcu.column_name FROM information_schema.table_constraints tc
       JOIN information_schema.key_column_usage kcu ON tc.constraint_name = kcu.constraint_name
       WHERE tc.table_name = $1 AND tc.constraint_type = 'PRIMARY KEY' LIMIT 1`,
      [tableName]
    );
    const primaryKey = pkResult.rows[0]?.column_name ?? "id";

    let offset = 0;
    let tableInserted = 0;
    let tableDeduped = 0;

    while (true) {
      const rows = await client.query<Record<string, unknown>>(
        `SELECT * FROM "${tableName}" ORDER BY "${primaryKey}" LIMIT $1 OFFSET $2`,
        [BATCH_SIZE, offset]
      );
      if (rows.rows.length === 0) break;

      const normalized = (
        await Promise.all(
          rows.rows.map((row) => adapter.normalize({ tableName, primaryKey, row }, { orgId: org.clerkId }))
        )
      ).flat();

      const result = await ingestor.ingest({
        integrationId: integration.id,
        orgId: org.clerkId,
        sourceType: "DATABASE",
        messages: normalized,
      });

      tableInserted += result.inserted;
      tableDeduped += result.deduped;
      offset += rows.rows.length;
      if (rows.rows.length < BATCH_SIZE) break;
    }

    summary[tableName] = { inserted: tableInserted, deduped: tableDeduped };
  }

  await client.end();

  // Save last sync time
  await prisma.integration.update({
    where: { id: integration.id },
    data: { config: { ...(integration.config as any), lastSyncAt: new Date().toISOString() } },
  });

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

---

5. server/src/index.ts — mount the router (add 2 lines)

// Around line 22 — imports
import { databaseIntegrationRouter } from '~server/integrations/database/database-router';  // ← add

// Around line 208 — route mounting
app.route('', slackIntegrationRouter);
app.route('', gmailIntegrationRouter);
app.route('', databaseIntegrationRouter);  // ← add

---

6. client/src/routes/$orgName/integrations/-integrations.tsx — frontend

a) Add to CONNECTIONS array (after Whatsapp):

{
  title: 'DataBase',
  description: 'Connect a PostgreSQL database to import your data directly into Coyax.',
  image: '/integrations/database.png',
  logoScale: 1.0,
},

b) Add DATABASE to the query filter:

const { data: integrations } = useFindManyIntegration({
  where: {
    org: { name: { equals: org, mode: 'insensitive' } },
    type: { in: ['SLACK', 'GMAIL', 'DATABASE'] },  // ← add DATABASE
  },
});

c) Add connected state:

const dbConnected = integrations?.some(
  integration => integration.type === 'DATABASE' && integration.isActive
) ?? false;

// In connectionsWithState map:
enabled:
  c.title === 'Slack' ? slackConnected :
  c.title === 'Gmail' ? gmailConnected :
  c.title === 'DataBase' ? dbConnected :
  false,

d) Add DatabaseConnectDrawer component (inside the same file):

function DatabaseConnectDrawer(props: { opened: boolean; onClose: () => void }) {
  const org = getOrgName();
  const [form, setForm] = useState({ host: '', port: '5432', database: '', user: '', password: '', ssl: false });
  const [testing, setTesting] = useState(false);
  const [saving, setSaving] = useState(false);

  const post = async () => {
    setSaving(true);
    const res = await fetch('/api/integrations/database/connect', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ ...form, orgName: org }),
    });
    const data = await res.json();
    setSaving(false);
    if (data.ok) { toast.success('Database connected!'); props.onClose(); }
    else toast.error(data.error ?? 'Connection failed');
  };

  return (
    <Drawer opened={props.opened} onClose={props.onClose} title="Connect Database" position="right" size="md">
      <Stack gap="sm">
        <TextInput label="Host" value={form.host} onChange={e => setForm(f => ({ ...f, host: e.target.value }))} />
        <TextInput label="Port" value={form.port} onChange={e => setForm(f => ({ ...f, port: e.target.value }))} />
        <TextInput label="Database name" value={form.database} onChange={e => setForm(f => ({ ...f, database: e.target.value }))} />
        <TextInput label="User" value={form.user} onChange={e => setForm(f => ({ ...f, user: e.target.value }))} />
        <PasswordInput label="Password" value={form.password} onChange={e => setForm(f => ({ ...f, password: e.target.value }))} />
        <Switch label="Use SSL" checked={form.ssl} onChange={e => setForm(f => ({ ...f, ssl: e.target.checked }))} />
        <Group justify="flex-end" mt="md">
          <Button variant="default" loading={testing} onClick={async () => {
            setTesting(true);
            const res = await fetch('/api/integrations/database/connect', {
              method: 'POST', headers: { 'Content-Type': 'application/json' },
              body: JSON.stringify({ ...form, orgName: org }),
            });
            const data = await res.json();
            setTesting(false);
            data.ok ? toast.success('Connection successful!') : toast.error(data.error);
          }}>Test connection</Button>
          <Button loading={saving} onClick={post}>Connect</Button>
        </Group>
      </Stack>
    </Drawer>
  );
}

---

File touch summary

File	Action	What changes
server/schema.zmodel	Edit	Add DATABASE to IntegrationType enum
server/src/integrations/database/database.ts	Create	DatabaseAdapter — normalize() converts one DB row to NormalizedMessage
server/src/integrations/database/database-router.ts	Create	3 endpoints: /connect, /tables, /sync
server/src/integrations/index.ts	Edit	Register DatabaseAdapter (2 lines)
server/src/index.ts	Edit	Mount databaseIntegrationRouter (2 lines)
client/src/routes/$orgName/integrations/-integrations.tsx	Edit	DB card, connected state, DatabaseConnectDrawer

Zero new Prisma models. Zero new queues. Zero new workers.

---

Build order

1. schema.zmodel → migrate → zenstack generate
2. database/database.ts (adapter)
3. database/database-router.ts (endpoints)
4. integrations/index.ts + server/src/index.ts (register + mount)
5. Frontend card + drawer

Each step is independently testable before moving to the next.

---

What is not in this MVP

Feature	Why deferred	When to add
Scheduled auto-sync	Adds BullMQ cron complexity	After manual sync is stable
Multiple DBs per org	Requires lifting @@unique constraint	When an org requests it
MySQL / SQL Server support	Swap pg for mysql2 — same adapter	Next connector after this one
Table selection UI	/tables endpoint is ready, need a checkbox UI	After connect flow works
Incremental sync (only new rows)	Track lastSyncAt per table	After full sync is stable