Feed API

"""Incremental sync: pages new/updated jobs since the last run, then sweeps expired IDs.
   Cursor is checkpointed so a crash resumes mid-stream. Pages serially — one
   in-flight request per cursor — and retries only 503 (per Retry-After)."""
import json, time, datetime as dt, requests, pathlib

API_KEY = "YOUR_API_KEY"
BASE = "https://connect.jobo.world/api/jobs"
STATE = pathlib.Path("sync_state.json")
OVERLAP = dt.timedelta(minutes=15)

state = json.loads(STATE.read_text()) if STATE.exists() else {"last_run_at": None, "cursor": None}
this_run_at = dt.datetime.now(dt.timezone.utc).isoformat()
store = {}  # ← your DB. key = job["id"]

def post(body):
    while True:
        r = requests.post(f"{BASE}/feed",
                          headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
                          json=body)
        if r.status_code == 503:
            time.sleep(int(r.headers.get("Retry-After", "5"))); continue
        r.raise_for_status(); return r.json()

# 1. Page the feed with updated_after = last_run - overlap.
#    A continuation body carries ONLY the cursor: it embeds the first
#    request's filters and batch size, and other values are ignored.
updated_after = None
if state["last_run_at"]:
    updated_after = (dt.datetime.fromisoformat(state["last_run_at"]) - OVERLAP).isoformat()

cursor = state.get("cursor")
while True:
    if cursor:
        body = {"cursor": cursor}
    else:
        body = {"batch_size": 1000}
        if updated_after: body["updated_after"] = updated_after
    data = post(body)

    for job in data["jobs"]:
        store[job["id"]] = job  # ← upsert in your DB

    cursor = data["next_cursor"]
    STATE.write_text(json.dumps({**state, "cursor": cursor}))  # checkpoint
    if not data["has_more"]: break

# 2. Sweep expired IDs since the last successful run
if state["last_run_at"]:
    cursor = None
    while True:
        params = {"expired_since": state["last_run_at"], "batch_size": 10000}
        if cursor: params["cursor"] = cursor
        r = requests.get(f"{BASE}/expired",
                         headers={"X-Api-Key": API_KEY}, params=params).json()
        for jid in r["job_ids"]:
            store.pop(jid, None)  # ← mark expired in your DB
        cursor = r["next_cursor"]
        if not r["has_more"]: break

# 3. Commit the new checkpoint only after both halves succeeded.
STATE.write_text(json.dumps({"last_run_at": this_run_at, "cursor": None}))
print(f"Synced. {len(store)} active jobs in store.")

// Same pattern as the Python sample: page /feed with updated_after, then sweep /expired.
import fs from "node:fs";

const API_KEY = "YOUR_API_KEY";
const BASE = "https://connect.jobo.world/api/jobs";
const STATE_FILE = "sync_state.json";
const OVERLAP_MS = 15 * 60 * 1000;

const state = fs.existsSync(STATE_FILE)
  ? JSON.parse(fs.readFileSync(STATE_FILE, "utf8"))
  : { last_run_at: null, cursor: null };
const thisRunAt = new Date().toISOString();
const store = new Map(); // ← your DB, keyed by job.id

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

async function post(body) {
  while (true) {
    const r = await fetch(`${BASE}/feed`, {
      method: "POST",
      headers: { "X-Api-Key": API_KEY, "Content-Type": "application/json" },
      body: JSON.stringify(body),
    });
    if (r.status === 503) {
      await sleep(parseInt(r.headers.get("Retry-After") || "5", 10) * 1000);
      continue;
    }
    if (!r.ok) throw new Error(`Feed ${r.status}: ${await r.text()}`);
    return r.json();
  }
}

// 1. Page feed with updated_after = last_run - overlap. A continuation body
//    carries ONLY the cursor — it embeds the first request's filters and
//    batch size, and other values are ignored.
const updatedAfter = state.last_run_at
  ? new Date(new Date(state.last_run_at).getTime() - OVERLAP_MS).toISOString()
  : null;

let cursor = state.cursor;
while (true) {
  const body = cursor ? { cursor } : { batch_size: 1000 };
  if (!cursor && updatedAfter) body.updated_after = updatedAfter;
  const data = await post(body);

  for (const job of data.jobs) store.set(job.id, job); // ← upsert

  cursor = data.next_cursor;
  fs.writeFileSync(STATE_FILE, JSON.stringify({ ...state, cursor })); // checkpoint
  if (!data.has_more) break;
}

// 2. Sweep expired IDs since the last successful run
if (state.last_run_at) {
  let c = null;
  do {
    const params = new URLSearchParams({
      expired_since: state.last_run_at,
      batch_size: "10000",
    });
    if (c) params.set("cursor", c);
    const r = await fetch(`${BASE}/expired?${params}`, {
      headers: { "X-Api-Key": API_KEY },
    }).then((res) => res.json());
    for (const id of r.job_ids) store.delete(id); // ← mark expired
    c = r.next_cursor;
    if (!r.has_more) break;
  } while (c);
}

// 3. Commit the new checkpoint only after both halves succeeded.
fs.writeFileSync(
  STATE_FILE,
  JSON.stringify({ last_run_at: thisRunAt, cursor: null })
);
console.log(`Synced. ${store.size} active jobs in store.`);