PostgreSQL

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.

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.

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). The privileges come from GRANT (GRANT):

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). 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:

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). 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).
• 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). The code sets the session to UTC first with SET TIME ZONE 'UTC' (SET), 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.

4. Send it

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

// postgresql.mjs. Environment: PG_URL (postgresql://dl_sync:PASSWORD@HOST:5432/DBNAME), DL_API_KEY.
import pg from 'pg';
import { sendToDimensionLabs, toMillis } from './dimensionlabs.mjs';

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).

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, and run it with the 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 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. Run it through the file checker first.

PostgreSQL reference

• CREATE ROLE
• GRANT
• Date/time functions and operators
• Date/time types
• SET
• LIMIT and OFFSET
• node-postgres: Connecting
• node-postgres: Queries
• node-postgres: Data types
• node-postgres: SSL

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