Syncing

import json
import os
from pathlib import Path

import requests

API = "https://api.pinakelai.com/v1"
SCOPE = {"jurisdiction": "AD,AL"}
STATE = Path("pinakel_feed_state.json")

session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['PINAKEL_API_KEY']}"


def load_cursor() -> str | None:
    return json.loads(STATE.read_text())["cursor"] if STATE.exists() else None


def save_cursor(cursor: str) -> None:
    tmp = STATE.with_suffix(".tmp")
    tmp.write_text(json.dumps({"cursor": cursor}))
    tmp.replace(STATE)  # atomic: a crash never leaves half a cursor


def fetch_full(ids: list[str]) -> list[dict]:
    """Full records for up to 100 ids."""
    resp = session.post(f"{API}/trademarks/batch", json={"ids": ids}, timeout=60)
    resp.raise_for_status()
    return resp.json()["data"]


def initial_load(upsert) -> None:
    resp = session.get(f"{API}/changes", params=SCOPE, timeout=30)   # 1. cursor first
    resp.raise_for_status()
    save_cursor(resp.json()["next_cursor"])
    params, ids = dict(SCOPE, limit=100), []
    while True:                                                       # 2. copy the scope
        page = session.get(f"{API}/trademarks", params=params, timeout=60)
        page.raise_for_status()
        body = page.json()
        ids += [tm["id"] for tm in body["data"]]
        while len(ids) >= 100 or (ids and not body["has_more"]):
            for record in fetch_full(ids[:100]):
                upsert(record)
            ids = ids[100:]
        if not body["has_more"]:
            return                                                    # 3. now run sync_once()
        params["cursor"] = body["next_cursor"]


def sync_once(upsert) -> None:
    """Drain the feed up to complete_through. Run it every 15 minutes or less often."""
    cursor = load_cursor()
    while True:
        resp = session.get(f"{API}/changes", params=dict(SCOPE, limit=100, cursor=cursor),
                           timeout=30)
        if resp.status_code == 503:
            return  # every register in scope is reloading, or the service is busy: next run
        resp.raise_for_status()
        page = resp.json()
        changed = [c["trademark"]["id"] for c in page["data"] if c["type"] == "trademark.changed"]
        for i in range(0, len(changed), 100):
            for record in fetch_full(changed[i:i + 100]):
                upsert(record)
        save_cursor(page["next_cursor"])  # after every page
        if not page["has_more"]:
            return

import { readFile, writeFile, rename } from "node:fs/promises";

const API = "https://api.pinakelai.com/v1";
const SCOPE = { jurisdiction: "AD,AL" };
const STATE = "pinakel_feed_state.json";
const headers = { Authorization: `Bearer ${process.env.PINAKEL_API_KEY}` };

async function loadCursor() {
  try { return JSON.parse(await readFile(STATE, "utf8")).cursor; } catch { return null; }
}

async function saveCursor(cursor) {
  await writeFile(`${STATE}.tmp`, JSON.stringify({ cursor }));
  await rename(`${STATE}.tmp`, STATE); // atomic
}

async function fetchFull(ids) {
  const res = await fetch(`${API}/trademarks/batch`, {
    method: "POST",
    headers: { ...headers, "Content-Type": "application/json" },
    body: JSON.stringify({ ids }),
  });
  if (!res.ok) throw new Error(`${res.status}: ${await res.text()}`);
  return (await res.json()).data;
}

// Drain the feed up to complete_through. Run it every 15 minutes or less often.
export async function syncOnce(upsert) {
  let cursor = await loadCursor();
  while (true) {
    const query = new URLSearchParams({ ...SCOPE, limit: "100" });
    if (cursor) query.set("cursor", cursor);
    const res = await fetch(`${API}/changes?${query}`, { headers });
    if (res.status === 503) return; // reloading or busy: try on the next run
    if (!res.ok) throw new Error(`${res.status}: ${await res.text()}`);
    const page = await res.json();
    const ids = page.data.filter((c) => c.type === "trademark.changed").map((c) => c.trademark.id);
    for (let i = 0; i < ids.length; i += 100) {
      for (const record of await fetchFull(ids.slice(i, i + 100))) await upsert(record);
    }
    cursor = page.next_cursor;
    await saveCursor(cursor); // after every page
    if (!page.has_more) return;
  }
}