import type { CatalogChange } from "@agg-build/sdk";

function isNewer(held: { version: number; changedAt: string } | undefined, item: CatalogChange) {
  if (!held) return true;
  if (item.version !== held.version) return item.version > held.version;
  return Date.parse(item.changedAt) > Date.parse(held.changedAt);
}

import { createAggClient } from "@agg-build/sdk";

const client = createAggClient({
  baseUrl: "https://api.agg.market",
  wsUrl: "wss://ws.agg.market/ws",
  appId: process.env.AGG_APP_ID,
  apiKey: process.env.AGG_API_KEY, // server-side only; sent as x-app-api-key
});

// onChange is not awaited. Chain your writes so they land in order.
let writes = Promise.resolve();

const handle = client.watchCatalog({
  since: (await loadCatalogPosition()) ?? new Date(Date.now() - 60 * 60_000),
  onChange: (change) => {
    writes = writes.then(() =>
      change.op === "deleted"
        ? db.remove(change.entity, change.id)
        : db.upsert(change.entity, change.id, change.event ?? change.market),
    );
  },
  onSynced: ({ lastChangedAt }) => console.log("catalog live from", lastChangedAt),
  onError: (err) => console.error("catalog feed", err),
  onResyncRequired: async () => {
    const startedAt = new Date();
    await relistCatalog(); // see "Resync after 24 hours" below
    handle.resync(startedAt);
  },
});

// Save the position only once everything before it is written.
setInterval(() => {
  const position = handle.lastChangedAt;
  if (position) void writes.then(() => saveCatalogPosition(position));
}, 10_000);

await handle.ready; // first catch-up done

import { useCatalogFeed } from "@agg-build/hooks";

function CatalogSync({ since }: { since: string }) {
  const { status, lastChangedAt, error } = useCatalogFeed({
    since,
    venues: ["polymarket", "kalshi"],
    matchedOnly: true,
    onChange: (change) => catalogStore.apply(change),
  });

  if (error) return <p>Catalog sync error: {String(error)}</p>;
  return (
    <p>
      Catalog {status}, up to {lastChangedAt ?? "start"}
    </p>
  );
}

{ "action": "subscribe", "channel": "catalog", "venues": ["polymarket", "kalshi"], "id": "cat-1" }

{
  "data": [
    {
      "entity": "market",
      "id": "vm_8f2c",
      "op": "created",
      "version": 1,
      "changedAt": "2026-10-21T12:00:41.512Z",
      "venue": "kalshi",
      "market": { "id": "vm_8f2c", "venueEventId": "ve_51aa", "question": "…", "matched": false, "clusterAnchorId": null, "clusterVenues": ["kalshi"], "…": "…" }
    }
  ],
  "nextCursor": "MTc5MjU5MzI0MTUxMjAwMHxtYXJrZXQ6dm1fOGYyYw",
  "hasMore": false
}

import asyncio, json, os
from datetime import datetime, timezone
import httpx, websockets

API = "https://api.agg.market"
app_id, app_key = os.environ["AGG_APP_ID"], os.environ["AGG_API_KEY"]
HEADERS = {"x-app-id": app_id, "x-app-api-key": app_key}
VENUES = ["polymarket", "kalshi"]

class ResyncRequired(Exception):
    pass

async def catch_up(http):
    since = datetime.fromtimestamp(load_position() - 60, timezone.utc).isoformat()
    params = {"since": since, "venues": VENUES}
    while True:
        r = await http.get(f"{API}/catalog/changes", params=params, headers=HEADERS)
        if r.status_code == 410:
            raise ResyncRequired()
        r.raise_for_status()
        page = r.json()
        for change in page["data"]:
            apply_if_newer(change)
        if not page["hasMore"]:
            return
        params = {"cursor": page["nextCursor"]}

async def run():
    url = f"wss://ws.agg.market/ws?appId={app_id}"
    async with httpx.AsyncClient() as http, websockets.connect(
        url, additional_headers={"x-app-api-key": app_key}
    ) as ws:
        held, task, again = [], None, False
        async for raw in ws:
            msg = json.loads(raw)
            kind = msg.get("type")
            if kind == "connected":
                await ws.send(json.dumps({"action": "subscribe", "channel": "catalog", "venues": VENUES}))
            elif kind == "catalog_change":
                held.append(msg)
            elif (kind == "subscribed" and msg.get("channel") == "catalog") or (
                kind == "stream_interrupted" and "catalog" in msg.get("channels", [])
            ):
                again = True  # catch up after every ack and every interruption
            if task is not None and task.done():
                task.result()  # raises ResyncRequired or the HTTP error
                task = None
            if again and task is None:
                again = False
                task = asyncio.create_task(catch_up(http))
            if task is None:  # live: apply held and new frames directly
                for change in held:
                    apply_if_newer(change)
                held.clear()

{
  "message": "Position is older than the catalog feed retention window; re-list with GET /venue-events and GET /venue-markets, then resume from when that re-list started",
  "code": "resync_required"
}