Amazon Redshift

If your conversations already land in a Redshift 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

• An AWS role that can create IAM roles and policies, and a Redshift user that can grant SELECT on the table that holds your conversations.
• An IAM role for the job, such as a Lambda execution role.
• A Dimension Labs API key for a new Universal API source. See Your API key.

1. Get access

This guide uses the Redshift Data API, which runs SQL over HTTPS through the AWS SDK, without a database driver or a persistent connection. It authenticates with credentials from AWS Secrets Manager or with temporary credentials derived from the caller’s IAM identity (Using the Amazon Redshift Data API).

1. Give the job’s IAM role a Data API policy. AWS suggests starting from the AmazonRedshiftDataFullAccess managed policy (Configuring IAM permissions).

2. Pick how it signs in to the database, per the ExecuteStatement reference (ExecuteStatement):

   • Redshift Serverless, temporary credentials: pass WorkgroupName and Database. The role needs redshift-serverless:GetCredentials. The database user is mapped from the IAM identity: a role named dl-sync-role becomes IAMR:dl-sync-role.
   • Provisioned cluster, temporary credentials: pass ClusterIdentifier and Database to sign in as the IAM identity (needs redshift:GetClusterCredentialsWithIAM), or add DbUser to sign in as a database user (needs redshift:GetClusterCredentials).
   • Secrets Manager: pass SecretArn and Database, plus WorkgroupName or ClusterIdentifier. The secret holds a username and password.

3. Grant that database user read access to the table:

   GRANT SELECT ON support.conversation_messages TO "IAMR:dl-sync-role";

2. Query new rows

Use the AWS SDK for JavaScript v3 client, @aws-sdk/client-redshift-data (client README). Each run asks for the rows changed since the last run, using updated_at as the cursor. DATE_PART(epoch, ...) returns seconds since 1970-01-01 00:00:00, independent of the cluster’s time zone, as a double (Date parts for date or timestamp functions).

SELECT message_id, conversation_id, customer_id, sender_type, body, channel,
       DATE_PART(epoch, sent_at)    AS sent_s,
       DATE_PART(epoch, updated_at) AS updated_s
FROM support.conversation_messages
WHERE DATE_PART(epoch, updated_at) * 1000 >= :since
ORDER BY updated_at;

• :since is a named parameter, in Unix milliseconds. Data API parameter values are sent as strings, and can’t be empty or NULL.
• The Data API is asynchronous. ExecuteStatement returns an Id. Poll DescribeStatement until its Status is FINISHED (or FAILED or ABORTED), then GetStatementResult returns the rows as Records. Pass the returned NextToken back until it is empty (DescribeStatement, GetStatementResult).
• Each value in a record is a Field holding one of stringValue, longValue, doubleValue, booleanValue, blobValue or isNull (Field).
• AWS’s documented limits include a 500 MB maximum query result size (after gzip compression) and results kept for up to 24 hours (Using the Amazon Redshift Data API).

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, in UTC
updated_at	When the row was written or last changed, in UTC

Dimension Labs	Redshift 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_s	Seconds. toMillis turns them into milliseconds.
platformJson	message_id, channel	Add any join keys you have, such as an account or store id. If your only customer key is an email, send a consistent hash of it instead.

4. Send it

Install the client in your project first: npm install @aws-sdk/client-redshift-data.

// redshift.mjs. Environment: REDSHIFT_WORKGROUP, REDSHIFT_DATABASE, DL_API_KEY (and AWS credentials from the job's role).
import { RedshiftDataClient, ExecuteStatementCommand, DescribeStatementCommand, GetStatementResultCommand } from '@aws-sdk/client-redshift-data';
import { sendToDimensionLabs, toMillis } from './dimensionlabs.mjs';

const HOUR = 60 * 60 * 1000;
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const client = new RedshiftDataClient({});
const SQL = `SELECT message_id, conversation_id, customer_id, sender_type, body, channel,
  DATE_PART(epoch, sent_at) AS sent_s, DATE_PART(epoch, updated_at) AS updated_s
  FROM support.conversation_messages
  WHERE DATE_PART(epoch, updated_at) * 1000 >= :since ORDER BY updated_at`;
const val = (f) => (f.isNull ? null : f.stringValue ?? f.longValue ?? f.doubleValue ?? f.booleanValue);
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 syncRedshift({ since, historical = false }) {
  const { Id } = await client.send(new ExecuteStatementCommand({
    WorkgroupName: process.env.REDSHIFT_WORKGROUP, Database: process.env.REDSHIFT_DATABASE, Sql: SQL,
    Parameters: [{ name: 'since', value: String(Math.floor(since)) }],
  }));
  for (let wait = 500; ; wait = Math.min(wait * 2, 5000)) {
    const { Status, Error: error } = await client.send(new DescribeStatementCommand({ Id }));
    if (Status === 'FINISHED') break;
    if (Status === 'FAILED' || Status === 'ABORTED') throw new Error(`Redshift query ${Status}: ${error}`);
    await sleep(wait);
  }
  let newest = since;
  let NextToken;
  do {
    const page = await client.send(new GetStatementResultCommand({ Id, NextToken }));
    for (const record of page.Records ?? []) {
      const [messageId, conversationId, customerId, senderType, body, channel, sentS, updatedS] = record.map(val);
      const text = clean(body);
      const id = String(conversationId).replaceAll(':', '-');
      if (text) {
        const ts = toMillis(Number(sentS));
        await sendToDimensionLabs({
          text,
          userId: customerId ? String(customerId) : id,
          incoming: senderType === 'customer',
          sessionId: id,
          conversationId: id,
          dimensionlabs_timestamp: ts,
          platformJson: { message_id: String(messageId), channel },
        }, { historical: historical || ts < Date.now() - HOUR });
      }
      newest = Math.max(newest, toMillis(Number(updatedS)));
    }
    NextToken = page.NextToken;
  } while (NextToken);
  return newest;
}

This example uses Redshift Serverless (WorkgroupName). For a provisioned cluster, pass ClusterIdentifier instead, as in step 1. The cursor follows updated_at, so a row that changes later is sent again; duplicates are removed automatically.

5. Run it

Save the code as redshift.mjs in a project set up as in Set up your project, and run it with the scheduled sync runner: import { syncRedshift as sync } from './redshift.mjs';. On Lambda, add the Data API permissions from step 1 to the function’s execution role.

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

You can also export the rows to a file and upload it.

Write Dimensions back

To bring Dimensions into Redshift, pull them with the Export API and load the files into a table. You need an Export API key, from Settings in the app (not the tracker key), and the integration ID of the source.

// export.mjs. Environment: DL_EXPORT_KEY, DL_INTEGRATION_ID. Usage: node export.mjs 2026-10-05 2026-10-13
import { writeFile } from 'node:fs/promises';

const [startDate, endDate] = process.argv.slice(2); // endDate is exclusive
const params = new URLSearchParams({ startDate, endDate, botIds: process.env.DL_INTEGRATION_ID });
const res = await fetch(`https://api.dimensionlabs.io/export?${params}`, { headers: { Authorization: process.env.DL_EXPORT_KEY } });
if (!res.ok) throw new Error(`Export returned ${res.status}: ${await res.text()}`);
const { zipFile } = await res.json();
// zipFile is either a download link or the zip itself as base64 text; this handles both.
const zip = /^https?:\/\//.test(zipFile) ? Buffer.from(await (await fetch(zipFile)).arrayBuffer()) : Buffer.from(zipFile, 'base64');
await writeFile('dimensions.zip', zip);

Then unzip -o dimensions.zip -d dimensions/. The zip holds the enrichments by day. Open one day’s file to see its format and columns, and create your table to match.

Upload the files to Amazon S3 and load them into your table with COPY, authorized with an IAM_ROLE (Using the COPY command to load from Amazon S3), with the format options that match the export files.

• endDate is exclusive, and dates are partitioned by US Pacific day. Overlap each export with the last by a day, so nothing falls between them.
• The overlap returns the same rows again. Replace each exported day’s rows in your table rather than appending them.

Amazon Redshift reference

• Using the Amazon Redshift Data API
• Configuring IAM permissions for the Data API
• ExecuteStatement
• DescribeStatement
• GetStatementResult
• Field
• Date parts for date or timestamp functions
• @aws-sdk/client-redshift-data README
• Using the COPY command to load from Amazon S3

Checked against AWS’s documentation on 2026-10-13.