Bulk exports

curl https://api.furniops.com/v1/exports \
  -H "Authorization: Bearer $FURNIOPS_API_KEY" \
  -H "Idempotency-Key: $(uuidgen)" \
  -H "Content-Type: application/json" \
  -d '{"kind": "products", "filters": {"updated_since": "2026-09-01T00:00:00Z"}}'

{
  "id": "exp_034XqBaL2mRLTZPXyRbaUm",
  "object": "export",
  "kind": "products",
  "status": "pending",
  "livemode": true,
  "filters": { "updated_since": "2026-09-01T00:00:00.000Z" },
  "rows": 0,
  "files": null,
  "error": null,
  "created_at": "2026-10-04T10:00:00.000Z",
  "completed_at": null,
  "expires_at": null
}

// Keep the key with your job: retrying the same request, even from another process, answers the
// export it started instead of starting a second one.
const started = await furniops.exports.create(
  { kind: "products", filters: { updated_since: "2026-09-01T00:00:00Z" } },
  { idempotencyKey: `export-${randomUUID()}` },
);
console.log(started.id, started.status); // "exp_…", "pending"

let current: Export = started;
while (current.status === "pending" || current.status === "running") {
  await new Promise((resolve) => setTimeout(resolve, 15_000));
  current = await furniops.exports.get(started.id);
}
if (current.status !== "succeeded") throw new Error(current.error ?? `export ${current.status}`);

{
  "id": "evt_034XqBaL2mRLTZPXyRbaUp",
  "type": "export.completed",
  "spec_version": "1.0",
  "source": "furniops/backend",
  "subject": "exp_034XqBaL2mRLTZPXyRbaUm",
  "time": "2026-10-04T10:03:41Z",
  "tenant_id": "org_034XqBaL2mRLTZPXyRbaUj",
  "version": 1,
  "schema_version": 1,
  "correlation_id": null,
  "causation_id": null,
  "livemode": true,
  "data": {
    "export": "exp_034XqBaL2mRLTZPXyRbaUm",
    "kind": "products",
    "status": "succeeded",
    "rows": 61250
  }
}

// `export.completed` reaches only webhook endpoints (the change feed does not carry it), and only
// those subscribed to it or to `*`. The SDK verifies it like any event, and `WebhookEvent` covers
// it: narrowing on `type` gives its data (`subject` is the export's ID).
const event: WebhookEvent = await furniops.webhooks.verify(raw, headers, secrets);
if (event.type === "export.completed" && event.data.status === "succeeded")
  await jobs.add("download-export", { id: event.data.export }); // then GET it for the links

{
  "id": "exp_034XqBaL2mRLTZPXyRbaUm",
  "object": "export",
  "kind": "products",
  "status": "succeeded",
  "livemode": true,
  "filters": { "updated_since": "2026-09-01T00:00:00.000Z" },
  "rows": 61250,
  "files": [
    {
      "dataset": "products",
      "url": "https://storage.example/exports/0199ae4c-6f3a-7d2e-9b1c-5e8f2a7d4c10/01a0f2b9-8c58-7ba3-bf2d-f4206b6564cc/part-0001.jsonl.gz?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Expires=900&X-Amz-Signature=5d1f",
      "rows": 50000,
      "bytes": 38214077
    },
    {
      "dataset": "products",
      "url": "https://storage.example/exports/0199ae4c-6f3a-7d2e-9b1c-5e8f2a7d4c10/01a0f2b9-8c58-7ba3-bf2d-f4206b6564cc/part-0002.jsonl.gz?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Expires=900&X-Amz-Signature=9a7c",
      "rows": 11250,
      "bytes": 8601932
    }
  ],
  "error": null,
  "created_at": "2026-10-04T10:00:00.000Z",
  "completed_at": "2026-10-04T10:03:41.000Z",
  "expires_at": "2026-10-11T10:03:41.000Z"
}

URL=$(curl -s https://api.furniops.com/v1/exports/exp_034XqBaL2mRLTZPXyRbaUm \
  -H "Authorization: Bearer $FURNIOPS_API_KEY" | jq -r '.files[0].url')

curl -s -o part-0001.jsonl.gz "$URL"            # no Authorization header
gunzip -c part-0001.jsonl.gz | wc -l           # one product per line
gunzip -c part-0001.jsonl.gz | jq -c '{id, title: .data.title, version}'

/** Every product in one export file, streamed: nothing holds the whole file in memory. */
export async function* readExportFile(url: string): AsyncGenerator<Product> {
  // No Authorization header: the link carries its own signature, and storage refuses a second one.
  const response = await fetch(url);
  if (!response.ok || response.body === null)
    throw new Error(`export file: HTTP ${response.status}`); // 403 once the link has expired
  const text = response.body
    .pipeThrough(new DecompressionStream("gzip"))
    .pipeThrough(new TextDecoderStream());
  let rest = "";
  for await (const chunk of text) {
    const lines = (rest + chunk).split("\n");
    rest = lines.pop() ?? ""; // a line cut across chunks waits for the next one
    for (const line of lines) if (line !== "") yield JSON.parse(line) as Product;
  }
  if (rest !== "") yield JSON.parse(rest) as Product;
}

// Links are valid for 15 minutes from the read that made them: read the export again before each
// file, so a slow consumer never starts on an expired link.
for (let part = 0; part < (current.files?.length ?? 0); part++) {
  const fresh = await furniops.exports.get(current.id);
  const file = fresh.files?.[part];
  if (file === undefined) break; // expired: the files were removed after 7 days
  for await (const product of readExportFile(file.url))
    await store.upsertProduct(product.id, product.version, product);
}

/** Saves one export file as it is (gzip JSON Lines), to decompress later. */
export async function saveExportFile(url: string, path: string): Promise<void> {
  const response = await fetch(url);
  if (!response.ok || response.body === null)
    throw new Error(`export file: HTTP ${response.status}`);
  await pipeline(response.body, createWriteStream(path));
}

try {
  await furniops.exports.create({ kind: "products" });
} catch (error) {
  if (error instanceof PermissionError && error.code === "feature_unavailable") {
    // The plan has no bulk exports: page GET /v1/products instead.
  } else if (error instanceof PermissionError && error.code === "insufficient_scope") {
    // The key holds exports:write but not products:read.
  } else if (error instanceof ConflictError && error.code === "conflict") {
    // Three exports are already pending or running in this mode: wait for export.completed.
  } else if (error instanceof RateLimitError && error.code === "quota_exceeded") {
    // This month's export_rows quota is spent; Retry-After counts down to the new month (UTC).
    await jobs.add("export", {}, { delay: error.retryAfterMs ?? 3_600_000 });
  } else throw error;
}