# Snowflake > Send conversation messages stored in Snowflake to Dimension Labs with the universal API. 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](/api/tracker). ## 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](/api/tracker#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](https://docs.snowflake.com/en/sql-reference/sql/create-user)). 1. Generate a private key and its public key, as shown in [Key-pair authentication](https://docs.snowflake.com/en/user-guide/key-pair-auth): ```bash 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. ```sql 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](https://docs.snowflake.com/en/user-guide/key-pair-auth)). 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](https://docs.snowflake.com/en/user-guide/admin-account-identifier)). 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](https://docs.snowflake.com/en/developer-guide/node-js/nodejs-driver-authenticate)). 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](https://docs.snowflake.com/sql-reference/functions/date_part)), so the cursor and the message time are both plain numbers. ```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; ``` The `?` is a bind variable, filled from `binds` when you call `connection.execute` ([Executing statements](https://docs.snowflake.com/en/developer-guide/node-js/nodejs-driver-execute)). 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`. ```js // snowflake.mjs. Environment: SNOWFLAKE_ACCOUNT (orgname-account_name), SNOWFLAKE_PRIVATE_KEY (PEM text of rsa_key.p8), DL_API_KEY. 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](/api/tracker#set-up-your-project), and run it with the [scheduled sync](/api/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](/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. You can also export the rows to a file and [upload it](/data/upload-a-file). 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](https://docs.snowflake.com/en/user-guide/tasks-intro)), and an external access integration lets UDF and stored procedure handlers reach network locations outside Snowflake ([External network access overview](https://docs.snowflake.com/en/developer-guide/external-network-access/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](/api/export) 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. ```js // export.mjs. Environment: DL_EXPORT_KEY, DL_INTEGRATION_ID. Usage: node export.mjs 2026-10-05 2026-10-13 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](https://docs.snowflake.com/en/user-guide/data-load-bulk)): ```sql 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](https://docs.snowflake.com/en/sql-reference/sql/create-user) - [Key-pair authentication and key-pair rotation](https://docs.snowflake.com/en/user-guide/key-pair-auth) - [Account identifiers](https://docs.snowflake.com/en/user-guide/admin-account-identifier) - [Node.js driver: Managing connections](https://docs.snowflake.com/en/developer-guide/node-js/nodejs-driver-connect) - [Node.js driver: Authenticating connections](https://docs.snowflake.com/en/developer-guide/node-js/nodejs-driver-authenticate) - [Node.js driver: Options reference](https://docs.snowflake.com/developer-guide/node-js/nodejs-driver-options) - [Node.js driver: Executing statements](https://docs.snowflake.com/en/developer-guide/node-js/nodejs-driver-execute) - [DATE_PART](https://docs.snowflake.com/sql-reference/functions/date_part) - [Introduction to tasks](https://docs.snowflake.com/en/user-guide/tasks-intro) - [External network access overview](https://docs.snowflake.com/en/developer-guide/external-network-access/external-network-access-overview) - [Bulk loading from a local file system](https://docs.snowflake.com/en/user-guide/data-load-bulk) Checked against Snowflake's documentation on 2026-10-13.