INtegration Plan final

[Zack-Rider]

Database Integration — System Design

Date: 2026-04-12
Author: Piyush
Status: READY FOR TEAM REVIEW
Goal: Let customers connect their own databases (PostgreSQL, MySQL, SAP, Oracle) to Coyax, select which tables they want, and have Coyax automatically sync that data so they can run business operations on it.

---

1. What We Are Building (Plain English)

A customer has their data in their own PostgreSQL, MySQL, SAP, or Oracle database. They want to bring that data into Coyax so they can use our workflows, AI agents, and views on top of it.

The flow is:

Customer's DB  →  Coyax syncs it  →  Lands in Coyax tables  →  Customer works on it
(PostgreSQL,       (on a schedule       (in their Org DB          (workflows, AI, views)
 MySQL, SAP,        or on demand)        in Neon/Postgres)
 Oracle)

Think of it like this: we are a "read pipe" from their DB into ours. We do NOT write back to their DB (v1). We just pull.

---

2. What Already Exists (We Build On Top of This)

Before adding anything, here is what we already have that we reuse:

Existing piece	Where it lives	How we use it
Integration model (in zmodel)	server/schema.zmodel	We extend the same IntegrationType enum with DATABASE
BullMQ queue (message-processing)	server/src/services/queues/message-queue.ts	We add a second queue: db-sync
Worker pattern	server/src/services/workers/message-worker.ts	We add a second worker: db-sync-worker.ts
IntegrationAdapter interface	server/src/integrations/base/types.ts	DB adapters implement the same pattern
IntegrationRegistry	server/src/integrations/base/registry.ts	Register DB adapters here
Org's Sequelize DB (Neon)	server/src/sql/templates4.ts	This is where synced data lands
tRPC routers	server/src/routers/	We add db-integration-router.ts
WorkflowRun + RunStatus	Prisma schema	We add DbSyncRun with same status pattern

We are NOT replacing anything. We are adding new pieces that follow the same patterns.

---

3. New Database Models

Add these to server/schema.zmodel (the ZenStack source, not schema.prisma directly):

// Enum additions
enum IntegrationDbType {
  POSTGRES
  MYSQL
  MSSQL
  ORACLE
  SAP_HANA
}

enum DbSyncStatus {
  PENDING
  RUNNING
  PARTIAL   // some tables succeeded, some failed
  SUCCESS
  FAILED
}

enum DbSyncTableStatus {
  PENDING
  RUNNING
  SUCCESS
  FAILED
  SKIPPED
}

// Stores the customer's DB connection
// credentials are AES-256 encrypted before storage
model DbIntegration {
  id            String              @id @default(uuid())
  org           Org                 @relation(fields: [orgId], references: [clerkId], onDelete: Cascade)
  orgId         String
  name          String              // e.g. "Production SAP", "Warehouse DB"
  dbType        IntegrationDbType
  host          String              // encrypted
  port          Int
  database      String              // database/schema name
  username      String              // encrypted
  passwordHash  String              // encrypted
  ssl           Boolean             @default(false)
  sslCert       String?             // encrypted, optional
  isActive      Boolean             @default(true)
  lastTestedAt  DateTime?
  lastTestOk    Boolean?
  createdAt     DateTime            @default(now())
  updatedAt     DateTime            @updatedAt
  tableMappings DbTableMapping[]
  syncRuns      DbSyncRun[]

  @@unique([orgId, name])
  @@index([orgId])
}

// Which tables to sync and how
model DbTableMapping {
  id              String         @id @default(uuid())
  integration     DbIntegration  @relation(fields: [integrationId], references: [id], onDelete: Cascade)
  integrationId   String
  sourceTable     String         // table name in the customer's DB
  targetTable     String         // table name in Coyax's Org DB
  columns         Json           // string[] — which columns to pull. Empty = all.
  primaryKey      String         // which column is the PK for upsert logic
  cronExpression  String         @default("0 * * * *")  // default: every hour
  isActive        Boolean        @default(true)
  lastSyncAt      DateTime?
  lastCursor      String?        // last synced PK value or timestamp, for incremental sync
  createdAt       DateTime       @default(now())
  updatedAt       DateTime       @updatedAt

  @@unique([integrationId, sourceTable])
  @@index([integrationId])
}

// One row per sync run (triggered by cron or manually)
model DbSyncRun {
  id            String         @id @default(uuid())
  integration   DbIntegration  @relation(fields: [integrationId], references: [id], onDelete: Cascade)
  integrationId String
  org           Org            @relation(fields: [orgId], references: [clerkId], onDelete: Cascade)
  orgId         String
  status        DbSyncStatus   @default(PENDING)
  triggeredBy   String         // "cron" | "manual" | userId
  startedAt     DateTime       @default(now())
  completedAt   DateTime?
  totalTables   Int            @default(0)
  succeededTables Int          @default(0)
  failedTables  Int            @default(0)
  errorSummary  String?
  tableRuns     DbSyncTableRun[]

  @@index([integrationId])
  @@index([orgId, status])
}

// One row per table per sync run
model DbSyncTableRun {
  id            String            @id @default(uuid())
  syncRun       DbSyncRun         @relation(fields: [syncRunId], references: [id], onDelete: Cascade)
  syncRunId     String
  sourceTable   String
  targetTable   String
  status        DbSyncTableStatus @default(PENDING)
  rowsRead      Int               @default(0)
  rowsWritten   Int               @default(0)
  rowsFailed    Int               @default(0)
  errorMessage  String?
  startedAt     DateTime          @default(now())
  completedAt   DateTime?

  @@index([syncRunId])
}

Why this structure:

• DbIntegration = the connection (one per customer DB). Credentials are always encrypted.
• DbTableMapping = which tables to sync, how often, and where the cursor is (for incremental sync).
• DbSyncRun = one row per time we run a sync. This is the "Job Started → N tables succeed → Remaining ones" from the whiteboard.
• DbSyncTableRun = one row per table per run. This is how we know exactly which table failed and why.

---

4. Security — Credential Encryption

This is the most important thing to get right. If we store raw passwords in the DB and we get breached, every customer's database is exposed.

We use AES-256-GCM encryption with a secret key stored in an environment variable (never in the DB).

// server/src/utils/crypto.ts  (new file, ~30 lines)

const ENCRYPTION_KEY = process.env.DB_CREDENTIAL_ENCRYPTION_KEY  // must be 32 bytes hex

encrypt(plaintext: string) → { encrypted: string, iv: string, tag: string }
decrypt(encrypted: string, iv: string, tag: string) → plaintext

Store in DB: host_encrypted, username_encrypted, password_encrypted — all as JSON { encrypted, iv, tag }.

When we need to connect: decrypt at runtime, use, discard. Never log credentials.

What to put in .env:

DB_CREDENTIAL_ENCRYPTION_KEY=<32-byte random hex, generated once, kept secret>

---

5. Connection Flow (User Connects a DB)

This is the wizard the user sees in the UI. Three steps:

Step 1 — Fill in connection details

The UI form (from the whiteboard):

• DB Type (dropdown: PostgreSQL, MySQL, SQL Server, Oracle, SAP HANA)
• Host
• Port (auto-filled based on DB type: Postgres=5432, MySQL=3306, Oracle=1521, etc.)
• Database name
• Username
• Password
• SSL toggle

Step 2 — Test connection (before saving)

User clicks "Click and test connection." This calls a tRPC mutation that:

1. Does NOT save to DB yet
2. Takes the raw credentials from the request
3. Opens a test connection to the customer's DB (with a 5-second timeout)
4. Runs SELECT 1 (or equivalent for each DB type)
5. Closes the connection immediately
6. Returns { success: true } or { success: false, error: "Connection refused" }

The UI shows the status indicators from the whiteboard:

• Testing... (spinner)
• Connected successfully (green check)
• Failed: [error message] (red)

// tRPC endpoint: dbIntegration.testConnection
// Input: { dbType, host, port, database, username, password, ssl }
// Output: { ok: boolean, error?: string, latencyMs: number }
// Note: credentials are NOT stored at this point

Step 3 — Discover tables and save

After a successful test, the same connection is used to:

1. Query the DB's information schema for a list of tables
2. Show the user a table picker (with column previews)
3. User selects which tables to sync and sets a schedule per table
4. User clicks Save → credentials are encrypted → DbIntegration + DbTableMapping rows are created

// tRPC endpoint: dbIntegration.discoverTables
// Input: { dbType, host, port, database, username, password, ssl }
// Output: { tables: [{ name, columns: [{ name, type }], rowCount }] }

---

6. Sync Execution — How a Sync Job Actually Works

6.1 What triggers a sync

Two ways:

1. Cron — a background scheduler checks DbTableMapping rows where isActive=true and nextRunAt <= now. We compute nextRunAt from the cronExpression using the cron-parser npm package.

2. Manual — user clicks "Sync now" in the UI → tRPC mutation → enqueue job immediately.

6.2 The sync queue

We add a second BullMQ queue called db-sync (separate from the existing message-processing queue so they do not interfere).

// server/src/services/queues/db-sync-queue.ts
Queue name: "db-sync"
Job payload: { integrationId, orgId, triggeredBy }
Job options:
  - attempts: 1  (we handle retries per-table inside the worker, not at job level)
  - removeOnComplete: after 24h
  - removeOnFail: after 7 days

6.3 The sync worker (the important part)

// server/src/services/workers/db-sync-worker.ts

Here is exactly what happens when a sync job runs. This maps to "Job Started → N tables succeed → Remaining ones" from the whiteboard:

1.  Create DbSyncRun row (status=RUNNING, totalTables=N)

2.  Decrypt credentials from DbIntegration

3.  Open one connection to the customer's DB
    - Use a connection pool with max 3 connections (do not hammer their DB)
    - Set connection timeout: 10 seconds

4.  For each active DbTableMapping (run in series, not parallel):

    4a. Create DbSyncTableRun row (status=RUNNING)

    4b. Fetch rows from source table:
        - If lastCursor is null: full sync (SELECT * FROM table LIMIT 10000)
        - If lastCursor has a value: incremental sync
          (SELECT * FROM table WHERE updated_at > lastCursor OR id > lastCursor LIMIT 10000)
        - Fetch in batches of 500 rows

    4c. For each batch:
        - Transform: map source column names to target column names
        - Upsert into Coyax's Org DB using Sequelize
          (INSERT ... ON CONFLICT (primaryKey) DO UPDATE SET ...)
        - Increment rowsWritten counter

    4d. On success:
        - Update DbTableMapping.lastCursor to the latest value
        - Update DbTableMapping.lastSyncAt = now()
        - Mark DbSyncTableRun (status=SUCCESS, rowsRead, rowsWritten)
        - Increment DbSyncRun.succeededTables

    4e. On failure for this table:
        - Log the error
        - Mark DbSyncTableRun (status=FAILED, errorMessage)
        - Increment DbSyncRun.failedTables
        - CONTINUE to the next table (do NOT stop the whole job)
        - This is the "remaining ones" from the whiteboard

5.  Close the customer DB connection

6.  Update DbSyncRun:
    - If failedTables = 0: status=SUCCESS
    - If failedTables > 0 AND succeededTables > 0: status=PARTIAL
    - If succeededTables = 0: status=FAILED
    - completedAt = now()

7.  Send notification to the user:
    - On SUCCESS: "Sync completed: 5 tables synced"
    - On PARTIAL: "Sync partially completed: 3/5 tables synced. 2 failed."
    - On FAILED: "Sync failed: connection error"

Key design rule: A failure in one table NEVER stops the other tables. Each table is independent. This is what makes the system reliable.

6.4 Incremental sync (cursor-based)

Full sync every time is expensive. After the first sync, we only pull new/changed rows.

How it works:

• Each DbTableMapping has a lastCursor — the highest value of updated_at (or id) we have seen
• On next sync: WHERE updated_at > lastCursor (if the table has an updated_at column) or WHERE id > lastCursor (if no timestamp)
• After successful sync: update lastCursor to the max value from this batch

The user picks the cursor column when setting up the table mapping (the UI shows a dropdown of columns from the table).

---

7. DB Adapters (How We Support Multiple DB Types)

We use the same adapter pattern that already exists for Gmail/Slack:

// server/src/integrations/database/base/db-adapter.ts

interface DbAdapter {
  dbType: IntegrationDbType
  connect(config: DecryptedDbConfig): Promise<DbConnection>
  testConnection(config: DecryptedDbConfig): Promise<{ ok: boolean; latencyMs: number }>
  discoverTables(conn: DbConnection): Promise<TableInfo[]>
  fetchRows(conn: DbConnection, table: string, cursor?: string, limit?: number): Promise<Row[]>
  disconnect(conn: DbConnection): Promise<void>
}

One file per DB type:

server/src/integrations/database/
  base/
    db-adapter.ts          ← interface
    db-registry.ts         ← register adapters, same pattern as IntegrationRegistry
  adapters/
    postgres-adapter.ts    ← uses `pg` npm package (BUILD THIS FIRST)
    mysql-adapter.ts       ← uses `mysql2` npm package
    mssql-adapter.ts       ← uses `mssql` npm package
    oracle-adapter.ts      ← uses `oracledb` npm package (requires Oracle client — defer)
    sap-hana-adapter.ts    ← uses `@sap/hana-client` (enterprise only — defer)

Build order: PostgreSQL first (it covers most use cases and is the easiest). MySQL second. SAP and Oracle are complex — defer to a later phase.

PostgreSQL adapter (how to write it)

// uses: import { Pool } from 'pg'

class PostgresAdapter implements DbAdapter {
  dbType = IntegrationDbType.POSTGRES

  async connect(config) {
    const pool = new Pool({
      host: config.host,
      port: config.port,
      database: config.database,
      user: config.username,
      password: config.password,
      ssl: config.ssl ? { rejectUnauthorized: false } : false,
      max: 3,               // max 3 connections — do not hammer customer DB
      connectionTimeoutMillis: 10000,
      idleTimeoutMillis: 30000,
    })
    await pool.query('SELECT 1')  // verify it works
    return pool
  }

  async discoverTables(conn) {
    // query information_schema.tables and information_schema.columns
    const tables = await conn.query(`
      SELECT table_name FROM information_schema.tables
      WHERE table_schema = 'public' AND table_type = 'BASE TABLE'
      ORDER BY table_name
    `)
    // for each table, get columns
    // return [{ name, columns: [{name, type}], rowCount }]
  }

  async fetchRows(conn, table, cursor, limit = 500) {
    // builds safe parameterized query — NEVER string-concatenate table/column names
    // uses pg-format or manual sanitization for table/column identifiers
  }
}

---

8. The Scheduler (Cron Trigger)

The scheduler is a simple background loop. It runs every minute inside the same server process (no separate service needed).

// server/src/services/db-sync-scheduler.ts

On server startup:
  setInterval(checkAndEnqueueDueJobs, 60_000)  // every 60 seconds

checkAndEnqueueDueJobs():
  1. Query all DbTableMapping where isActive=true
  2. Group by integrationId
  3. For each integration where ANY table is due (nextRunAt <= now):
     - Enqueue one job to db-sync queue: { integrationId, orgId, triggeredBy: 'cron' }
  4. Update nextRunAt for enqueued tables

Why one job per integration, not one per table? Because we open one connection to the customer's DB per job. Opening a connection per table would be too expensive.

How to compute nextRunAt: Use cron-parser npm package:

import parser from 'cron-parser'
const interval = parser.parseExpression(cronExpression)
const nextRunAt = interval.next().toDate()

---

9. tRPC Router

// server/src/routers/db-integration-router.ts

dbIntegration.testConnection       → test without saving
dbIntegration.discoverTables       → get table list (used during setup wizard)
dbIntegration.create               → save a new DB integration + table mappings
dbIntegration.list                 → get all integrations for an org
dbIntegration.get                  → get one integration with its table mappings
dbIntegration.update               → update name, credentials, or table mappings
dbIntegration.delete               → soft delete (set isActive=false)
dbIntegration.syncNow              → manually trigger a sync run
dbIntegration.getSyncHistory       → get list of DbSyncRun for an integration
dbIntegration.getSyncRunDetail     → get one DbSyncRun with its DbSyncTableRun rows

All routes are protected (p — authenticated). Credentials are never returned in API responses — only metadata (host, dbType, name, lastTestedAt).

---

10. What the UI Looks Like (Screens to Build)

Screen 1 — Integration list page

Route: /$orgName/integrations/databases

Shows cards for each connected database:

• Name, DB type icon, last sync time, status badge (Active / Failed / Syncing)
• "Sync now" button
• "Manage" button

Screen 2 — Add database wizard (3 steps)

Step 1: "Enter connection details" — the form from the whiteboard (DB type, host, port, database, username, password, SSL)

Step 2: "Test & discover" — shows the status from the whiteboard:

○ Testing connection...         → spinner
✓ Connection successful (48ms)  → green
✓ Discovering tables...         → spinner
✓ Found 12 tables               → green

If connection fails, show the error message and let them fix credentials.

Step 3: "Select tables to sync" — a table with checkboxes:

□  Table Name      Columns      Cursor column       Schedule
✓  products        47 cols      updated_at          Every hour
✓  suppliers       12 cols      id                  Every 6 hours
□  order_history   8 cols       -                   -

Screen 3 — Sync run history

Route: /$orgName/integrations/databases/:id/history

Shows a list of sync runs with:

• Start time, duration, status badge
• "3/5 tables synced" summary
• Click to expand → shows per-table results (which ones succeeded, which failed, row counts, error messages)

This is the "Job Started → N tables succeed → Remaining ones" from the whiteboard.

---

11. Failure Handling — What Happens When Things Go Wrong

Scenario A — Customer DB is down

• Connection attempt times out after 10 seconds
• DbSyncRun marked FAILED, error: "Connection timed out"
• No data is written to Coyax DB
• Next scheduled sync will try again automatically

Scenario B — One table fails (e.g. permission denied)

• Other tables continue syncing normally
• DbSyncTableRun for that table is marked FAILED with the error
• DbSyncRun is marked PARTIAL
• User sees notification: "Sync partially completed: 4/5 tables synced. products table failed: permission denied"

Scenario C — Network drops mid-sync

• The batch that was in progress is lost
• lastCursor is only updated after a successful batch write
• So on the next run, we pick up from the last successful cursor position
• No data is lost. Some rows may be written twice (upsert handles this safely)

Scenario D — Target table schema mismatch

• If a column exists in source but not in target: skip that column, log a warning
• If a column type changed: attempt the insert, catch cast error, mark that row as failed, continue
• This means some rows may not fully sync — we log it in DbSyncTableRun.rowsFailed

Scenario E — Credentials changed (password rotated)

• Next sync fails with auth error
• User gets notification: "Sync failed: authentication error. Update your credentials."
• User goes to manage screen, updates credentials, re-tests, saves
• Next sync resumes normally

---

12. What NOT to Build in v1

Keep it simple. Do NOT include these in v1:

What	Why not yet
Write-back to customer DB	Too risky. Read-only first, earn trust.
Real-time CDC (Change Data Capture)	Requires Debezium or AWS DMS — too heavy. Polling is fine for v1.
SAP HANA and Oracle adapters	Require native client libraries. Complex to set up. Defer.
Schema auto-migration	If the source table adds a column, we do not auto-add it in Coyax. Manual for now.
Multi-region routing	Not needed at this stage.
Column-level transformations	Map 1:1 for now. Transformations (rename, cast, formula) are v2.

---

13. New Files to Create

server/src/
  integrations/
    database/
      base/
        db-adapter.ts            ← interface
        db-registry.ts           ← adapter registry
      adapters/
        postgres-adapter.ts      ← BUILD FIRST
        mysql-adapter.ts         ← BUILD SECOND
        mssql-adapter.ts         ← phase 2
  services/
    queues/
      db-sync-queue.ts           ← second BullMQ queue
    workers/
      db-sync-worker.ts          ← sync job processor
    db-sync-scheduler.ts         ← cron loop that enqueues due jobs
  routers/
    db-integration-router.ts     ← tRPC router
  utils/
    crypto.ts                    ← AES-256-GCM encrypt/decrypt for credentials

client/src/routes/$orgName/integrations/
  databases/
    index.tsx                    ← list page
    new.tsx                      ← add wizard
    $dbId/
      index.tsx                  ← manage + sync history

Existing files to modify:

• server/schema.zmodel — add new models above
• server/src/index.ts — register dbIntegrationRouter + start db-sync-scheduler
• server/src/integrations/index.ts — register DB adapters at startup

---

14. Build Order

Phase 1 — Core (build this first, 1–2 weeks)

1. Add models to schema.zmodel, run npx zenstack generate && npx prisma migrate dev
2. Write crypto.ts (encrypt/decrypt)
3. Write postgres-adapter.ts (test connection + discover tables + fetch rows)
4. Write db-integration-router.ts (testConnection, discoverTables, create, list, get)
5. Build the add-wizard UI (3 steps: form → test → table picker)
6. Test end-to-end: connect a real Postgres DB, verify table list shows up

Definition of done for Phase 1: A user can connect a PostgreSQL database, see the available tables, and the connection is saved with encrypted credentials.

Phase 2 — Sync engine (1–2 weeks)

1. Write db-sync-queue.ts
2. Write db-sync-worker.ts (the full sync loop from section 6.3)
3. Write db-sync-scheduler.ts (cron trigger)
4. Start the scheduler in server/src/index.ts
5. Add syncNow, getSyncHistory, getSyncRunDetail to the router
6. Build the sync history UI (section 10, Screen 3)

Definition of done for Phase 2: A user can manually trigger a sync, see the run history, see which tables succeeded and which failed, and the data appears in their Coyax tables.

Phase 3 — Reliability and more DBs (ongoing)

1. Add MySQL adapter
2. Add SQL Server adapter
3. Add incremental sync (cursor-based, replace full sync)
4. Add failure notifications (use existing Notification model)
5. Add "re-sync failed tables only" button
6. Add SQL Server adapter

---

15. How This Fits Into the Existing Architecture

                          ┌─────────────────────────────────────┐
                          │           EXISTING SYSTEM            │
                          │                                      │
                          │   Gmail / Slack Integration          │
                          │   → BullMQ (message-processing)      │
                          │   → message-worker                   │
                          │   → WorkflowEngine                   │
                          └─────────────────────────────────────┘

                          ┌─────────────────────────────────────┐
                          │           NEW: DB INTEGRATION        │
                          │                                      │
  Customer's DB ──────────┤   DbIntegration (credentials)       │
  (Postgres, MySQL,       │   DbTableMapping (what + when)       │
   SAP, Oracle)           │   → BullMQ (db-sync) NEW QUEUE      │
                          │   → db-sync-worker NEW WORKER        │
                          │   → DbSyncRun + DbSyncTableRun       │
                          │                     ↓                │
                          │   Org's Neon DB (Sequelize)         │
                          │   (same place workflows write to)    │
                          └─────────────────────────────────────┘

Once data lands in the Org's Neon DB, the existing workflow engine can immediately use it. A workflow triggered by "when a record is created" will fire as soon as a sync run writes new rows. No extra wiring needed.

---

16. Environment Variables to Add

# Encryption key for DB credentials (generate once with: openssl rand -hex 32)
DB_CREDENTIAL_ENCRYPTION_KEY=your_32_byte_hex_key_here

# Redis is already configured for BullMQ — no new Redis env needed
# The db-sync queue uses the same Redis instance

---

Last updated: 2026-04-12 | Author: Piyush | Status: Ready for team review