# Google BigQuery

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

If your conversations already land in a BigQuery 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 Google Cloud role that can create service accounts and grant IAM roles on the project and dataset.
- A service account for the job.
- A Dimension Labs API key for a new **Universal API** source. See [Your API key](/api/tracker#your-api-key).

## 1. Get access

1. Create a service account: open **Create service account** in the Google Cloud console, pick the project, enter a name, and click **Create and continue** ([Create service accounts](https://docs.cloud.google.com/iam/docs/service-accounts-create)).
2. Grant it two roles:
   - **BigQuery Job User** (`roles/bigquery.jobUser`) on the project, which "provides permissions to run jobs, including queries, within the project". It can only be granted at the project, folder or organization level.
   - **BigQuery Data Viewer** (`roles/bigquery.dataViewer`) on the dataset that holds your conversations, which lets it read the tables.

   Both are described in [BigQuery IAM roles and permissions](https://docs.cloud.google.com/bigquery/docs/access-control#bigquery.dataViewer).
3. Run the job with this service account as its identity. On Cloud Run, the client library picks up the service identity through Application Default Credentials, and Google advises against setting `GOOGLE_APPLICATION_CREDENTIALS` on services that use a service identity ([Introduction to service identity](https://docs.cloud.google.com/run/docs/securing/service-identity)). For local testing, `gcloud auth application-default login` gives you credentials from your own account ([Get started with authentication](https://docs.cloud.google.com/bigquery/docs/authentication/getting-started)).

## 2. Query new rows

Use the official Node.js client, `@google-cloud/bigquery` ([Google BigQuery: Node.js Client](https://docs.cloud.google.com/nodejs/docs/reference/bigquery/latest)). Each run asks for the rows changed since the last run, using `updated_at` as the cursor. `UNIX_MILLIS` turns a TIMESTAMP into Unix milliseconds and `TIMESTAMP_MILLIS` does the reverse, so the cursor and the message time are both plain numbers:

```sql
SELECT message_id, conversation_id, customer_id, sender_type, body, channel,
       UNIX_MILLIS(sent_at) AS sent_ms, UNIX_MILLIS(updated_at) AS updated_ms
FROM `YOUR_PROJECT.support.conversation_messages`
WHERE updated_at >= TIMESTAMP_MILLIS(@since)
ORDER BY updated_at
```

`bigquery.query()` takes the SQL as `query` and the named values as `params` ([BigQuery class reference](https://docs.cloud.google.com/nodejs/docs/reference/bigquery/latest/bigquery/bigquery)).

## 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 (TIMESTAMP) |
| `updated_at` | When the row was written or last changed (TIMESTAMP) |

| Dimension Labs | BigQuery 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 client in your project first: `npm install @google-cloud/bigquery`.

```js
// bigquery.mjs. Credentials: Application Default Credentials (the job's service account). Environment: DL_API_KEY.



const HOUR = 60 * 60 * 1000;
const bigquery = new BigQuery();
const query = `SELECT message_id, conversation_id, customer_id, sender_type, body, channel,
  UNIX_MILLIS(sent_at) AS sent_ms, UNIX_MILLIS(updated_at) AS updated_ms
  FROM \`YOUR_PROJECT.support.conversation_messages\`
  WHERE updated_at >= TIMESTAMP_MILLIS(@since) 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 syncBigQuery({ since, historical = false }) {
  const [rows] = await bigquery.query({ query, params: { since: Math.floor(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) : 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;
}
```

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 `bigquery.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 { syncBigQuery as sync } from './bigquery.mjs';`. On Google Cloud, run `run.mjs` as a Cloud Run job on a Cloud Scheduler trigger, as the scheduled sync page describes.

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

## Write Dimensions back

To bring Dimensions into BigQuery, 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 a BigQuery load job, from a local file or from Cloud Storage ([Batch load data](https://docs.cloud.google.com/bigquery/docs/batch-loading-data)), for example with `bq load` and the source format that matches 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.

## BigQuery reference

- [Create service accounts](https://docs.cloud.google.com/iam/docs/service-accounts-create)
- [BigQuery IAM roles and permissions](https://docs.cloud.google.com/bigquery/docs/access-control#bigquery.dataViewer)
- [Basic roles and permissions](https://docs.cloud.google.com/bigquery/docs/access-control-basic-roles)
- [Get started with authentication](https://docs.cloud.google.com/bigquery/docs/authentication/getting-started)
- [Introduction to service identity (Cloud Run)](https://docs.cloud.google.com/run/docs/securing/service-identity)
- [Google BigQuery: Node.js Client](https://docs.cloud.google.com/nodejs/docs/reference/bigquery/latest)
- [BigQuery class reference](https://docs.cloud.google.com/nodejs/docs/reference/bigquery/latest/bigquery/bigquery)
- [Batch load data](https://docs.cloud.google.com/bigquery/docs/batch-loading-data)

Checked against Google Cloud's documentation on 2026-10-13.