> ## Documentation Index
> Fetch the complete documentation index at: https://docs.octavehq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Snowflake

> Generate a deployable pipeline that loads GTM events into Octave from Snowflake

A query-based source: the extract is a `SELECT` with a watermark predicate, run on a schedule. The types are where the care goes — `NUMBER` and `TIMESTAMP_NTZ` both need deliberate handling on the way out.

## What you need

* A role with `SELECT` on the source tables, and a warehouse to run on
* An Octave workspace API key, from **Settings → Integrations**
* Somewhere to run a scheduled job — AWS, GCP or Azure

## Where this runs

There is no object-landing event to hook, so this runs on a timer. A scheduled function that queries Snowflake and posts the results is the usual shape; keeping it inside Snowflake as a Task is possible but buys less than it costs.

|                              |                                                         |
| ---------------------------- | ------------------------------------------------------- |
| **Runtime**                  | AWS Lambda or Cloud Run job, Python 3.12                |
| **Trigger**                  | EventBridge or Cloud Scheduler, hourly                  |
| **Secrets**                  | Platform secret manager — key-pair auth, not a password |
| **Watermark**                | Last `modified_at` processed, in DynamoDB or Firestore  |
| **Initial backfill**         | Same job, run over a date range with a longer timeout   |
| **In-warehouse alternative** | Snowflake Task plus an external access integration      |

## The prompt

Copy this into Claude Code, Cursor or any coding agent. It carries the whole flow —
reading your warehouse, mapping the columns, posting to Octave, and deploying the result.

Fill in section 1 with your real schema and a few sample rows (`DESCRIBE TABLE analytics.gtm.opportunities` plus `SELECT * … LIMIT 5`). The more of
your actual data it can see, the less it has to guess.

```text theme={null}
I want to load GTM history from Snowflake into Octave, and keep it current after that.
Start by working out the column mapping. Do not write any code until the mapping is
agreed.

## 1. Here is my data

Warehouse: Snowflake, database `analytics`, schema `gtm`
Tables and what they hold:
  opportunities  -> deals
  email_log      -> outbound sends and replies
Incremental column: modified_at (TIMESTAMP_NTZ)

<PASTE YOUR SCHEMA AND 3-5 SAMPLE ROWS HERE, ONE BLOCK PER TABLE>

## 2. Where this needs to run

Runtime: AWS Lambda or a Cloud Run job, Python 3.12 - pick one and say why
Trigger: EventBridge or Cloud Scheduler, hourly
Secrets: Snowflake key-pair credentials and the Octave API key, both in the
platform's secret manager
Watermark: the last modified_at processed, in DynamoDB or Firestore
Constraint: Lambda caps at 15 minutes; a Cloud Run job does not. Say which limit
applies to your choice and how the initial multi-year backfill fits inside it.
If you think a Snowflake Task with an external access integration is the better
answer, argue for it before writing anything.

## 3. Here is what Octave needs

Four event types. Every event carries a "type" discriminator.

CRM event - a deal:
{
  "type": "crm",
  "eventTimestamp": "2024-02-03T16:00:00.000Z",
  "eventType": "opportunity_created",
  "opportunityId": "opp_12345",
  "opportunityName": "Acme Corp - Enterprise Plan",
  "amount": "50000.00",
  "currency": "USD",
  "stage": "Qualification",
  "stageCategory": "open",
  "accountName": "Acme Corp",
  "crmAccountId": "acct_67890",
  "crmAccountDomain": "acme.com",
  "contactEmail": "john@acme.com",
  "contactEmails": ["john@acme.com", "priya@acme.com"],
  "ownerEmail": "sarah@company.com",
  "closeDate": "2024-03-31T00:00:00.000Z",
  "crmLastModifiedAt": "2024-02-03T16:00:00.000Z",
  "eventId": "<stable id derived from my row>"
}

Email event:
{
  "type": "email",
  "eventTimestamp": "2024-02-03T10:30:00.000Z",
  "eventType": "sent",
  "subject": "Follow up on our conversation",
  "body": { "text": "Hi John, just wanted to follow up..." },
  "from": { "email": "sarah@company.com", "name": "Sarah Johnson" },
  "to": [{ "email": "john@acme.com", "name": "John Smith" }],
  "crmOpportunityId": "opp_12345",
  "eventId": "<stable id derived from my row>"
}

Call event:
{
  "type": "call",
  "eventTimestamp": "2024-02-03T14:00:00.000Z",
  "eventType": "transcript",
  "title": "Discovery Call - Acme Corp",
  "transcript": [
    { "speaker": { "name": "Sarah Johnson", "email": "sarah@company.com", "role": "internal" },
      "text": "Thanks for making the time. What pushed you to look at this now?" },
    { "speaker": { "name": "John Smith", "email": "john@acme.com", "role": "external" },
      "text": "Our current process breaks down past about fifty reps." }
  ],
  "participants": [
    { "name": "Sarah Johnson", "email": "sarah@company.com", "role": "internal" },
    { "name": "John Smith", "email": "john@acme.com", "role": "external" }
  ],
  "crmOpportunityId": "opp_12345",
  "eventId": "<stable id derived from my row>"
}

Social event:
{
  "type": "social",
  "eventTimestamp": "2024-02-03T11:00:00.000Z",
  "eventType": "message_sent",
  "body": { "text": "Thanks for connecting, John." },
  "from": { "email": "sarah@company.com", "name": "Sarah Johnson" },
  "to": [{ "email": "john@acme.com", "name": "John Smith" }],
  "eventId": "<stable id derived from my row>"
}

Required fields - everything else is optional:
  email:  type, eventTimestamp, eventType, subject, body, from, to
  call:   type, eventTimestamp, eventType, title, transcript, participants
  crm:    type, eventTimestamp, eventType, opportunityId, opportunityName
  social: type, eventTimestamp, eventType, body, from, to

eventType must be one of:
  email:  sent | reply | opened | clicked | bounced | unsubscribed
  call:   transcript | scheduled | completed | missed
  crm:    opportunity_created | deal_won | deal_lost | meeting_booked
  social: connection_sent | connection_accepted | message_sent | message_received

Only email sent/reply and call transcript are analysed. All CRM and all social
types are analysed. Other email and call types are stored but produce nothing.

## 4. Map my columns onto those structures

Apply these rules:

- eventTimestamp accepts ISO 8601, epoch milliseconds (>= 1e12) or epoch seconds.
  Pass my column straight through if it is already one of those. It must resolve
  within 50 years past and 5 years future - flag sentinel dates (9999-12-31,
  1900-01-01, epoch zero) as rows to drop rather than rows to send.
- Combine separate name and email columns into { "email": ..., "name": ... }.
  Only email is required. Split comma-separated recipient strings into an array.
- Send "amount" as a STRING, not a number, so my decimal precision survives.
- Derive CRM "eventType" from my stage or status column, and set "stageCategory"
  to open, won or lost explicitly rather than letting Octave infer it.
- ONE SOURCE ROW MAY BE SEVERAL EVENTS. A deal row with both a created date and a
  closed date is an opportunity_created event AND a deal_won or deal_lost event,
  each with its own eventTimestamp and its own eventId. Do not collapse them.
- Always send "crmLastModifiedAt" on CRM events. Without it an updated deal can
  hash identical to the previous push and be dropped as a duplicate.
- Prefer a structured transcript array over a flat string. Decide speaker role by
  email domain: my company's domain is "internal", everything else is "external".
- Set "eventId" (<= 255 chars) from the row's stable primary key. Where one row
  expands into several events, suffix it per event so each one stays distinct.
- Populate "contactEmails" with every contact on the deal, not just the primary -
  it is the biggest single lever on how much activity attaches to the opportunity.
  Same for "crmAccountId", "crmAccountDomain" and "lossReason" where I have them.
- On email, call and social rows that already reference a deal, set
  "crmOpportunityId" - it skips Octave's participant-matching inference.
- Omit keys whose source value is NULL. Do not send explicit nulls.

Show me the proposed mapping as a table - my column, Octave field, transformation.
List every Octave field you could not source, and every column of mine you ignored.
Wait for me to confirm the mapping before you write any code.

## 5. Then build and deploy it

Target: POST https://app.octavehq.com/api/v2/event/import
Header: api_key: <my Octave workspace key>
Body:   { "events": [ ... ] }     # max 1000 per request, mixed types allowed

Response: { "jobs": [ { "eventType", "jobOId", "accepted", "preSkipped",
                        "duplicatesInBatch" } ],
            "totalAccepted", "totalPreSkipped" }

One job is created per event type present in the request. Poll each one:
GET https://app.octavehq.com/api/v2/event/import/status?jobOId=<jobOId>
every 5-10 seconds until status is COMPLETED (or FAILED / CANCELLED).
counts.ingested is the authoritative success count. counts.skipped covers
duplicates and non-processable event types and is not an error - alert on
counts.failed instead.

A 400 means NOTHING in that request was imported; the message lists the offending
zero-based indices. Retry 429 and 5xx with exponential backoff, and keep only a
few requests in flight.

Build this for the runtime I named in section 2, and give me:
- the handler code,
- infrastructure-as-code to deploy it - tell me which format you are using and why,
- the watermark held in a managed store rather than on local disk, so a cold start
  resumes where the previous run finished,
- credentials read from the platform's secret manager, never baked into the code,
- structured logs carrying ingested / skipped / failed per batch, and an alert
  condition worth paging on,
- a note on what happens if a run hits the runtime's execution limit, and what to
  change if my history is too large to backfill in a single invocation.

Keep the column mapping in one dict at the top of the handler so I can correct it
without touching the transport or the deployment code.
```

<Tip>
  The instruction to show the mapping as a table and wait for confirmation is the part worth
  keeping. Without it an agent guesses at your column names and buries the guess inside a
  handler, where a wrong `eventType` looks exactly like a right one until the data is in.
</Tip>

## Hints worth adding

These are the things that go wrong with Snowflake specifically. Paste whichever apply into
section 1 of the prompt — an agent cannot infer them from a schema.

**`NUMBER(18,2)` becomes a float somewhere on the way to JSON.** Ask for
`TO_VARCHAR(amount)` in the query and `amount` sent as a string, or deal values arrive
subtly wrong and nobody notices for a quarter.

**`TIMESTAMP_NTZ` carries no timezone.** Whatever the pipeline does with it is an
assumption. Say which zone the column is really in and ask for an explicit
`CONVERT_TIMEZONE('UTC', …)` at the boundary.

**An unbounded backfill query keeps the warehouse busy for the whole run.** Ask for the
history to be walked in date ranges rather than issued as one query — it is the difference
between a predictable bill and a surprising one.