Data flow + steps

# Database Integration — Step by Step Implementation

Each step has: what to do → the code → how to test before moving on.

---

## Data Flow

There are three separate flows. Each is triggered by a different user action.

---

### Flow 1 — Connect (save credentials)

User fills the form and clicks **Connect**.

```
[Browser]
  │
  │  POST /api/integrations/database/connect
  │  { host, port, database, user, password, ssl, orgName }
  │
  ▼
[database-router.ts — /connect]
  │
  ├─ 1. Validate body with Zod schema
  │       missing field → 400 immediately, nothing written to DB
  │
  ├─ 2. Lookup org by lowercaseName
  │       not found → 404
  │
  ├─ 3. Open a pg.Client with the provided credentials
  │       run SELECT 1
  │       ┌─ throws → return { ok: false, error: "Cannot connect: ..." } 400
  │       └─ passes → close client
  │
  ├─ 4. encryptConfig({ host, port, database, user, password, ssl })
  │       AES-256-GCM → { iv, data, tag, __encrypted: "true" }
  │       [server/src/utils/crypto.ts]
  │
  ├─ 5. prisma.integration.upsert
  │       where: { orgId, type: "DATABASE" }
  │       config = encrypted blob
  │       [Integration table in our DB]
  │
  └─ return { ok: true }

[Browser]
  toast.success("Database connected!")
  drawer closes
  useFindManyIntegration refetches → card shows "Disconnect"
```

---

### Flow 2 — List Tables

User opens the drawer after connecting, or the frontend calls this before showing the Sync UI.

```
[Browser]
  │
  │  GET /api/integrations/database/tables?orgName=xxx
  │
  ▼
[database-router.ts — /tables]
  │
  ├─ 1. Lookup org
  │
  ├─ 2. prisma.integration.findUnique({ orgId, type: "DATABASE" })
  │       not found → 404 "Not connected"
  │
  ├─ 3. decryptConfig(integration.config)
  │       AES-256-GCM decrypt → { host, port, database, user, password, ssl }
  │       [server/src/utils/crypto.ts]
  │
  ├─ 4. Open pg.Client with decrypted credentials
  │
  ├─ 5. Query information_schema.tables + pg_class for row estimates
  │       SELECT table_name, reltuples AS row_estimate
  │       WHERE table_schema = 'public'
  │
  └─ return { ok: true, tables: [{ table_name, row_estimate }, ...] }

[Browser]
  renders table list with checkboxes
  user selects which tables to sync
```

---

### Flow 3 — Sync (import rows)

User selects tables and clicks **Sync Now**.

```
[Browser]
  │
  │  POST /api/integrations/database/sync
  │  { orgName, tables: ["users", "orders"] }
  │
  ▼
[database-router.ts — /sync]
  │
  ├─ 1. Lookup org + integration (same as /tables)
  │
  ├─ 2. decryptConfig → open pg.Client
  │
  ├─ 3. For each table in tables[]:
  │   │
  │   ├─ a. Discover primary key
  │   │       SELECT column_name FROM information_schema WHERE constraint_type = 'PRIMARY KEY'
  │   │       fallback: "id"
  │   │
  │   └─ b. Batch loop (500 rows at a time):
  │           SELECT * FROM "tableName" ORDER BY pk LIMIT 500 OFFSET N
  │           │
  │           ▼
  │       [DatabaseAdapter.normalize()]              database.ts
  │           converts each row →
  │           NormalizedMessage {
  │             sourceId:      "users:42"            ← dedup key
  │             conversationId: "users"              ← table name
  │             sender:        "database"
  │             content:       '{"id":42,"name":"Alice"}'
  │             metadata:      { tableName, primaryKey, primaryKeyValue, syncedAt }
  │             rawData:       { id: 42, name: "Alice" }
  │           }
  │           │
  │           ▼
  │       [IntegrationMessageService.ingest()]       message-ingestion.ts
  │           ├─ pre-filter: findMany existing sourceIds
  │           ├─ remove already-seen rows
  │           └─ createMany({ skipDuplicates: true })
  │               → IntegrationMessage table in our DB
  │
  ├─ 4. Close pg.Client
  │
  ├─ 5. Update Integration.config with lastSyncAt timestamp
  │
  └─ return { ok: true, summary: { users: { inserted: 38, deduped: 4 } } }

[Browser]
  toast.success("Sync complete")
  shows summary per table
```

---

### Where data lives after sync

```
External DB (org's PostgreSQL)           Our DB (Coyax)
─────────────────────────────            ──────────────────────────────────────
users table                              IntegrationMessage table
  id | name  | email                       sourceId        = "users:1"
  1  | Alice | alice@x.com    ──────►      conversationId  = "users"
  2  | Bob   | bob@x.com                   content         = '{"id":1,...}'
                                           metadata        = { tableName, pk }
orders table                               rawData         = { id:1, name:... }
  id | amount | status                     integrationId   = Integration.id
  1  | 500    | paid          ──────►      orgId           = org.clerkId
  2  | 200    | pending                    sourceType      = DATABASE
                                           status          = RECEIVED
```

---

### How this connects to the existing Gmail/Slack pipeline

```
Gmail  ──► GmailAdapter.normalize()   ──►┐
Slack  ──► SlackAdapter.normalize()   ──►├──► IntegrationMessageService.ingest()
Database ► DatabaseAdapter.normalize() ──►┘         └──► IntegrationMessage table

All three sources write to the same table.
All three use the same deduplication (sourceId unique constraint).
Database sync does NOT go through BullMQ — it is synchronous in the HTTP request.
Gmail/Slack go through BullMQ → Worker → TriggerResolver → WorkflowEngine (future for DB too).
```

---

## Step 1 — Add `DATABASE` to the enum

**File:** `server/schema.zmodel` around line 121

**What to change:**
```prisma
// BEFORE
enum IntegrationType {
    GMAIL
    WHATSAPP
    SLACK
}

// AFTER
enum IntegrationType {
    GMAIL
    WHATSAPP
    SLACK
    DATABASE
}
```

**Run after the change:**
```bash
cd server
bunx prisma migrate dev --name add_database_integration_type
bunx zenstack generate
```

**Test — open a bun shell and verify Prisma knows about it:**
```bash
cd server
bun run -e "import { IntegrationType } from '@prisma/client'; console.log(IntegrationType.DATABASE)"
# Expected output: DATABASE
```

If it prints `DATABASE`, the enum is live. If it throws, the migration did not run.

---

## Step 2 — Create the crypto helper

Passwords cannot be stored as plaintext. Before writing the router, we need an `encryptConfig` / `decryptConfig` utility.

**File:** `server/src/utils/crypto.ts` (new file)

```typescript
import { createCipheriv, createDecipheriv, randomBytes } from "crypto";

const ALGORITHM = "aes-256-gcm";

function getKey(): Buffer {
  const key = process.env.INTEGRATION_ENCRYPTION_KEY;
  if (!key) throw new Error("INTEGRATION_ENCRYPTION_KEY env var is not set");
  const buf = Buffer.from(key, "hex");
  if (buf.length !== 32) throw new Error("INTEGRATION_ENCRYPTION_KEY must be 32 bytes (64 hex chars)");
  return buf;
}

export function encryptConfig(data: Record<string, unknown>): Record<string, string> {
  const iv = randomBytes(12);
  const cipher = createCipheriv(ALGORITHM, getKey(), iv);
  const json = JSON.stringify(data);
  const encrypted = Buffer.concat([cipher.update(json, "utf8"), cipher.final()]);
  const authTag = cipher.getAuthTag();
  return {
    iv: iv.toString("hex"),
    data: encrypted.toString("hex"),
    tag: authTag.toString("hex"),
    __encrypted: "true",
  };
}

export function decryptConfig(config: Record<string, string>): Record<string, unknown> {
  if (config.__encrypted !== "true") return config; // not encrypted — pass through (dev only)
  const decipher = createDecipheriv(
    ALGORITHM,
    getKey(),
    Buffer.from(config.iv, "hex")
  );
  decipher.setAuthTag(Buffer.from(config.tag, "hex"));
  const decrypted = Buffer.concat([
    decipher.update(Buffer.from(config.data, "hex")),
    decipher.final(),
  ]);
  return JSON.parse(decrypted.toString("utf8"));
}
```

**Add to your `.env`:**
```bash
# Generate a key once: node -e "console.log(require('crypto').randomBytes(32).toString('hex'))"
INTEGRATION_ENCRYPTION_KEY=your_64_hex_char_key_here
```

**Test file:** `server/src/utils/crypto.test.ts`

```typescript
import { describe, expect, test } from "bun:test";
import { encryptConfig, decryptConfig } from "./crypto";

// Set a test key before importing
process.env.INTEGRATION_ENCRYPTION_KEY = "a".repeat(64); // 32 bytes as hex

describe("crypto helpers", () => {
  test("encryptConfig returns iv, data, tag, __encrypted", () => {
    const result = encryptConfig({ host: "localhost", password: "secret" });
    expect(result.__encrypted).toBe("true");
    expect(result.iv).toBeDefined();
    expect(result.data).toBeDefined();
    expect(result.tag).toBeDefined();
    // raw password must not appear in any field
    expect(JSON.stringify(result)).not.toContain("secret");
  });

  test("decryptConfig round-trips correctly", () => {
    const original = { host: "localhost", port: 5432, password: "secret", ssl: false };
    const encrypted = encryptConfig(original);
    const decrypted = decryptConfig(encrypted);
    expect(decrypted).toEqual(original);
  });

  test("decryptConfig passes through unencrypted config (dev mode)", () => {
    const plain = { host: "localhost" };
    expect(decryptConfig(plain as any)).toEqual(plain);
  });

  test("decryptConfig throws if tag is tampered", () => {
    const encrypted = encryptConfig({ host: "localhost" });
    encrypted.tag = "00".repeat(16); // corrupt the auth tag
    expect(() => decryptConfig(encrypted)).toThrow();
  });
});
```

**Run:**
```bash
cd server && bun test src/utils/crypto.test.ts
# All 4 tests should pass
```

---

## Step 3 — Create the DatabaseAdapter

**File:** `server/src/integrations/database/database.ts` (new file)

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

interface DbRowPayload {
  tableName: string;
  primaryKey: string;
  row: Record<string, unknown>;
}

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

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

    return [
      {
        sourceId: `${tableName}:${pkValue}`,     // dedup key
        conversationId: tableName,               // group all rows of same table together
        sender: "database",
        content: JSON.stringify(row),            // full row stored as JSON string
        metadata: {
          tableName,
          primaryKey,
          primaryKeyValue: pkValue,
          syncedAt: new Date().toISOString(),
        },
        rawData: row,
        sourceDate: new Date(),
      },
    ];
  }
}
```

**Test file:** `server/src/integrations/database/database.test.ts`

```typescript
import { describe, expect, test } from "bun:test";
import { DatabaseAdapter } from "./database";

const adapter = new DatabaseAdapter();
const ctx = { orgId: "org_test" };

describe("DatabaseAdapter.normalize()", () => {
  test("type is DATABASE", () => {
    expect(adapter.type).toBe("DATABASE");
  });

  test("sourceId is tableName:pkValue", async () => {
    const result = await adapter.normalize(
      { tableName: "users", primaryKey: "id", row: { id: 42, name: "Alice" } },
      ctx
    );
    expect(result[0].sourceId).toBe("users:42");
  });

  test("conversationId is the table name", async () => {
    const result = await adapter.normalize(
      { tableName: "orders", primaryKey: "id", row: { id: 1 } },
      ctx
    );
    expect(result[0].conversationId).toBe("orders");
  });

  test("content is the full row as JSON", async () => {
    const row = { id: 1, name: "Alice", email: "alice@example.com" };
    const result = await adapter.normalize(
      { tableName: "users", primaryKey: "id", row },
      ctx
    );
    expect(JSON.parse(result[0].content)).toEqual(row);
  });

  test("metadata contains tableName, primaryKey, primaryKeyValue", async () => {
    const result = await adapter.normalize(
      { tableName: "products", primaryKey: "sku", row: { sku: "ABC-1", price: 99 } },
      ctx
    );
    expect(result[0].metadata?.tableName).toBe("products");
    expect(result[0].metadata?.primaryKey).toBe("sku");
    expect(result[0].metadata?.primaryKeyValue).toBe("ABC-1");
  });

  test("returns exactly one NormalizedMessage per row", async () => {
    const result = await adapter.normalize(
      { tableName: "users", primaryKey: "id", row: { id: 1 } },
      ctx
    );
    expect(result).toHaveLength(1);
  });

  test("sender is always 'database'", async () => {
    const result = await adapter.normalize(
      { tableName: "users", primaryKey: "id", row: { id: 1 } },
      ctx
    );
    expect(result[0].sender).toBe("database");
  });
});
```

**Run:**
```bash
cd server && bun test src/integrations/database/database.test.ts
# All 6 tests should pass
```

---

## Step 4 — Register the adapter

**File:** `server/src/integrations/index.ts`

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

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

export { IntegrationRegistry } from "./base/registry";
export { IntegrationMessageService } from "./message-ingestion";

// AFTER — add 2 lines
import { IntegrationRegistry } from "./base/registry";
import { SlackAdapter } from "./slack/slack";
import { GmailAdapter } from "./gmail/gmail";
import { DatabaseAdapter } from "./database/database";       // ← add

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

export { IntegrationRegistry } from "./base/registry";
export { IntegrationMessageService } from "./message-ingestion";
```

**Test — registry lookup works:**

```typescript
// server/src/integrations/database/database-registry.test.ts
import { describe, expect, test } from "bun:test";

// Import the index — this triggers all registrations
import "~/server/integrations";
import { IntegrationRegistry } from "../base/registry";

describe("IntegrationRegistry — DATABASE", () => {
  test("DATABASE adapter is registered", () => {
    const adapter = IntegrationRegistry.get("DATABASE");
    expect(adapter).toBeDefined();
    expect(adapter.type).toBe("DATABASE");
  });

  test("get() throws for unregistered type", () => {
    expect(() => IntegrationRegistry.get("NOTION" as any)).toThrow();
  });
});
```

**Run:**
```bash
cd server && bun test src/integrations/database/database-registry.test.ts
```

---

## Step 5 — Create the router

**File:** `server/src/integrations/database/database-router.ts` (new file)

```typescript
import { Hono } from "hono";
import { z } from "zod";
import { Client } from "pg";
import type { Bindings } from "~server/index";
import { 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 ──────────────────────────────────
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 the connection 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);
  }

  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 ────────────────────────
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 Record<string, string>);
  const client = new Client({
    host: config.host as string,
    port: config.port as number,
    database: config.database as string,
    user: config.user as string,
    password: config.password as string,
    ssl: config.ssl ? { rejectUnauthorized: false } : false,
    connectionTimeoutMillis: 10_000,
  });

  try {
    await client.connect();
    const result = await client.query(
      `SELECT t.table_name, (c.reltuples)::bigint AS row_estimate
       FROM information_schema.tables t
       JOIN pg_class c ON c.relname = t.table_name
       WHERE t.table_schema = 'public'
       ORDER BY t.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 ─────────────────────────────────────
databaseIntegrationRouter.post("/api/integrations/database/sync", async (c) => {
  const prisma = getPrisma(c);
  const { orgName, tables } = await c.req.json() as { orgName: string; tables: string[] };

  if (!orgName || !tables?.length) {
    return c.json({ ok: false, error: "orgName and tables are required" }, 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: { id: true, config: true },
  });
  if (!integration) return c.json({ ok: false, error: "Not connected" }, 404);

  const config = decryptConfig(integration.config as Record<string, string>);
  const client = new Client({
    host: config.host as string,
    port: config.port as number,
    database: config.database as string,
    user: config.user as string,
    password: config.password as string,
    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) {
    // Discover 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();

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

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

**Test file:** `server/src/integrations/database/database-router.test.ts`

```typescript
import { describe, expect, mock, test, beforeEach } from "bun:test";

// Set encryption key before anything imports
process.env.INTEGRATION_ENCRYPTION_KEY = "a".repeat(64);

// ─── Mock pg Client ────────────────────────────────────────────────────────────
const mockQuery = mock(async (sql: string, params?: unknown[]) => ({ rows: [] }));
const mockConnect = mock(async () => {});
const mockEnd = mock(async () => {});

mock.module("pg", () => ({
  Client: class {
    connect = mockConnect;
    end = mockEnd;
    query = mockQuery;
  },
}));

// ─── Mock Prisma ───────────────────────────────────────────────────────────────
const fakeOrg = { clerkId: "org_abc" };
const fakeIntegration = {
  id: "int_1",
  config: { host: "localhost", port: 5432, database: "mydb", user: "u", password: "p", ssl: false },
};

mock.module("~server/utils/utils", () => ({
  getPrisma: () => ({
    org: { findFirst: async () => fakeOrg },
    integration: {
      findUnique: async () => fakeIntegration,
      upsert: async () => fakeIntegration,
      update: async () => fakeIntegration,
    },
    integrationMessage: {
      findMany: async () => [],
      createMany: async () => ({ count: 1 }),
    },
  }),
  getEnv: () => ({}),
}));

const { databaseIntegrationRouter } = await import("./database-router");

// ─── Tests ─────────────────────────────────────────────────────────────────────
describe("POST /api/integrations/database/connect", () => {
  test("returns 400 if body is missing required fields", async () => {
    const req = new Request("http://localhost/api/integrations/database/connect", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ host: "localhost" }), // missing user, password, database, orgName
    });
    const res = await databaseIntegrationRouter.fetch(req);
    expect(res.status).toBe(400);
    const json = await res.json() as any;
    expect(json.ok).toBe(false);
  });

  test("returns 200 when connection succeeds", async () => {
    mockQuery.mockResolvedValueOnce({ rows: [{ "?column?": 1 }] }); // SELECT 1
    const req = new Request("http://localhost/api/integrations/database/connect", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        host: "localhost", port: 5432, database: "mydb",
        user: "u", password: "p", ssl: false, orgName: "test-org",
      }),
    });
    const res = await databaseIntegrationRouter.fetch(req);
    expect(res.status).toBe(200);
    const json = await res.json() as any;
    expect(json.ok).toBe(true);
  });

  test("returns 400 if pg connection throws", async () => {
    mockConnect.mockRejectedValueOnce(new Error("Connection refused"));
    const req = new Request("http://localhost/api/integrations/database/connect", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        host: "bad-host", port: 5432, database: "mydb",
        user: "u", password: "p", ssl: false, orgName: "test-org",
      }),
    });
    const res = await databaseIntegrationRouter.fetch(req);
    expect(res.status).toBe(400);
    const json = await res.json() as any;
    expect(json.ok).toBe(false);
    expect(json.error).toContain("Connection refused");
  });
});

describe("GET /api/integrations/database/tables", () => {
  test("returns table list from connected DB", async () => {
    mockQuery.mockResolvedValueOnce({
      rows: [
        { table_name: "users", row_estimate: 100 },
        { table_name: "orders", row_estimate: 500 },
      ],
    });
    const req = new Request("http://localhost/api/integrations/database/tables?orgName=test-org");
    const res = await databaseIntegrationRouter.fetch(req);
    expect(res.status).toBe(200);
    const json = await res.json() as any;
    expect(json.ok).toBe(true);
    expect(json.tables).toHaveLength(2);
    expect(json.tables[0].table_name).toBe("users");
  });

  test("returns 400 if orgName is missing", async () => {
    const req = new Request("http://localhost/api/integrations/database/tables");
    const res = await databaseIntegrationRouter.fetch(req);
    expect(res.status).toBe(400);
  });
});

describe("POST /api/integrations/database/sync", () => {
  test("returns summary with inserted count", async () => {
    // Primary key discovery
    mockQuery.mockResolvedValueOnce({ rows: [{ column_name: "id" }] });
    // First batch of rows
    mockQuery.mockResolvedValueOnce({ rows: [{ id: 1, name: "Alice" }, { id: 2, name: "Bob" }] });
    // Second batch — empty → stop
    mockQuery.mockResolvedValueOnce({ rows: [] });

    const req = new Request("http://localhost/api/integrations/database/sync", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ orgName: "test-org", tables: ["users"] }),
    });
    const res = await databaseIntegrationRouter.fetch(req);
    expect(res.status).toBe(200);
    const json = await res.json() as any;
    expect(json.ok).toBe(true);
    expect(json.summary.users).toBeDefined();
    expect(json.summary.users.inserted).toBeGreaterThanOrEqual(0);
  });

  test("returns 400 if tables is empty", async () => {
    const req = new Request("http://localhost/api/integrations/database/sync", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ orgName: "test-org", tables: [] }),
    });
    const res = await databaseIntegrationRouter.fetch(req);
    expect(res.status).toBe(400);
  });
});
```

**Run:**
```bash
cd server && bun test src/integrations/database/database-router.test.ts
```

---

## Step 6 — Mount the router in the server

**File:** `server/src/index.ts`

```typescript
// Add to imports (around line 23)
import { databaseIntegrationRouter } from '~server/integrations/database/database-router';

// Add to route mounting (around line 210, after gmailIntegrationRouter)
app.route('', slackIntegrationRouter);
app.route('', gmailIntegrationRouter);
app.route('', databaseIntegrationRouter);   // ← add
```

**Test — start the server and hit it with curl:**
```bash
cd server && bun run dev &

# Test: missing body → 400
curl -s -X POST http://localhost:3000/api/integrations/database/connect \
  -H "Content-Type: application/json" \
  -d '{}' | jq .
# Expected: { "ok": false, "error": { ... } }

# Test: route exists (don't need a real DB for this check)
curl -s http://localhost:3000/api/integrations/database/tables?orgName=test | jq .
# Expected: { "ok": false, "error": "Org not found" }  ← 404, route is live
```

---

## Step 7 — Frontend: add the Database card

**File:** `client/src/routes/$orgName/integrations/-integrations.tsx`

**Change 1 — add to `CONNECTIONS` array (after Whatsapp, around line 73):**
```tsx
{
  title: 'DataBase',
  description: 'Connect a PostgreSQL database to import your data directly into Coyax.',
  image: '/integrations/database.png',
  logoScale: 1.0,
},
```

**Change 2 — add `DATABASE` to the query (around line 83):**
```tsx
type: { in: ['SLACK', 'GMAIL', 'DATABASE'] },
```

**Change 3 — add connected state (around line 91):**
```tsx
const dbConnected = integrations?.some(
  integration => integration.type === 'DATABASE' && integration.isActive
) ?? false;
```

**Change 4 — add to the `connectionsWithState` map (around line 114):**
```tsx
enabled:
  c.title === 'Slack' ? slackConnected :
  c.title === 'Gmail' ? gmailConnected :
  c.title === 'DataBase' ? dbConnected :
  false,
```

**Change 5 — add `DatabaseConnectDrawer` at the bottom of the file (before the last `}`):**
```tsx
function DatabaseConnectDrawer(props: { opened: boolean; onClose: () => void; onSuccess: () => 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 (closeOnSuccess: boolean) => {
    closeOnSuccess ? setSaving(true) : setTesting(true);
    try {
      const res = await fetch('/api/integrations/database/connect', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ ...form, port: Number(form.port), orgName: org }),
      });
      const data = await res.json() as { ok: boolean; error?: string };
      if (data.ok) {
        toast.success(closeOnSuccess ? 'Database connected!' : 'Connection successful!');
        if (closeOnSuccess) { props.onSuccess(); props.onClose(); }
      } else {
        toast.error(data.error ?? 'Failed');
      }
    } finally {
      closeOnSuccess ? setSaving(false) : setTesting(false);
    }
  };

  return (
    <Drawer opened={props.opened} onClose={props.onClose} title="Connect Database" position="right" size="md">
      <Stack gap="sm">
        <TextInput
          label="Host"
          placeholder="localhost or 192.168.1.1"
          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.currentTarget.checked }))}
        />
        <Group justify="flex-end" mt="md">
          <Button variant="default" loading={testing} onClick={() => post(false)}>
            Test connection
          </Button>
          <Button loading={saving} onClick={() => post(true)}>
            Connect
          </Button>
        </Group>
      </Stack>
    </Drawer>
  );
}
```

**Change 6 — in `IntegrationCard`, open the drawer for Database instead of redirecting:**

In the `IntegrationCard` function, add a `const [dbDrawerOpen, { open: openDbDrawer, close: closeDbDrawer }] = useDisclosure(false);` and in the Connect button:

```tsx
onClick={() => {
  if (connection.title === 'DataBase') {
    openDbDrawer();
    return;
  }
  const url = connection.connectionUrl?.();
  if (url) window.location.href = url;
}}
```

And render the drawer below the `IntegrationSettingsDrawer`:
```tsx
{connection.title === 'DataBase' && (
  <DatabaseConnectDrawer
    opened={dbDrawerOpen}
    onClose={closeDbDrawer}
    onSuccess={() => queryClient.invalidateQueries({ ... })} // same invalidation as handleDisconnect
  />
)}
```

**Test — visual check:**
1. Start the dev server: `bun run dev` in both `/server` and `/client`
2. Open the integrations page
3. You should see a **DataBase** card with "Connect" button
4. Clicking Connect should open the drawer with the form fields
5. Fill in wrong credentials → "Test connection" should show an error toast
6. Fill in correct credentials → "Connect" should close the drawer and show success toast
7. After connecting, the card should show "Disconnect" instead of "Connect"

---

## Step 8 — Add a database logo

Drop a `database.png` into `client/public/integrations/database.png`.

You can use any PostgreSQL or database icon. Free options:
- [simpleicons.org/postgresql](https://simpleicons.org) → download `postgresql.svg`, convert to PNG
- Or use any database icon from [iconify.design](https://iconify.design)

**Test:**
- Image loads without a broken img icon on the card

---

## Full test run

Once all steps are done, run all integration tests together:

```bash
cd server
bun test src/utils/crypto.test.ts \
         src/integrations/database/database.test.ts \
         src/integrations/database/database-registry.test.ts \
         src/integrations/database/database-router.test.ts
```

Expected: all tests green, no failures.

---

## Checklist

- [ ] Step 1 — Enum added, migration run, zenstack generated
- [ ] Step 2 — `crypto.ts` created, `INTEGRATION_ENCRYPTION_KEY` in `.env`, 4 tests pass
- [ ] Step 3 — `database.ts` adapter created, 6 tests pass
- [ ] Step 4 — Adapter registered in `index.ts`, registry test passes
- [ ] Step 5 — Router created, 6 tests pass
- [ ] Step 6 — Router mounted in `server/src/index.ts`, curl checks pass
- [ ] Step 7 — Frontend card, drawer, and connected state added, visual check passes
- [ ] Step 8 — Logo added, image loads


Navigation Menu

Data flow + steps #23

Database Integration — Step by Step Implementation

Data Flow

Flow 1 — Connect (save credentials)

Flow 2 — List Tables

Flow 3 — Sync (import rows)

Where data lives after sync

How this connects to the existing Gmail/Slack pipeline

Step 1 — Add `DATABASE` to the enum

Step 2 — Create the crypto helper

Step 3 — Create the DatabaseAdapter

Step 4 — Register the adapter

Step 5 — Create the router

Step 6 — Mount the router in the server

Step 7 — Frontend: add the Database card

Step 8 — Add a database logo

Full test run

Checklist

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Data flow + steps #23

Description

Database Integration — Step by Step Implementation

Data Flow

Flow 1 — Connect (save credentials)

Flow 2 — List Tables

Flow 3 — Sync (import rows)

Where data lives after sync

How this connects to the existing Gmail/Slack pipeline

Step 1 — Add DATABASE to the enum

Step 2 — Create the crypto helper

Step 3 — Create the DatabaseAdapter

Step 4 — Register the adapter

Step 5 — Create the router

Step 6 — Mount the router in the server

Step 7 — Frontend: add the Database card

Step 8 — Add a database logo

Full test run

Checklist

Activity

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions

Step 1 — Add `DATABASE` to the enum