# PostgreSQL

> Send conversation messages stored in PostgreSQL to Dimension Labs with the universal API.

If your conversations already land in a PostgreSQL table, you can send them to Dimension Labs from there. A scheduled job on your side queries the rows that are new since its last run and posts each message to the [universal API](/api/tracker).

## What you need

- A PostgreSQL role that can create a role and grant access to the table that holds your conversations.
- The database's host, port and name, and network access to it from wherever the job runs. Managed hosts often need the job's IP address on an allowlist.
- A Dimension Labs API key for a new **Universal API** source. See [Your API key](/api/tracker#your-api-key).

## 1. Get access

Create a role for the job that can sign in and only read the conversations table. `CREATE USER` is the same as `CREATE ROLE` with `LOGIN` ([CREATE ROLE](https://www.postgresql.org/docs/current/sql-createrole.html)). The privileges come from `GRANT` ([GRANT](https://www.postgresql.org/docs/current/sql-grant.html)):

```sql
CREATE ROLE dl_sync WITH LOGIN PASSWORD 'choose-a-long-password';
GRANT USAGE ON SCHEMA support TO dl_sync;
GRANT SELECT ON support.conversation_messages TO dl_sync;
```

Build a connection string from it, in the form node-postgres documents: `postgresql://dl_sync:PASSWORD@HOST:5432/DBNAME` ([Connecting](https://node-postgres.com/features/connecting)). Keep it in a secrets manager, never in code you commit.

## 2. Query new rows

Use node-postgres, the `pg` package. Each run asks for the rows changed since the last run, using `updated_at` as the cursor:

```sql
SELECT message_id, conversation_id, customer_id, sender_type, body, channel,
       floor(EXTRACT(EPOCH FROM sent_at) * 1000)::bigint    AS sent_ms,
       floor(EXTRACT(EPOCH FROM updated_at) * 1000)::bigint AS updated_ms
FROM support.conversation_messages
WHERE updated_at >= to_timestamp($1::double precision / 1000)
ORDER BY updated_at;
```

- `$1` is a query parameter, passed as the second argument to `client.query` ([Queries](https://node-postgres.com/features/queries)). It holds the cursor in Unix milliseconds.
- `to_timestamp` turns Unix seconds into a `timestamp with time zone`, and `EXTRACT(EPOCH ...)` turns a timestamp back into seconds ([Date/time functions](https://www.postgresql.org/docs/current/functions-datetime.html)).
- For a `timestamp without time zone` column, `EXTRACT(EPOCH ...)` counts its date and time as if they were UTC, and comparisons with a `timestamp with time zone` use the session's time zone ([Date/time types](https://www.postgresql.org/docs/current/datatype-datetime.html)). The code sets the session to UTC first with `SET TIME ZONE 'UTC'` ([SET](https://www.postgresql.org/docs/current/sql-set.html)), so both column types work, as long as plain timestamps hold UTC.
- node-postgres returns `bigint` values as strings, so the code turns them into numbers.

## 3. Map the columns

This guide assumes a table with one row per message, with these columns. Rename them to match yours.

| Column | Holds |
|---|---|
| `message_id` | The message's own id |
| `conversation_id` | The conversation, ticket or chat the message belongs to |
| `customer_id` | Your internal id for the customer in that conversation, on every row, including agent and bot rows. Not an email address or phone number. |
| `sender_type` | Who wrote the message, for example `customer`, `agent` or `bot` |
| `body` | The message text |
| `channel` | Chat, email, voice and so on |
| `sent_at` | When the message was sent (`timestamptz`, or `timestamp` in UTC) |
| `updated_at` | When the row was written or last changed (same type) |

| Dimension Labs | PostgreSQL column | Notes |
|---|---|---|
| `text` | `body` | Skip rows where it is empty. Strip HTML if bodies are HTML. |
| `userId` | `customer_id` | If it is null, use the `sessionId`. |
| `incoming` | `sender_type` | `true` only when the customer wrote it. Agent and bot rows are `false`. |
| `sessionId` | `conversation_id` | Replace any colons. |
| `conversationId` | `conversation_id` | Same value as `sessionId`. |
| `dimensionlabs_timestamp` | `sent_ms` | Already milliseconds. Pass it through `toMillis`. |
| `platformJson` | `message_id`, `channel` | Add any join keys you have, such as an account or store id. |

**Personal data.** Don't select email, phone or name columns. If your only customer key is an email, send a consistent hash of it instead. Message text can still hold personal details: redact them in the function before you send. See [PII redaction](/data/pii-redaction).

## 4. Send it

Install the driver in your project first: `npm install pg`.

```js
// postgresql.mjs. Environment: PG_URL (postgresql://dl_sync:PASSWORD@HOST:5432/DBNAME), DL_API_KEY.



const HOUR = 60 * 60 * 1000;
const SQL = `SELECT message_id, conversation_id, customer_id, sender_type, body, channel,
  floor(EXTRACT(EPOCH FROM sent_at) * 1000)::bigint AS sent_ms,
  floor(EXTRACT(EPOCH FROM updated_at) * 1000)::bigint AS updated_ms
  FROM support.conversation_messages
  WHERE updated_at >= to_timestamp($1::double precision / 1000) ORDER BY updated_at`;
const clean = (s) => String(s ?? '').replace(/<[^>]+>/g, ' ').replace(/\s+/g, ' ').trim();

// Sends every row written or changed at or after `since` (Unix ms). Returns the new `since`.
export async function syncPostgres({ since, historical = false }) {
  const client = new pg.Client({ connectionString: process.env.PG_URL });
  await client.connect();
  try {
    await client.query(`SET TIME ZONE 'UTC'`);
    const { rows } = await client.query(SQL, [since]);
    let newest = since;
    for (const row of rows) {
      const text = clean(row.body);
      const id = String(row.conversation_id).replaceAll(':', '-');
      if (text) {
        const ts = toMillis(Number(row.sent_ms));
        await sendToDimensionLabs({
          text,
          userId: row.customer_id ? String(row.customer_id).replaceAll(':', '-') : id,
          incoming: row.sender_type === 'customer',
          sessionId: id,
          conversationId: id,
          dimensionlabs_timestamp: ts,
          platformJson: { message_id: String(row.message_id), channel: row.channel },
        }, { historical: historical || ts < Date.now() - HOUR });
      }
      newest = Math.max(newest, Number(row.updated_ms));
    }
    return newest;
  } finally {
    await client.end();
  }
}
```

If your host requires TLS, add an `ssl` object to the `pg.Client` options; node-postgres passes it to Node's TLS socket ([SSL](https://node-postgres.com/features/ssl)).

The cursor follows `updated_at`, so a row that changes later (a status column, say) is sent again. That's fine: duplicates are removed automatically. An index on `updated_at` keeps the query fast as the table grows.

## 5. Run it

Save the code as `postgresql.mjs` in a project set up as in [Set up your project](/api/tracker#set-up-your-project), and run it with the [scheduled sync](/api/scheduled-sync) runner: `import { syncPostgres as sync } from './postgresql.mjs';`.

- **History first:** `node run.mjs 2025-01-01` sends every row since that date to `/trackhistorical`. Then start a [backfill](/dimensions/backfill) from the **Backfill** tab.
- **Then every 15 minutes:** the scheduled job sends new rows to `/track`, and they are enriched automatically every day. A message whose `sent_at` is more than an hour old when the job sees it (a late-loaded batch, say) goes to `/trackhistorical`, and needs a backfill.
- **Re-runs are safe:** duplicates are removed automatically.
- **No API rate limit:** the database is the only limit. The query reads all new rows into memory at once. For a very large history, start from a recent date and upload the older rows as a file.

You can also export the rows to a CSV file and [upload it](/data/upload-a-file). Run it through the [file checker](/data/check-a-file) first.

## PostgreSQL reference

- [CREATE ROLE](https://www.postgresql.org/docs/current/sql-createrole.html)
- [GRANT](https://www.postgresql.org/docs/current/sql-grant.html)
- [Date/time functions and operators](https://www.postgresql.org/docs/current/functions-datetime.html)
- [Date/time types](https://www.postgresql.org/docs/current/datatype-datetime.html)
- [SET](https://www.postgresql.org/docs/current/sql-set.html)
- [LIMIT and OFFSET](https://www.postgresql.org/docs/current/queries-limit.html)
- [node-postgres: Connecting](https://node-postgres.com/features/connecting)
- [node-postgres: Queries](https://node-postgres.com/features/queries)
- [node-postgres: Data types](https://node-postgres.com/features/types)
- [node-postgres: SSL](https://node-postgres.com/features/ssl)

Checked against PostgreSQL's and node-postgres's documentation on 2026-10-26.