Syncing data

BASE="${FILMO_BASE_URL:-https://app.filmo.io}"

# Server time: the Date header of any response (here, a one-row read of the same list).
curl -sS -D headers.txt -o /dev/null -G "$BASE/api/v1/orders" \
  -H "Authorization: Bearer $FILMO_API_KEY" \
  --data-urlencode "limit=1"
SERVER_DATE=$(grep -i '^date:' headers.txt | cut -d' ' -f2- | tr -d '\r')

# B = server time minus 2 minutes and 1 second (GNU date; use gdate on macOS).
B=$(date -u -d "$SERVER_DATE - 121 seconds" +%Y-%m-%dT%H:%M:%S.000Z)
A=2026-10-01T00:00:00.000Z   # your stored watermark

# The window [A, B): one page, sorted by updatedAt.
curl -sS -G "$BASE/api/v1/orders" \
  -H "Authorization: Bearer $FILMO_API_KEY" \
  --data-urlencode "updatedFrom=$A" \
  --data-urlencode "updatedBefore=$B" \
  --data-urlencode "sort=updatedAt" \
  --data-urlencode "limit=500" \
  | jq '{records: (.data | length), hasMore: .pagination.hasMore}'

# If hasMore is true, run the same request again on [A, M) and [M, B),
# where M is the midpoint. Never pass the cursor. Store B only when
# every sub-window has been read.

// filmo-sync.mjs: Node 22+, ESM, no dependencies.
import { readFile, rename, writeFile } from "node:fs/promises";
import { fileURLToPath } from "node:url";

const BASE = process.env.FILMO_BASE_URL ?? "https://app.filmo.io";
const KEY = process.env.FILMO_API_KEY;
const INITIAL_FROM = process.env.FILMO_SYNC_FROM ?? "2000-01-01T00:00:00Z";

export const LIMIT = 500;
export const SAFETY_MS = 2 * 60_000 + 1_000; // 2 minutes + the 1 s precision of `Date`

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
export const iso = (ms) => new Date(ms).toISOString();

/** GET /api/v1<path>. Waits out a 429. Returns { status, body, date } for the statuses in `ok`. */
export async function api(path, params = {}, ok = [200]) {
  const url = new URL(`/api/v1${path}`, BASE);
  for (const [name, value] of Object.entries(params)) url.searchParams.set(name, String(value));
  for (;;) {
    const res = await fetch(url, {
      headers: { Authorization: `Bearer ${KEY}`, Accept: "application/json" },
    });
    if (res.status === 429) {
      await sleep(Number(res.headers.get("Retry-After") ?? "1") * 1000);
      continue;
    }
    const body = await res.json().catch(() => null);
    if (!ok.includes(res.status)) {
      throw new Error(`${res.status} ${body?.code ?? ""} (${res.headers.get("X-Request-Id")})`);
    }
    const date = Date.parse(res.headers.get("Date") ?? "");
    if (Number.isNaN(date)) throw new Error("the response has no Date header");
    return { status: res.status, body, date };
  }
}

/** Merge two id -> record maps, keeping the copy with the later updatedAt. */
function merge(into, from) {
  for (const [id, row] of from) {
    const known = into.get(id);
    if (!known || Date.parse(row.updatedAt) >= Date.parse(known.updatedAt)) into.set(id, row);
  }
  return into;
}

/**
 * Read the half-open window [a, b) (epoch ms) of a list: one page per query,
 * split in halves while pagination.hasMore is true. `field` is "updated" for
 * the incremental pass and "created" for the deletion sweep. Returns Map id -> record.
 */
export async function syncWindow(listPath, params, a, b, field = "updated") {
  const query = {
    ...params,
    [`${field}From`]: iso(a),
    [`${field}Before`]: iso(b),
    sort: `${field}At`,
    limit: LIMIT,
  };
  const { body } = await api(listPath, query);
  const rows = new Map(body.data.map((row) => [row.id, row]));
  if (!body.pagination.hasMore) return rows;
  if (b - a > 1) {
    const mid = a + Math.floor((b - a) / 2);
    const left = await syncWindow(listPath, params, a, mid, field);
    return merge(left, await syncWindow(listPath, params, mid, b, field));
  }
  return readOneMillisecond(listPath, query);
}

/** More than LIMIT records share one millisecond: follow the cursor, repeat until two reads agree. */
async function readOneMillisecond(listPath, query) {
  let previous = null;
  for (let attempt = 0; attempt < 5; attempt++) {
    const rows = new Map();
    let cursor;
    do {
      const { body } = await api(listPath, cursor ? { ...query, cursor } : query);
      for (const row of body.data) rows.set(row.id, row);
      cursor = body.pagination.hasMore ? body.pagination.nextCursor : undefined;
    } while (cursor);
    const signature = [...rows.values()].map((row) => `${row.id}@${row.updatedAt}`).sort().join("|");
    if (signature === previous) return rows;
    previous = signature;
  }
  throw new Error("more than 500 records share one millisecond and keep changing; retry later");
}

/** A watermark store is { load(): Promise<string | null>, save(watermark: string): Promise<void> }. */
export function jsonFileStore(file) {
  return {
    async load() {
      try {
        return JSON.parse(await readFile(file, "utf8")).watermark;
      } catch (error) {
        if (error.code === "ENOENT") return null;
        throw error;
      }
    },
    async save(watermark) {
      await writeFile(`${file}.tmp`, JSON.stringify({ watermark }));
      await rename(`${file}.tmp`, file); // never leave a half-written file
    },
  };
}

/**
 * One incremental pass. `variants` are the query parameters of a full pass
 * (see the table of hidden records): [{}] for orders, [{}, { archived: "true" }] for projects.
 * `apply` must upsert by id: after a crash the same window is read again.
 */
export async function incrementalPass(listPath, variants, store, apply) {
  const probe = await api(listPath, { ...variants[0], limit: 1 });
  const b = probe.date - SAFETY_MS; // server time minus the margin, never your own clock
  const a = Date.parse((await store.load()) ?? INITIAL_FROM);
  if (b <= a) return;
  const rows = new Map();
  for (const variant of variants) merge(rows, await syncWindow(listPath, variant, a, b));
  await apply([...rows.values()]);
  await store.save(iso(b)); // the watermark moves only after every sub-window was read
}

if (process.argv[1] === fileURLToPath(import.meta.url)) {
  await incrementalPass("/orders", [{}], jsonFileStore("watermark-orders.json"), async (rows) => {
    console.log(`${rows.length} orders changed`); // upsert them by id in your database here
  });
}

# filmo_sync.py: Python 3.11+, requests.
import json
import os
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime

import requests

BASE = os.environ.get("FILMO_BASE_URL", "https://app.filmo.io")
INITIAL_FROM = os.environ.get("FILMO_SYNC_FROM", "2000-01-01T00:00:00Z")
LIMIT = 500
SAFETY_MS = 2 * 60_000 + 1_000  # 2 minutes + the 1 s precision of `Date`

session = requests.Session()
session.headers.update(
    {"Authorization": f"Bearer {os.environ['FILMO_API_KEY']}", "Accept": "application/json"}
)


def iso(ms):
    stamp = datetime.fromtimestamp(ms / 1000, tz=timezone.utc)
    return stamp.isoformat(timespec="milliseconds").replace("+00:00", "Z")


def to_ms(text):
    return round(datetime.fromisoformat(text).timestamp() * 1000)


def api(path, params=None, ok=(200,)):
    """GET /api/v1<path>. Waits out a 429. Returns (status, body, server_time_ms)."""
    while True:
        res = session.get(f"{BASE}/api/v1{path}", params=params, timeout=30)
        if res.status_code == 429:
            time.sleep(int(res.headers.get("Retry-After", "1")))
            continue
        try:
            body = res.json()
        except ValueError:
            body = None
        if res.status_code not in ok:
            code = body.get("code") if isinstance(body, dict) else ""
            raise RuntimeError(f"{res.status_code} {code} ({res.headers.get('X-Request-Id')})")
        date = res.headers.get("Date")
        if not date:
            raise RuntimeError("the response has no Date header")
        return res.status_code, body, round(parsedate_to_datetime(date).timestamp() * 1000)


def merge(into, other):
    """Merge two id -> record dicts, keeping the copy with the later updatedAt."""
    for id_, row in other.items():
        known = into.get(id_)
        if known is None or to_ms(row["updatedAt"]) >= to_ms(known["updatedAt"]):
            into[id_] = row
    return into


def sync_window(list_path, params, a, b, field="updated"):
    """Read the half-open window [a, b) (epoch ms) of a list: one page per query,
    split in halves while pagination.hasMore is true. `field` is "updated" for the
    incremental pass and "created" for the deletion sweep. Returns {id: record}."""
    query = {
        **params,
        f"{field}From": iso(a),
        f"{field}Before": iso(b),
        "sort": f"{field}At",
        "limit": LIMIT,
    }
    _, body, _ = api(list_path, query)
    rows = {row["id"]: row for row in body["data"]}
    if not body["pagination"]["hasMore"]:
        return rows
    if b - a > 1:
        mid = a + (b - a) // 2
        left = sync_window(list_path, params, a, mid, field)
        return merge(left, sync_window(list_path, params, mid, b, field))
    return read_one_millisecond(list_path, query)


def read_one_millisecond(list_path, query):
    """More than LIMIT records share one millisecond: follow the cursor, repeat until two reads agree."""
    previous = None
    for _ in range(5):
        rows, cursor = {}, None
        while True:
            _, body, _ = api(list_path, {**query, "cursor": cursor} if cursor else query)
            rows.update({row["id"]: row for row in body["data"]})
            if not body["pagination"]["hasMore"]:
                break
            cursor = body["pagination"]["nextCursor"]
        signature = sorted((row["id"], row["updatedAt"]) for row in rows.values())
        if signature == previous:
            return rows
        previous = signature
    raise RuntimeError("more than 500 records share one millisecond and keep changing; retry later")


class JsonFileStore:
    """A watermark store is any object with load() -> str | None and save(watermark: str)."""

    def __init__(self, path):
        self.path = path

    def load(self):
        try:
            with open(self.path) as handle:
                return json.load(handle)["watermark"]
        except FileNotFoundError:
            return None

    def save(self, watermark):
        with open(f"{self.path}.tmp", "w") as handle:
            json.dump({"watermark": watermark}, handle)
        os.replace(f"{self.path}.tmp", self.path)  # never leave a half-written file


def incremental_pass(list_path, variants, store, apply):
    """One incremental pass. `variants` are the query parameters of a full pass
    (see the table of hidden records): [{}] for orders, [{}, {"archived": "true"}] for projects.
    `apply` must upsert by id: after a crash the same window is read again."""
    _, _, server_now = api(list_path, {**variants[0], "limit": 1})
    b = server_now - SAFETY_MS  # server time minus the margin, never your own clock
    a = to_ms(store.load() or INITIAL_FROM)
    if b <= a:
        return
    rows = {}
    for variant in variants:
        merge(rows, sync_window(list_path, variant, a, b))
    apply(list(rows.values()))
    store.save(iso(b))  # the watermark moves only after every sub-window was read


if __name__ == "__main__":
    incremental_pass(
        "/orders",
        [{}],
        JsonFileStore("watermark-orders.json"),
        lambda rows: print(f"{len(rows)} orders changed"),  # upsert them by id in your database here
    )

BASE="${FILMO_BASE_URL:-https://app.filmo.io}"

# First page, newest update first. status=all keeps won and lost opportunities in.
curl -sS -G "$BASE/api/v1/crm/opportunities" \
  -H "Authorization: Bearer $FILMO_API_KEY" \
  --data-urlencode "status=all" \
  --data-urlencode "sort=-updatedAt" \
  --data-urlencode "limit=500" \
  | jq '{newest: .data[0].updatedAt, hasMore: .pagination.hasMore, next: .pagination.nextCursor}'

# Read the next page with --data-urlencode "cursor=<next>" (same sort and limit)
# until you meet a record whose updatedAt is older than your watermark minus the overlap.

// early-stop.mjs: for lists with no updatedFrom/updatedBefore that sort by updatedAt.
import { api, iso, jsonFileStore, LIMIT, SAFETY_MS } from "./filmo-sync.mjs";

const OVERLAP_MS = 60 * 60_000; // re-read the last hour on every pass

/** `idKey` is the field that holds the id: "id" for most records, "opportunityId" for opportunities. */
export async function earlyStopPass(listPath, params, idKey, store, apply) {
  const probe = await api(listPath, { ...params, limit: 1 });
  const passStart = probe.date - SAFETY_MS;
  const saved = await store.load();
  const stopAt = saved ? Date.parse(saved) - OVERLAP_MS : -Infinity;

  const rows = new Map();
  let cursor;
  scan: for (;;) {
    const { body } = await api(listPath, {
      ...params,
      sort: "-updatedAt",
      limit: LIMIT,
      ...(cursor && { cursor }),
    });
    for (const row of body.data) {
      // updatedAt can be null on a few lists: a null never stops the scan
      if (row.updatedAt && Date.parse(row.updatedAt) < stopAt) break scan;
      rows.set(row[idKey], row);
    }
    if (!body.pagination.hasMore) break;
    cursor = body.pagination.nextCursor;
  }
  await apply([...rows.values()]);
  await store.save(iso(passStart));
}

// Opportunities: status=all, otherwise won and lost ones disappear from the list.
await earlyStopPass(
  "/crm/opportunities",
  { status: "all" },
  "opportunityId",
  jsonFileStore("watermark-opportunities.json"),
  async (rows) => console.log(`${rows.length} opportunities changed`),
);

# early_stop.py: for lists with no updatedFrom/updatedBefore that sort by updatedAt.
from filmo_sync import LIMIT, SAFETY_MS, JsonFileStore, api, iso, to_ms

OVERLAP_MS = 60 * 60_000  # re-read the last hour on every pass


def early_stop_pass(list_path, params, id_key, store, apply):
    """`id_key` is the field that holds the id: "id" for most records, "opportunityId" for opportunities."""
    _, _, server_now = api(list_path, {**params, "limit": 1})
    pass_start = server_now - SAFETY_MS
    saved = store.load()
    stop_at = to_ms(saved) - OVERLAP_MS if saved else float("-inf")

    rows, cursor, done = {}, None, False
    while not done:
        query = {**params, "sort": "-updatedAt", "limit": LIMIT}
        if cursor:
            query["cursor"] = cursor
        _, body, _ = api(list_path, query)
        for row in body["data"]:
            # updatedAt can be null on a few lists: a null never stops the scan
            if row["updatedAt"] and to_ms(row["updatedAt"]) < stop_at:
                done = True
                break
            rows[row[id_key]] = row
        if not body["pagination"]["hasMore"]:
            break
        cursor = body["pagination"]["nextCursor"]
    apply(list(rows.values()))
    store.save(iso(pass_start))


if __name__ == "__main__":
    # Opportunities: status=all, otherwise won and lost ones disappear from the list.
    early_stop_pass(
        "/crm/opportunities",
        {"status": "all"},
        "opportunityId",
        JsonFileStore("watermark-opportunities.json"),
        lambda rows: print(f"{len(rows)} opportunities changed"),
    )

BASE="${FILMO_BASE_URL:-https://app.filmo.io}"

# The ids Filmo reports for one createdAt window [FROM, TO). Same halving as the
# incremental pass when hasMore is true; shown here for a window that fits one page.
curl -sS -G "$BASE/api/v1/orders" \
  -H "Authorization: Bearer $FILMO_API_KEY" \
  --data-urlencode "createdFrom=2026-09-01T00:00:00.000Z" \
  --data-urlencode "createdBefore=2026-10-01T00:00:00.000Z" \
  --data-urlencode "sort=createdAt" \
  --data-urlencode "limit=500" \
  | jq -r '.data[] | [.id, .updatedAt] | @tsv' > remote.tsv

# For every id you hold in that window that is NOT in remote.tsv, ask the detail GET.
# 404 on an order means archived or deleted. 200 means the sweep skipped a live record.
curl -sS -o /dev/null -w "%{http_code}\n" "$BASE/api/v1/orders/$ORDER_ID" \
  -H "Authorization: Bearer $FILMO_API_KEY"

// reconcile.mjs: deletion sweep. Usage: node reconcile.mjs 2026-09-01T00:00:00Z 2026-10-01T00:00:00Z
import { readFile } from "node:fs/promises";
import { api, LIMIT, SAFETY_MS, syncWindow } from "./filmo-sync.mjs";

/**
 * Sweep one createdAt window [a, b) (epoch ms) of a list that has createdFrom/createdBefore.
 * `local` is Map id -> { createdAt, updatedAt } from your database.
 * Returns the records to upsert and the ids whose detail GET answered 404.
 */
export async function sweepWindow({ listPath, detailPath, variants, local, a, b }) {
  const remote = new Map();
  for (const variant of variants) {
    for (const [id, row] of await syncWindow(listPath, variant, a, b, "created")) remote.set(id, row);
  }

  const upsert = []; // new records and records whose updatedAt differs from yours
  for (const [id, row] of remote) {
    const mine = local.get(id);
    if (!mine || Date.parse(mine.updatedAt) !== Date.parse(row.updatedAt)) upsert.push(row);
  }

  const gone = []; // missing from the sweep AND 404 on the detail GET
  for (const [id, mine] of local) {
    const created = Date.parse(mine.createdAt);
    if (created < a || created >= b || remote.has(id)) continue;
    const { status, body } = await api(`${detailPath}/${encodeURIComponent(id)}`, {}, [200, 404]);
    if (status === 404) gone.push(id);
    else upsert.push(body.data); // the sweep skipped a live record
  }
  return { upsert, gone };
}

/** Lists with no created filters: every id by cursor. A list of candidates, never a verdict. */
export async function readIds(listPath, params, idKey = "id") {
  const ids = new Set();
  let cursor;
  do {
    const { body } = await api(listPath, { ...params, limit: LIMIT, ...(cursor && { cursor }) });
    for (const row of body.data) ids.add(row[idKey]);
    cursor = body.pagination.hasMore ? body.pagination.nextCursor : undefined;
  } while (cursor);
  return ids;
}

/** With a detail GET only a 404 counts. Without one, the id must be missing in two sweeps in a row. */
export async function confirmGone(candidates, previousCandidates, detailPath) {
  const gone = [];
  for (const id of candidates) {
    if (detailPath) {
      const { status } = await api(`${detailPath}/${encodeURIComponent(id)}`, {}, [200, 404]);
      if (status === 404) gone.push(id);
    } else if (previousCandidates.has(id)) {
      gone.push(id);
    }
  }
  return gone;
}

if (process.argv[2] && process.argv[3]) {
  const local = new Map(Object.entries(JSON.parse(await readFile("local-orders.json", "utf8"))));
  const a = Date.parse(process.argv[2]);
  const probe = await api("/orders", { limit: 1 });
  const b = Math.min(Date.parse(process.argv[3]), probe.date - SAFETY_MS); // server time, not your clock
  const { upsert, gone } = await sweepWindow({
    listPath: "/orders",
    detailPath: "/orders",
    variants: [{}],
    local,
    a,
    b,
  });
  // For orders a 404 means archived or deleted: mark them, do not delete.
  console.log(JSON.stringify({ upsert: upsert.map((row) => row.id), archivedOrDeleted: gone }, null, 2));
}

# reconcile.py: deletion sweep.
# Usage: python3 reconcile.py 2026-09-01T00:00:00Z 2026-10-01T00:00:00Z
import json
import sys
from urllib.parse import quote

from filmo_sync import LIMIT, SAFETY_MS, api, sync_window, to_ms


def sweep_window(list_path, detail_path, variants, local, a, b):
    """Sweep one createdAt window [a, b) (epoch ms) of a list that has createdFrom/createdBefore.
    `local` is {id: {"createdAt": ..., "updatedAt": ...}} from your database.
    Returns the records to upsert and the ids whose detail GET answered 404."""
    remote = {}
    for variant in variants:
        remote.update(sync_window(list_path, variant, a, b, "created"))

    upsert = []  # new records and records whose updatedAt differs from yours
    for id_, row in remote.items():
        mine = local.get(id_)
        if mine is None or to_ms(mine["updatedAt"]) != to_ms(row["updatedAt"]):
            upsert.append(row)

    gone = []  # missing from the sweep AND 404 on the detail GET
    for id_, mine in local.items():
        created = to_ms(mine["createdAt"])
        if created < a or created >= b or id_ in remote:
            continue
        status, body, _ = api(f"{detail_path}/{quote(id_, safe='')}", None, (200, 404))
        if status == 404:
            gone.append(id_)
        else:
            upsert.append(body["data"])  # the sweep skipped a live record
    return upsert, gone


def read_ids(list_path, params, id_key="id"):
    """Lists with no created filters: every id by cursor. A list of candidates, never a verdict."""
    ids, cursor = set(), None
    while True:
        query = {**params, "limit": LIMIT}
        if cursor:
            query["cursor"] = cursor
        _, body, _ = api(list_path, query)
        ids.update(row[id_key] for row in body["data"])
        if not body["pagination"]["hasMore"]:
            return ids
        cursor = body["pagination"]["nextCursor"]


def confirm_gone(candidates, previous_candidates, detail_path):
    """With a detail GET only a 404 counts. Without one, the id must be missing in two sweeps in a row."""
    gone = []
    for id_ in candidates:
        if detail_path:
            status, _, _ = api(f"{detail_path}/{quote(id_, safe='')}", None, (200, 404))
            if status == 404:
                gone.append(id_)
        elif id_ in previous_candidates:
            gone.append(id_)
    return gone


if __name__ == "__main__" and len(sys.argv) == 3:
    with open("local-orders.json") as handle:
        local = json.load(handle)
    _, _, server_now = api("/orders", {"limit": 1})
    a = to_ms(sys.argv[1])
    b = min(to_ms(sys.argv[2]), server_now - SAFETY_MS)  # server time, not your clock
    upsert, gone = sweep_window("/orders", "/orders", [{}], local, a, b)
    # For orders a 404 means archived or deleted: mark them, do not delete.
    print(json.dumps({"upsert": [row["id"] for row in upsert], "archivedOrDeleted": gone}, indent=2))