Stage 1: GTM data architecture and reverse ETL

The short answer

Go-to-market (GTM) is everything a company does to find, win and keep customers: marketing, sales, onboarding, renewals. A GTM engineer builds the data and automation systems behind that work. Stage 1 is the foundation. You pull billing, product usage and CRM data into one warehouse, define each metric once in SQL, and push the results back into the tools sellers use, such as the CRM (customer relationship management system, the database where sellers log deals). Pushing results back is called reverse ETL. Every later stage breaks if these tables are wrong, so you own them.

Why this matters

Picture a renewal call. The rep sees $4,900 a month in the CRM. Finance sees $3,200 in Stripe, the payments and billing service. Each tool computes MRR (monthly recurring revenue) from its own rules, so both are "right" and nobody can say which number is true.

The roadmap behind this course says: "GTM without a single, code-level source of truth is just guessing." Test that claim.

• What holds up. Billing lives in Stripe, usage in your product database, deals in the CRM. Only a warehouse can join all three. Without one you copy numbers by hand or write scripts that disagree.
• What is overstated. Many teams ran on the CRM alone for years, so "guessing" is too strong. We found no controlled study that teams with a warehouse out-sell teams without one. The case is engineering logic, not a measured lift.
• The nuance. "Single" works per field. The warehouse owns derived metrics such as MRR and usage. The CRM keeps owning what reps type in, such as deal stage. Pick one owner per field and write it down.

We found no GTM-specific study of what bad revenue data costs, so this course quotes none. The evidence on demand for the skill comes from job postings:

• Bloomberry analyzed 1,000 GTM Engineering job postings (updated January 25, 2026). Postings rose 205% in 2025 versus 2024 (January to September compared). SQL and Python each appeared in 38% of postings, HubSpot in 52%, Salesforce in 45%. The median posted salary was $127,500.

Core concepts

GTM and the GTM Engineer

A GTM team spans marketing (create demand), sales (close it), customer success (keep and grow it) and RevOps (revenue operations, the people who run the systems). A GTM engineer is the technical builder in that group. In Bloomberry's postings the top duties were building GTM workflows, integrating the tech stack, and owning the CRM. Bloomberry also found that 9 of 10 responsibilities overlap with RevOps jobs. The difference in practice is that GTM engineers write code and call APIs.

Warehouse vs CRM: who owns which truth

A CRM (HubSpot, Salesforce) is built for people to type into. A warehouse (Snowflake, BigQuery, DuckDB) is built for machines to join large tables.

Raw facts keep their original home. Derived facts get computed in the warehouse. Reverse ETL writes them into the CRM as read-only fields.

ELT and the modern data stack

ETL (extract, transform, load) cleans data before storing it. ELT loads raw data first, then transforms it inside the warehouse with SQL. Raw copies let you fix a formula and rebuild without calling the API again. The "modern data stack" is this pattern: a loader, a warehouse, a SQL transformation tool and a reverse ETL tool.

• Fivetran is a managed loader. Its Free plan lists 500,000 monthly active rows (MAR) for connections.
• Airbyte has a free open-source edition (Airbyte Core) and a paid cloud.
• dlt is an open-source Python library with a Stripe source (dlt init stripe_analytics duckdb). Plain Python is easy to debug.
• dbt turns SQL select statements, called models, into warehouse tables with tests and version control.

Two October 2026 updates. Fivetran and dbt Labs merged on June 1, 2026 and operate as Fivetran + dbt Labs. dbt's versions page lists dbt v2.0 (released September 14, 2026) and v1.12 (July 16, 2026), both supported. The DuckDB adapter is built into dbt v2.

Modeling: one definition of MRR

A model is a query that gives a business term one meaning. The project uses this rule:

active and past_due count. trialing and canceled do not. Yearly prices divide by 12, and a price billed every 3 months divides by 3. The model ignores discounts, tax and usage-based fees. That is a business choice, and finance may choose differently. What matters is that the choice lives in one SQL file everyone can read. Store money as integer cents, because float division gives values like 15833.33 and rounding errors add up.

Reverse ETL and sync modes

Reverse ETL copies modeled rows from the warehouse into operational tools. A sync connects one query to one destination object and runs on a schedule. Between runs, tools such as Hightouch compare the current result with the previous run and send only added, changed or removed rows. You will build that diff yourself.

Worked example. The model returns cus_mock001, Acme, mrr_cents 19900, and the sync creates one HubSpot record. The next run sends nothing. If Acme upgrades to 39,800 cents, only that row is sent:

{"idProperty": "stripe_customer_id", "id": "cus_mock001", "properties": {"customer_name": "Acme", "mrr_cents": "39800"}}

In this course, an upsert means update the record if its key exists, otherwise create it.

Names differ by vendor. Hightouch calls its overwrite-everything mode "All (Mirror mode)" and says archive is destructive. Mirror is dangerous: if an upstream job breaks and the model returns zero rows, a mirror sync can empty the CRM. Refuse to run when the row count drops sharply.

Vendor status, October 2026. Recheck before you rely on it.

• Fivetran announced on May 1, 2025 that it would acquire Census. Its docs now call the product Fivetran Activations (formerly Census). Month-to-month Census plans ended in March 2026, the Census Store was set to sunset on August 1, 2026, and pricing uses Monthly Active Rows.
• Hightouch raised a $150 million Series D at a $2.75 billion valuation on April 29, 2026. Its pricing page leads with an "Agentic Marketing Platform" and a "Composable CDP". A "Basic Reverse ETL" tier is free with up to 2 active syncs.

Identity resolution

Identity resolution decides which records, across systems, are the same person or company. Stripe knows cus_123 and an email. The CRM knows a contact and a company. Without a match you cannot connect revenue to a lead.

• Deterministic matching uses exact keys such as the same email or company domain. It is fast and clear.
• Probabilistic matching scores partial agreement across fields. The standard model is Fellegi-Sunter. If 95% of true matches agree on last name and 1% of random pairs agree, agreement multiplies the odds of a match by 95 (about 6.6 bits). Disagreement divides them by about 20. Splink implements it on DuckDB and is MIT licensed.
• Domain matching is the B2B workhorse. Take the text after @. Exclude free mail such as gmail.com, or every Gmail user becomes one company. Expect subsidiaries and shared domains.
• Account-to-lead matching attaches a person to their company's account, so a new lead from [email protected] routes to the Acme owner.

-- Deterministic account-to-lead match: same company domain, free mail excluded.
create or replace table raw_crm_contacts as
select * from (values
  ('[email protected]', 'Ana'), ('[email protected]', 'Bo'), ('[email protected]', 'Cy'),
  ('[email protected]', 'Dee'), ('[email protected]', 'Eli')
) t(email, first_name);

select
    c.email,
    a.stripe_customer_id,
    case when a.stripe_customer_id is null then 'no account' else 'domain match' end as match_type
from raw_crm_contacts c
left join account_billing a
       on a.company_domain = lower(split_part(c.email, '@', 2))
order by c.email;

Start deterministic. Add probabilistic matching only when you can measure what the simple rule misses.

Webhooks: signatures, idempotency, retries

A webhook is an HTTP request a vendor sends to your server when something happens. Polling (asking repeatedly) wastes calls and adds delay. Webhooks add new risks, and Stripe's webhook docs answer each one:

• Anyone can POST to your URL. Verify the Stripe-Signature header. It holds a timestamp (t=) and signatures (v1=): an HMAC (a hash that only a holder of the secret can compute) with SHA-256 over the timestamp, a dot and the raw body, keyed by your endpoint secret (whsec_...). Official libraries do this, but they need the raw body, so do not parse the JSON first.
• Requests can be replayed. Libraries reject timestamps older than 5 minutes by default. Never set tolerance to 0, which disables the check.
• Events repeat. Store each event id and skip ids you have seen.
• Events arrive out of order. Stripe does not guarantee order, and created has one-second resolution. Fetch the current object from the API when order matters.
• Your server can be down. Return a 2xx status before slow work. In live mode Stripe retries for up to three days with exponential backoff. In a sandbox it retries three times over a few hours. You can resend an event for 15 days from the Dashboard, or 30 with the CLI.

Stripe's docs now recommend thin events (small payloads you expand through the API) for new integrations. This stage uses snapshot events, which carry the full object and keep the code short. Choose Snapshot when you create a Dashboard event destination.

Idempotency means doing something twice has the same effect as once. Stripe accepts an Idempotency-Key header on POST requests, saves the first response, and returns it for repeats. Keys can be pruned after 24 hours (docs). Upserts are idempotent when they match on a unique key.

CRM custom objects: HubSpot and Salesforce

A custom object is a table you define in the CRM. Here, a "billing account" holds MRR and usage.

HubSpot. Custom objects need an Enterprise subscription (schemas guide). Create the schema with POST /crm-object-schemas/2026-09/schemas. HubSpot uses date-based API versions, and its docs list 2026-09 as current and 2026-03 as previous. Mark one property hasUniqueValue to get an external key. Write records with POST /crm/objects/2026-09/{objectTypeId}/batch/upsert, sending id and idProperty per input, up to 100 inputs per call (guide). A service key is the simplest script credential. Send it as Authorization: Bearer .... Select the scopes crm.schemas.custom.write, crm.objects.custom.read and crm.objects.custom.write when you create it. HubSpot's service key page does not list these scopes, so confirm the names in the key form.

Salesforce. A custom object's API name ends in __c. Mark a custom field as External ID and unique, then upsert with PATCH /services/data/vXX.0/sobjects/Billing_Account__c/Stripe_Customer_Id__c/cus_123 (REST docs). It returns 201 on create, 200 on update (API version 46.0 and later, 204 before that), and 300 if the key matches several records. GET /services/data/ lists supported versions. The Salesforce docs site blocks automated fetches, so this paragraph was checked against Salesforce's REST API Developer Guide PDF (version 68.0). For practice, Salesforce offers a free Developer Edition org. Its signup page blocked our check, so confirm it is still open.

API rate limits

Every API caps your speed. Design for the cap before the first sync.

Example: sync 50,000 rows into a HubSpot Free/Starter account. One call per row is 50,000 calls, 20% of the 250,000 daily cap. In batches of 100 it is 500 calls, 0.2% of the cap, and at least 50 seconds at 100 calls per 10 seconds. On a 429, wait for Retry-After if present, then retry with jittered backoff.

Key terms

The tools

Build the project

You will build a small pipeline end to end. Raw Stripe subscriptions and usage events land in DuckDB, one SQL model builds account_billing, and a reverse ETL script upserts only changed rows into a HubSpot custom object. A webhook receiver shows how real-time events enter the same system. The no-budget path needs no accounts: a mock seed script, plus DRY_RUN=1 to print HubSpot payloads without calling HubSpot.

1. Make a folder, create a Python virtual environment, and install stripe, duckdb, requests and flask.
2. Pick a data source. With no accounts, run seed_mock.py. For real data, create a free Stripe account, add a few test customers and subscriptions in its sandbox (a test copy of Stripe with fake money) using test card 4242 4242 4242 4242, then run extract_stripe.py.
3. Save models/account_billing.sql and run it against gtm.duckdb.
4. Run the three check queries. Fix the model before you sync anything.
5. Create a HubSpot developer test account and a service key with custom object scopes. Run create_hubspot_schema.py and save the printed object type ID.
6. Run reverse_etl.py with DRY_RUN=1 and read the payloads. Then run it for real.
7. Open the Billing accounts object in HubSpot and confirm the MRR values match your SQL.
8. Run the sync again and confirm it reports 0 changed rows. Then, on mock data, run update raw_stripe_subscriptions set payload = json_merge_patch(payload, '{"status": "canceled"}') where id = 'sub_mock001' in DuckDB (with real data, cancel a sandbox subscription and rerun extract_stripe.py), rerun the model and the sync, and confirm exactly 1 row is sent.
9. Install the Stripe CLI, run stripe login, then run stripe listen --forward-to localhost:5000/stripe/webhook and copy the whsec_ secret it prints.
10. In a second terminal, export that secret as STRIPE_WEBHOOK_SECRET and start the receiver with flask --app webhook_app run.
11. In a third terminal run stripe trigger payment_intent.succeeded, then confirm the stripe listen terminal shows a 200 response and inbox.sqlite holds the event.
12. Save and run test_webhook.py. It signs requests the way Stripe does and must print good 200, dup 200, bad 400, old 400, nohd 400, then one stored row.

Install and set secrets in environment variables:

python -m venv venv && source venv/bin/activate
pip install stripe duckdb requests flask
export STRIPE_SECRET_KEY=sk_test_...        # sandbox key only
export STRIPE_WEBHOOK_SECRET=whsec_...      # printed by stripe listen (step 9)
export HUBSPOT_TOKEN=...                    # HubSpot service key
export HUBSPOT_OBJECT_TYPE_ID=2-...         # printed by create_hubspot_schema.py

Mock data:

"""Create fake raw Stripe and product-usage tables in DuckDB. No accounts needed."""
import json
import random
from datetime import datetime, timedelta, timezone

import duckdb

random.seed(7)
now = datetime.now(timezone.utc)
con = duckdb.connect("gtm.duckdb")
con.execute("create or replace table raw_stripe_customers (id varchar, payload json)")
con.execute("create or replace table raw_stripe_subscriptions (id varchar, payload json)")
con.execute("create or replace table raw_usage_events (user_email varchar, event varchar, occurred_at timestamp)")

domains = ["acme.io", "globex.com", "initech.co", "umbrella.dev", "gmail.com", "stark.ai"]
statuses = ["active", "active", "active", "past_due", "trialing", "canceled"]
plans = [("month", 4900), ("month", 19900), ("year", 190000)]

for i, (domain, status) in enumerate(zip(domains, statuses), 1):
    cid = f"cus_mock{i:03d}"
    customer = {"id": cid, "email": f"billing@{domain}", "name": domain.split(".")[0].title()}
    interval, amount = random.choice(plans)
    item = {"quantity": random.choice([1, 3]), "price": {"unit_amount": amount, "recurring": {"interval": interval}}}
    sub = {"id": f"sub_mock{i:03d}", "customer": cid, "status": status, "items": {"data": [item]}}
    con.execute("insert into raw_stripe_customers values (?, ?)", [cid, json.dumps(customer)])
    con.execute("insert into raw_stripe_subscriptions values (?, ?)", [sub["id"], json.dumps(sub)])
    for user in range(random.randint(1, 5)):
        for _ in range(random.randint(1, 8)):
            when = now - timedelta(days=random.randint(0, 60), hours=random.randint(0, 23))
            con.execute("insert into raw_usage_events values (?, ?, ?)",
                        [f"user{user}@{domain}", random.choice(["login", "export", "invite"]), when.replace(tzinfo=None)])
print("seeded", len(domains), "customers")

Real Stripe data, loaded raw:

"""Load raw Stripe customers and subscriptions into DuckDB (the E and L of ELT)."""
import json
import os

import duckdb
from stripe import StripeClient

client = StripeClient(os.environ["STRIPE_SECRET_KEY"])  # use a sandbox key: sk_test_...
con = duckdb.connect("gtm.duckdb")


def load(table, rows):
    con.execute(f"create or replace table {table} (id varchar, payload json)")
    n = 0
    for obj in rows:
        con.execute(f"insert into {table} values (?, ?)", [obj.id, json.dumps(obj.to_dict())])
        n += 1
    print(table, n, "rows")


load("raw_stripe_customers", client.v1.customers.list(params={"limit": 100}).auto_paging_iter())
load("raw_stripe_subscriptions",
     client.v1.subscriptions.list(params={"limit": 100, "status": "all"}).auto_paging_iter())

The model. In Stripe's current API the period end sits on each subscription item, not on the subscription, so raw payloads keep items.data nested and the SQL unnests it:

/* One row per Stripe customer. This file is the single definition of MRR. */
create or replace table account_billing as
with free_mail(domain) as (
    values ('gmail.com'), ('yahoo.com'), ('outlook.com'), ('hotmail.com'), ('icloud.com'), ('proton.me')
),
customers as (
    select
        payload->>'$.id'                          as stripe_customer_id,
        lower(payload->>'$.email')                as email,
        payload->>'$.name'                        as customer_name,
        lower(split_part(payload->>'$.email', '@', 2)) as email_domain
    from raw_stripe_customers
),
sub_items as (
    select
        payload->>'$.customer' as stripe_customer_id,
        payload->>'$.status'   as status,
        item
    from raw_stripe_subscriptions,
         unnest(cast(payload->'$.items.data' as json[])) as t(item)
),
mrr as (
    select
        stripe_customer_id,
        /* Rule: active and past_due count toward MRR. trialing and canceled do not. */
        cast(round(sum(case when status in ('active', 'past_due') then
              cast(item->>'$.price.unit_amount' as bigint)
              * coalesce(cast(item->>'$.quantity' as bigint), 1)
              / (case item->>'$.price.recurring.interval'
                     when 'year' then 12 when 'month' then 1 else null end
                 * coalesce(cast(item->>'$.price.recurring.interval_count' as bigint), 1))
            else 0 end)) as bigint)               as mrr_cents,  /* week and day prices count as 0 here */
        max(status)                               as subscription_status  /* assumes one subscription per customer */
    from sub_items
    group by 1
),
usage as (
    select
        lower(split_part(user_email, '@', 2))     as email_domain,
        count(distinct user_email) filter (where occurred_at >= now() - interval 30 day) as active_users_30d,
        max(occurred_at)                          as last_event_at
    from raw_usage_events
    group by 1
)
select
    c.stripe_customer_id,
    c.customer_name,
    case when c.email_domain in (select domain from free_mail) then null else c.email_domain end as company_domain,
    coalesce(m.mrr_cents, 0)                      as mrr_cents,
    coalesce(m.subscription_status, 'none')       as subscription_status,
    coalesce(u.active_users_30d, 0)               as active_users_30d,
    strftime(u.last_event_at, '%Y-%m-%dT%H:%M:%SZ') as last_event_at
from customers c
left join mrr m using (stripe_customer_id)
left join usage u on u.email_domain = c.email_domain;

Checks. All three must return zero rows:

-- 1. The key must be unique
select stripe_customer_id, count(*) from account_billing group by 1 having count(*) > 1;
-- 2. MRR must never be negative or null
select * from account_billing where mrr_cents is null or mrr_cents < 0;
-- 3. Every active customer must have a positive MRR
select * from account_billing where subscription_status = 'active' and mrr_cents = 0;

Create the HubSpot object once:

"""Create the custom object 'billing_account' in HubSpot. Run once. Needs an Enterprise-trial test account."""
import os

import requests

BASE = os.environ.get("HUBSPOT_BASE_URL", "https://api.hubapi.com")
VERSION = "2026-09"
headers = {"Authorization": f"Bearer {os.environ['HUBSPOT_TOKEN']}"}

def text(name, label, **extra):
    return {"name": name, "label": label, "type": "string", "fieldType": "text", **extra}

def number(name, label):
    return {"name": name, "label": label, "type": "number", "fieldType": "number"}

schema = {
    "name": "billing_account",
    "labels": {"singular": "Billing account", "plural": "Billing accounts"},
    "primaryDisplayProperty": "customer_name",
    "requiredProperties": ["stripe_customer_id"],
    "searchableProperties": ["stripe_customer_id", "customer_name", "company_domain"],
    "associatedObjects": ["0-2"],  # 0-2 is the HubSpot company object
    "properties": [
        text("stripe_customer_id", "Stripe customer ID", hasUniqueValue=True),
        text("customer_name", "Customer name"),
        text("company_domain", "Company domain"),
        number("mrr_cents", "MRR (cents)"),
        text("subscription_status", "Subscription status"),
        number("active_users_30d", "Active users (30d)"),
        {"name": "last_event_at", "label": "Last product event", "type": "datetime", "fieldType": "date"},
    ],
}
r = requests.post(f"{BASE}/crm-object-schemas/{VERSION}/schemas", headers=headers, json=schema, timeout=30)
r.raise_for_status()
print("Set HUBSPOT_OBJECT_TYPE_ID to", r.json()["objectTypeId"])

The reverse ETL job hashes each row, sends only rows whose hash changed, batches 100 at a time, retries on 429, and records state only after a batch succeeds:

"""Reverse ETL: send changed warehouse rows to a HubSpot custom object (sync mode: upsert)."""
import hashlib
import json
import os
import random
import time

import duckdb
import requests

BASE = os.environ.get("HUBSPOT_BASE_URL", "https://api.hubapi.com")
VERSION = "2026-09"
DRY_RUN = os.environ.get("DRY_RUN") == "1"  # print the payloads instead of calling HubSpot
OBJECT = os.environ.get("HUBSPOT_OBJECT_TYPE_ID", "2-0")  # printed by create_hubspot_schema.py
HEADERS = {"Authorization": f"Bearer {os.environ.get('HUBSPOT_TOKEN', '')}"}
URL = f"{BASE}/crm/objects/{VERSION}/{OBJECT}/batch/upsert"
BATCH = 100  # HubSpot batch endpoints accept up to 100 inputs per request
KEY = "stripe_customer_id"

con = duckdb.connect("gtm.duckdb")
con.execute("create table if not exists sync_state (id varchar primary key, row_hash varchar)")


def changed_rows():
    """Rows whose hash differs from the last run: change data capture in a few lines."""
    cur = con.execute("select * from account_billing")
    cols = [c[0] for c in cur.description]
    values_list = cur.fetchall()  # read first: a second execute() on this connection resets the cursor
    sent = dict(con.execute("select id, row_hash from sync_state").fetchall())
    for values in values_list:
        row = dict(zip(cols, values))
        digest = hashlib.sha256(json.dumps(row, sort_keys=True, default=str).encode()).hexdigest()
        if sent.get(row[KEY]) != digest:
            yield row, digest


def post(body):
    if DRY_RUN:
        print(json.dumps(body["inputs"][0]), "... and", len(body["inputs"]) - 1, "more")
        return {"numErrors": 0}
    for attempt in range(6):
        r = requests.post(URL, headers=HEADERS, json=body, timeout=30)
        if r.status_code == 429:  # rate limited: wait, then retry the same batch
            wait = float(r.headers.get("Retry-After", 2 ** attempt)) + random.random()
            print(f"429, waiting {wait:.1f}s")
            time.sleep(wait)
            continue
        r.raise_for_status()
        return r.json()
    raise RuntimeError("still rate limited after 6 attempts")


def to_input(row):
    props = {k: str(v) for k, v in row.items() if v is not None}  # includes the key, so a create sets it
    return {"idProperty": KEY, "id": row[KEY], "properties": props}


def main():
    pending = list(changed_rows())
    print(len(pending), "changed rows")
    for i in range(0, len(pending), BATCH):
        chunk = pending[i:i + BATCH]
        result = post({"inputs": [to_input(row) for row, _ in chunk]})
        if result.get("numErrors") or result.get("errors"):
            # Do not mark this chunk as sent. The next run retries it, and upsert is safe to repeat.
            print("errors:", json.dumps(result.get("errors"))[:500])
            continue
        if DRY_RUN:
            continue  # a dry run never records state
        con.executemany(
            "insert or replace into sync_state values (?, ?)",
            [[row[KEY], digest] for row, digest in chunk],
        )
        print("synced", len(chunk), "rows")


if __name__ == "__main__":
    main()

The webhook receiver:

"""Receive Stripe webhooks: verify the signature, store the raw event once, answer 200 fast."""
import os
import sqlite3

import stripe
from flask import Flask, request

app = Flask(__name__)
SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]  # whsec_..., one per endpoint
db = sqlite3.connect(os.environ.get("INBOX_DB", "inbox.sqlite"), check_same_thread=False)
db.execute("""create table if not exists stripe_events (
    id text primary key, type text, created integer, payload text, processed integer default 0)""")


@app.post("/stripe/webhook")
def stripe_webhook():
    payload = request.get_data()  # raw bytes: do not use request.json, it changes the body
    try:
        event = stripe.Webhook.construct_event(payload, request.headers.get("Stripe-Signature"), SECRET)
    except (ValueError, stripe.SignatureVerificationError):
        return "invalid", 400
    # The primary key makes this idempotent: a duplicate delivery inserts nothing.
    db.execute("insert or ignore into stripe_events (id, type, created, payload) values (?, ?, ?, ?)",
               (event.id, event.type, event.created, payload.decode()))
    db.commit()
    return "", 200  # do the slow work later, from the inbox table

Its test, which signs requests with the same HMAC scheme Stripe documents:

import hashlib, hmac, json, os, time
os.environ["STRIPE_WEBHOOK_SECRET"] = "whsec_test123"
os.environ["INBOX_DB"] = ":memory:"  # keep test rows out of the real inbox
import webhook_app
c = webhook_app.app.test_client()
body = json.dumps({"id": "evt_1", "object": "event", "type": "invoice.paid", "created": 1, "data": {"object": {"id": "in_1"}}}).encode()
def sign(body, secret="whsec_test123", ts=None):
    ts = ts or int(time.time())
    sig = hmac.new(secret.encode(), f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
    return f"t={ts},v1={sig}"
print("good", c.post("/stripe/webhook", data=body, headers={"Stripe-Signature": sign(body)}).status_code)
print("dup", c.post("/stripe/webhook", data=body, headers={"Stripe-Signature": sign(body)}).status_code)
print("bad", c.post("/stripe/webhook", data=body, headers={"Stripe-Signature": sign(body, "whsec_wrong")}).status_code)
print("old", c.post("/stripe/webhook", data=body, headers={"Stripe-Signature": sign(body, ts=int(time.time()) - 3600)}).status_code)
print("nohd", c.post("/stripe/webhook", data=body).status_code)
print(webhook_app.db.execute("select count(*) from stripe_events").fetchone())

What done looks like. account_billing has one row per customer and passes the checks. HubSpot shows the same MRR as your SQL. A second run sends 0 rows, and one change sends 1. The test script prints the six expected lines: a duplicate event is stored once, and bad, stale and unsigned requests get 400.

Stretch goals. Add mirror mode: archive HubSpot records whose customer vanished, behind a row-count guard. Associate each billing account with its HubSpot company by domain. Move the SQL into a dbt model with a unique test.

Common mistakes

• Syncing raw tables. Send modeled rows, not copies of Stripe's tables. Otherwise sellers build their own formulas again.
• No unique key. An upsert needs a stable key such as the Stripe customer ID. Matching on email creates duplicates when someone changes it.
• Trusting webhook order and delivery. Events duplicate, arrive out of order and can fail for days. Verify the signature, store the event ID, and read current state from the API.
• Mirror syncs with no guard. An empty upstream table can delete real CRM records.
• Pasting secrets into code. Use environment variables, sandbox keys while you learn, and never commit a .env file.

Check yourself

How to learn this faster

Learn first: SQL joins and aggregates, then the Stripe object model (customer, subscription, invoice), then one CRM's API. Learn dbt after SQL feels easy.

Skip for now: Kafka, streaming, your own identity graph, and paid reverse ETL contracts.

Drill: Break your own pipeline on purpose. Send the same webhook twice. Send a bad signature. Empty the raw table and watch the sync. Write down each failure and its guard.

• Day 1: Set up the project, run the mock seed, build the model and pass the three checks.
• Day 2: Read the Stripe webhook and idempotency docs. Build the receiver and test the duplicate, bad-signature, stale and unsigned cases.
• Day 3: Create the HubSpot test account and schema. Run the sync with DRY_RUN=1, then for real.
• Day 4: Add a row-count guard that stops the sync when account_billing has under half the rows listed in sync_state. Change a row and watch only that row sync. Compute your daily call budget from the HubSpot limits page.
• Day 5: Run the identity SQL on sample contacts. Write a note on which system owns each field.

Resources

• Stripe docs: Receive events in your webhook endpointDocsFree

  Signatures, retries, duplicates and ordering from the source.

• Stripe docs: Idempotent requestsDocsFree

  The shortest correct explanation of idempotency keys.

• HubSpot docs: Custom object records API guideDocsFree

  Batch upsert with a unique property, with request bodies.

• HubSpot docs: API usage guidelinesDocsFree

  Rate limits per tier for private and public apps.

• HubSpot docs: Account types and developer test accountsDocsFree

  How to get a free test CRM with Enterprise features.

• Hightouch docs: Sync types and modesDocsFree

  Clear definitions of update, insert, upsert, archive and mirror.

• Fivetran Activations docs: SyncsDocsFree

  The Census lineage, with upsert, mirror and append behaviors.

• dlt documentationDocsFree

  The Python-first way to build an ELT pipeline into DuckDB.

• Bloomberry: 1,000 GTM Engineering jobs analyzedResearchFree

  What employers ask for: tools, skills and pay.

• Trailhead: 360 Platform API BasicsCourseFree

  A hands-on Salesforce REST API module of about 1.5 hours.

Next module

Stage 2 is Signal and Intent Pipelines. With trustworthy account data in your warehouse, you will feed it fresh signals such as web visits, hiring changes and product events, and turn them into scored, routed leads.