# Sync your book

Keep a copy of your BrokerPlus records in your own database or warehouse. This guide walks through
the code: the first full copy, the incremental sync after it, and the records that drop out of a
list. For when to use the change feed, `updated_since`, or webhooks, read [Sync](/sync) first.

Give the sync its own key with the `read_only` limit. A sync never needs to write. See
[Authentication](/authentication).

## Sync with the change feed

The change feed is the better path. It never skips a change, and it tells you when a record is
deleted, merged, or archived. `GET /v1/changes` covers contacts, leads, deals, tasks, intakes,
documents, appointments, and communications.

1. Call `GET /v1/changes?after=now` and save the `next_checkpoint`.
2. Copy every record from the lists, for example `GET /v1/contacts`, page by page.
3. Call `GET /v1/changes?after=<checkpoint>`, apply each entry, and save the new
   `next_checkpoint`.
4. While `has_more` is `true`, repeat step 3 right away. When it is `false`, wait a while, then
   repeat.

Take the checkpoint in step 1 **before** the copy in step 2. Anything that changes during the copy
is then waiting for you in the feed.

To apply an entry, read the record again by `record_id` and upsert it. Three actions need a little
more:

- `deleted`: mark the record as removed in your copy.
- `archived`: read the record again. It left the active book but you can still read it, and an
  archived lead stays in the lead list.
- `merged`: point your references at `merged_into_id`, then remove the record.

<CodeTabs syncKey="lang">

```bash title="curl"
curl "https://api.mos.getbrokerplus.ca/v1/changes?after=MS4xMjAwLjE3OTE...&types=contact,deal" \
  -H "Authorization: Bearer $BROKERPLUS_API_KEY"
```

```ts title="TypeScript"
type Change = {
  id: string;
  record_type: string;
  record_id: string;
  action: "created" | "updated" | "deleted" | "merged" | "archived";
  changed_fields: string[];
  merged_into_id: string | null;
  occurred_at: string;
};

const API = "https://api.mos.getbrokerplus.ca/v1";
const headers = { Authorization: `Bearer ${process.env.BROKERPLUS_API_KEY}` };

// Reads the feed from a saved checkpoint until it is empty. Returns the checkpoint to save.
export async function readChanges(checkpoint: string, apply: (change: Change) => Promise<void>) {
  for (;;) {
    const url = new URL(`${API}/changes`);
    url.search = new URLSearchParams({ after: checkpoint, types: "contact,deal", limit: "100" }).toString();
    const res = await fetch(url, { headers });
    if (res.status === 410) throw new Error("Checkpoint expired. Copy the lists again from a new checkpoint.");
    if (!res.ok) throw new Error(`changes: ${res.status} ${await res.text()}`);

    const { data, pagination } = await res.json();
    for (const change of data as Change[]) await apply(change);
    checkpoint = pagination.next_checkpoint;
    if (!pagination.has_more) return checkpoint;
  }
}
```

```python title="Python"
import os

import requests

API = "https://api.mos.getbrokerplus.ca/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['BROKERPLUS_API_KEY']}"}


def read_changes(checkpoint, apply):
    """Reads the feed from a saved checkpoint until it is empty. Returns the checkpoint to save."""
    while True:
        params = {"after": checkpoint, "types": "contact,deal", "limit": 100}
        res = requests.get(f"{API}/changes", params=params, headers=HEADERS, timeout=60)
        if res.status_code == 410:
            raise RuntimeError("Checkpoint expired. Copy the lists again from a new checkpoint.")
        res.raise_for_status()

        body = res.json()
        for change in body["data"]:
            apply(change)
        checkpoint = body["pagination"]["next_checkpoint"]
        if not body["pagination"]["has_more"]:
            return checkpoint
```

</CodeTabs>

```json
{
  "success": true,
  "data": [
    {
      "id": "1201",
      "record_type": "contact",
      "record_id": "cmsctz1fa...",
      "action": "updated",
      "changed_fields": ["email", "updated_at"],
      "merged_into_id": null,
      "occurred_at": "2026-10-07T14:20:00.000Z"
    }
  ],
  "pagination": { "next_checkpoint": "MS4xMjAyLjE3OTE...", "has_more": false }
}
```

Things that trip people up:

- A checkpoint belongs to the `types` you sent with it. Keep one checkpoint per `types` value.
  Sending it with more types returns `400 INVALID_CHECKPOINT`.
- A checkpoint older than 30 days returns `410 CHECKPOINT_EXPIRED`. Start again from step 1.
- A page can hold fewer entries than `limit` and still have `has_more: true`. Decide on
  `has_more`, never on the count.
- Entries arrive in the order the changes became visible, which can differ from `occurred_at`.
- The feed covers the record's own fields. Addresses, notes, the mortgage application, and tags
  aren't in it yet, so read those from their own endpoints on a schedule.

## Sync with `updated_since`

Use this loop for a list the change feed doesn't cover, or for a one-off copy. Most record lists
take `updated_since` and `after_id`; the [API reference](/reference) shows which. Small lists such
as users, teams, pipelines, and lenders don't need it: read them in full each run.

With `updated_since`, rows come back oldest `updated_at` first, with `id` breaking ties. So the
last row you saved is your position in the list. Send it back to get what comes next:

| Parameter       | Value                                       |
| --------------- | ------------------------------------------- |
| `updated_since` | The `updated_at` of the last row you saved. |
| `after_id`      | The `id` of the last row you saved.         |

Keep `page` at `1` and stop when `data` is empty. For each list:

1. Load your saved cursor (an `updated_at` and an `id`). On the first run there is none: send
   `updated_since=1970-01-01T00:00:00Z` and no `after_id`.
2. Request the list with `page_size=100`.
3. Upsert each row into your database by `id`.
4. Save the last row's `updated_at` and `id` as the new cursor.
5. If `data` wasn't empty, go back to step 2.

<CodeTabs syncKey="lang">

```bash title="curl"
curl "https://api.mos.getbrokerplus.ca/v1/contacts?updated_since=2026-10-01T14:03:22.118Z&after_id=cmcontact8f2...&page_size=100" \
  -H "Authorization: Bearer $BROKERPLUS_API_KEY"
```

```ts title="TypeScript"
type Cursor = { updatedAt: string; id: string | null };
type Row = { id: string; updated_at: string };

const API = "https://api.mos.getbrokerplus.ca/v1";

export async function syncList(
  list: string,
  cursor: Cursor,
  upsert: (rows: Row[]) => Promise<void>,
  saveCursor: (cursor: Cursor) => Promise<void>,
) {
  for (;;) {
    const url = new URL(`${API}/${list}`);
    url.searchParams.set("updated_since", cursor.updatedAt);
    if (cursor.id) url.searchParams.set("after_id", cursor.id);
    url.searchParams.set("page_size", "100");

    const res = await fetch(url, {
      headers: { Authorization: `Bearer ${process.env.BROKERPLUS_API_KEY}` },
    });
    if (res.status === 429) {
      await new Promise((r) => setTimeout(r, Number(res.headers.get("Retry-After") ?? 60) * 1000));
      continue;
    }
    if (!res.ok) throw new Error(`${list}: ${res.status} ${await res.text()}`);

    const { data } = (await res.json()) as { data: Row[] };
    if (data.length === 0) return;

    await upsert(data);
    const last = data[data.length - 1];
    cursor = { updatedAt: last.updated_at, id: last.id };
    await saveCursor(cursor);
  }
}

// First run: await syncList("contacts", { updatedAt: "1970-01-01T00:00:00Z", id: null }, upsert, save);
```

```python title="Python"
import os
import time

import requests

API = "https://api.mos.getbrokerplus.ca/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['BROKERPLUS_API_KEY']}"}


def sync_list(name, cursor, upsert, save_cursor):
    """cursor is {"updated_at": str, "id": str | None}."""
    while True:
        params = {"updated_since": cursor["updated_at"], "page_size": 100}
        if cursor["id"]:
            params["after_id"] = cursor["id"]
        res = requests.get(f"{API}/{name}", params=params, headers=HEADERS, timeout=60)
        if res.status_code == 429:
            time.sleep(int(res.headers.get("Retry-After", "60")))
            continue
        res.raise_for_status()

        data = res.json()["data"]
        if not data:
            return
        upsert(data)
        last = data[-1]
        cursor = {"updated_at": last["updated_at"], "id": last["id"]}
        save_cursor(cursor)


# First run: sync_list("contacts", {"updated_at": "1970-01-01T00:00:00Z", "id": None}, upsert, save)
```

</CodeTabs>

Save the cursor **after** you write the rows. If the job dies in between, the next run reads a few
rows twice, which an upsert makes harmless. A cursor saved too early skips rows.

Don't sync with `page=2`, `page=3`, and so on. A row that changes mid-sync jumps to the end of the
list, every row behind it shifts up one place, and you miss one. The cursor doesn't move when other
rows change. [Sync](/sync) has the details.

## Catch records that leave a list

A list quietly drops a record that was deleted, archived, or merged, and the `updated_since` loop
never sees it again. Use the change feed where it covers the record. Otherwise, run a reconcile job
once a day or once a week:

1. Read the full list, with no `updated_since`, and collect every `id`.
2. Compare that set with the ids in your copy.
3. For each id only you have, call `GET /v1/{list}/{id}`. A `404` means the record is gone from
   your key's reach, so mark it removed.

A `404` doesn't always mean deleted. A record that moved to a team outside your key's reach also
returns `404`. And a merged or archived contact still answers `GET /v1/contacts/{id}`, so trust the
list, not the single read, to decide whether a contact is still in the book. [Sync](/sync) lists
which records leave which list.

## Stay inside the limits

- Send `updated_since` exactly as the API returned `updated_at`, for example
  `2026-10-01T14:03:22.118Z`. Anything that isn't an ISO 8601 date-time, or `after_id` without
  `updated_since`, gets `400 VALIDATION_ERROR`.
- At 100 rows a request, the 600-requests-a-minute limit reads 60,000 rows a minute. See
  [Authentication](/authentication#rate-limit).
- Run one sync at a time per key. Two jobs on one key share one rate limit.

## Get told instead of polling

A webhook tells you something changed, so you don't have to poll. Treat it as a nudge to read the
change feed from your checkpoint, not as the copy itself: a webhook can arrive late, twice, or not
at all. See [Webhooks](/webhooks).