Snowflake

If your conversations already land in a Snowflake 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 Snowflake role that can create a role and a user, and grant access to the table that holds your conversations.
• A key pair for a Snowflake service user.
• A Dimension Labs API key for a new Universal API source. See Your API key.

1. Get access

Create a service user that signs in with a key pair. Snowflake describes a TYPE = SERVICE user as one that “interacts with Snowflake without human intervention” (CREATE USER).

1. Generate a private key and its public key, as shown in Key-pair authentication:

   openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt
   openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

2. Create a read-only role and the service user, and give the user the public key. Paste the contents of rsa_key.pub without the BEGIN and END lines.

   CREATE ROLE dl_sync_role;
   GRANT USAGE ON WAREHOUSE YOUR_WAREHOUSE TO ROLE dl_sync_role;
   GRANT USAGE ON DATABASE support TO ROLE dl_sync_role;
   GRANT USAGE ON SCHEMA support.public TO ROLE dl_sync_role;
   GRANT SELECT ON TABLE support.public.conversation_messages TO ROLE dl_sync_role;
   
   CREATE USER dl_sync TYPE = SERVICE DEFAULT_ROLE = dl_sync_role;
   GRANT ROLE dl_sync_role TO USER dl_sync;
   ALTER USER dl_sync SET RSA_PUBLIC_KEY = 'MIIBIjANBgkqh...';

   Setting a user’s key needs the MODIFY PROGRAMMATIC AUTHENTICATION METHODS privilege on that user, or ownership of it. Snowflake also offers ALTER USER ... ADD KEY PAIR, which it recommends over SET RSA_PUBLIC_KEY (Key-pair authentication).

3. Copy your account identifier. In Snowsight, open the account selector, view the account details, and copy the identifier. The preferred format is orgname-account_name (Account identifiers).

Keep the private key in a secrets manager, never in code you commit.

2. Query new rows

Use the official Node.js driver, snowflake-sdk, with key pair authentication: set authenticator to SNOWFLAKE_JWT and pass the PEM key as privateKey (Authenticating connections).

Each run asks for the rows changed since the last run, using updated_at as the cursor. DATE_PART(epoch_millisecond, ...) returns Unix milliseconds for a timestamp (DATE_PART), so the cursor and the message time are both plain numbers.

SELECT message_id, conversation_id, customer_id, sender_type, body, channel,
       DATE_PART(epoch_millisecond, sent_at)    AS sent_ms,
       DATE_PART(epoch_millisecond, updated_at) AS updated_ms
FROM support.public.conversation_messages
WHERE DATE_PART(epoch_millisecond, updated_at) >= ?
ORDER BY updated_at;

The ? is a bind variable, filled from binds when you call connection.execute (Executing statements). Snowflake returns unquoted column names in upper case, so the rows arrive as MESSAGE_ID, SENT_MS and so on.

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

Dimension Labs	Snowflake 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. If your only customer key is an email, send a consistent hash of it instead.

4. Send it

Install the driver in your project first: npm install snowflake-sdk.

// snowflake.mjs. Environment: SNOWFLAKE_ACCOUNT (orgname-account_name), SNOWFLAKE_PRIVATE_KEY (PEM text of rsa_key.p8), DL_API_KEY.
import snowflake from 'snowflake-sdk';
import { sendToDimensionLabs, toMillis } from './dimensionlabs.mjs';

const HOUR = 60 * 60 * 1000;
const SQL = `SELECT message_id, conversation_id, customer_id, sender_type, body, channel,
  DATE_PART(epoch_millisecond, sent_at) AS sent_ms, DATE_PART(epoch_millisecond, updated_at) AS updated_ms
  FROM support.public.conversation_messages
  WHERE DATE_PART(epoch_millisecond, updated_at) >= ? 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 syncSnowflake({ since, historical = false }) {
  const connection = snowflake.createConnection({
    account: process.env.SNOWFLAKE_ACCOUNT,
    username: 'DL_SYNC',
    authenticator: 'SNOWFLAKE_JWT',
    privateKey: process.env.SNOWFLAKE_PRIVATE_KEY,
    role: 'DL_SYNC_ROLE', warehouse: 'YOUR_WAREHOUSE',
  });
  await new Promise((ok, fail) => connection.connect((err) => (err ? fail(err) : ok())));
  try {
    const rows = await new Promise((ok, fail) => connection.execute({
      sqlText: SQL, binds: [since], complete: (err, stmt, rows) => (err ? fail(err) : ok(rows)),
    }));
    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) : 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 new Promise((ok) => connection.destroy(() => ok()));
  }
}

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.

5. Run it

Save the code as snowflake.mjs in a project set up as in Set up your project, and run it with the scheduled sync runner: import { syncSnowflake as sync } from './snowflake.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.

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

To keep the job inside Snowflake instead, Snowflake documents two pieces you would combine: tasks can run stored procedures on a SCHEDULE (Introduction to tasks), and an external access integration lets UDF and stored procedure handlers reach network locations outside Snowflake (External network access overview). A new task starts suspended, so run ALTER TASK ... RESUME after you create it.

Write Dimensions back

To bring Dimensions into Snowflake, 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.

Load the files with PUT into an internal stage and COPY INTO your table (Bulk loading from a local file system):

CREATE STAGE IF NOT EXISTS dl_exports;
PUT file://dimensions/* @dl_exports OVERWRITE = TRUE;
COPY INTO analytics.dimension_labs.dimensions FROM @dl_exports
  FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1); -- match the format of 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.

Snowflake reference

• CREATE USER
• Key-pair authentication and key-pair rotation
• Account identifiers
• Node.js driver: Managing connections
• Node.js driver: Authenticating connections
• Node.js driver: Options reference
• Node.js driver: Executing statements
• DATE_PART
• Introduction to tasks
• External network access overview
• Bulk loading from a local file system

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