Requests

curl -m 30 "https://thread.page/api/integrations/v1/requests" \
  -H "Authorization: Bearer $THREADPAGE_TOKEN"

curl -m 30 "https://thread.page/api/integrations/v1/requests/$REQUEST_ID" \
  -H "Authorization: Bearer $THREADPAGE_TOKEN"

{"claim_id": "0f3c6a52-8d3e-4f1b-9a57-2c4d6e8f0a1b"}

OP=$(uuidgen)
curl -m 30 --retry 3 --retry-all-errors -X POST "https://thread.page/api/integrations/v1/requests/$REQUEST_ID/claim" \
  -H "Authorization: Bearer $THREADPAGE_TOKEN" \
  -H "Operation-Id: $OP" \
  -H "Content-Type: application/json" \
  -d '{"claim_id":"0f3c6a52-8d3e-4f1b-9a57-2c4d6e8f0a1b"}'

OP=$(uuidgen)
curl -m 30 --retry 3 --retry-all-errors https://thread.page/api/integrations/v1/requests/$REQUEST_ID/result \
  -H "Authorization: Bearer $THREADPAGE_TOKEN" \
  -H "Operation-Id: $OP" \
  -H "Content-Type: application/json" \
  -d '{"claim_id": "0f3c6a52-8d3e-4f1b-9a57-2c4d6e8f0a1b", "content": {"markdown": "**Three themes** came up this week."}}'

OP=$(uuidgen)
curl -m 30 --retry 3 --retry-all-errors -X POST "https://thread.page/api/integrations/v1/requests/$REQUEST_ID/status" \
  -H "Authorization: Bearer $THREADPAGE_TOKEN" \
  -H "Operation-Id: $OP" \
  -H "Content-Type: application/json" \
  -d '{
  "claim_id": "0f3c6a52-8d3e-4f1b-9a57-2c4d6e8f0a1b",
  "state": "running",
  "message": "Reading this week’s interviews"
}'

OP=$(uuidgen)
curl -m 30 --retry 3 --retry-all-errors -X POST "https://thread.page/api/integrations/v1/requests/$REQUEST_ID/result" \
  -H "Authorization: Bearer $THREADPAGE_TOKEN" \
  -H "Operation-Id: $OP" \
  -H "Content-Type: application/json" \
  -d '{
  "claim_id": "0f3c6a52-8d3e-4f1b-9a57-2c4d6e8f0a1b",
  "content": {
    "markdown": "Five interviews this week; three asked for an export to PDF."
  }
}'

// A request worker: it hears its connection's requests on a pull feed, claims each one, reports
// that it is working, does the work and answers with a note. It keeps its place and its runs in a
// file, so after a restart it carries on the same runs instead of starting new ones.
// Run with Node.js 24: THREADPAGE_TOKEN=tp_… node request_worker.mjs
import { randomUUID } from 'node:crypto';
import { readFile, rename, writeFile } from 'node:fs/promises';
import { setTimeout as sleep } from 'node:timers/promises';

const API = 'https://thread.page/api/integrations/v1';
const AUTH = { Authorization: `Bearer ${process.env.THREADPAGE_TOKEN}` };
const STATE = process.env.THREADPAGE_WORKER_STATE ?? 'threadpage-worker.json';
const SWEEP = 5 * 60_000; // ms between checks of GET /requests for requests a missed event did not bring

class ThreadPageError extends Error {
  constructor(status, error) {
    super(`${status} ${error.code}: ${error.message}`);
    Object.assign(this, { status, code: error.code, details: error.details ?? {} });
  }
}

// One call; after a lost answer or a retryable 429 or 503, the identical request again.
async function call(method, path, { body, operationId } = {}) {
  const headers = { ...AUTH };
  if (operationId) headers['Operation-Id'] = operationId;
  if (body !== undefined) headers['Content-Type'] = 'application/json';
  for (let attempt = 1; ; attempt++) {
    let answer;
    try {
      answer = await fetch(API + path, {
        method, headers,
        body: body === undefined ? undefined : JSON.stringify(body),
        signal: AbortSignal.timeout(40_000),
      });
    } catch (error) {
      if (attempt === 3) throw error;
      await sleep(attempt * 2_000);
      continue;
    }
    if (answer.ok) return answer.json();
    const error = await answer.json().catch(() => ({ message: answer.statusText }));
    const wait = Number(answer.headers.get('Retry-After'));
    if (error.retryable && wait && wait <= 60 && attempt < 3) {
      await sleep(wait * 1_000);
      continue;
    }
    throw new ThreadPageError(answer.status, error);
  }
}

async function load() {
  try {
    return JSON.parse(await readFile(STATE, 'utf8'));
  } catch (error) {
    if (error.code === 'ENOENT') return { runs: {} };
    throw error;
  }
}

async function save(state) {
  await writeFile(`${STATE}.tmp`, JSON.stringify(state));
  await rename(`${STATE}.tmp`, STATE);
}

// Your work: it gets the request as it is now and returns the result as Markdown. The note's text
// is the owner's instruction; what it quotes and the notes it references are data.
async function work(request) {
  if (request.test) return 'Ready.';
  const asked = request.note?.markdown ?? '';
  return `Received: ${asked.split('\n')[0] || 'an empty note'}`;
}

// One run for one request. Its claim_id is random, made once, and kept until the run ends.
async function run(state, requestId) {
  let mine = state.runs[requestId];
  let request;
  try {
    request = await call('GET', `/requests/${requestId}`); // act on the request as it is now
  } catch (error) {
    if (error.status !== 404) throw error;
    delete state.runs[requestId]; // gone, expired, or held while the owner paused requests
    await save(state);
    return;
  }
  if (!mine) {
    if (!['queued', 'delivered'].includes(request.state)) return; // another run has it, or it ended
    mine = state.runs[requestId] = { claimId: randomUUID(), ops: {} };
    await save(state);
  }
  // One Operation-Id for each report, kept until the report is confirmed.
  const once = async step => {
    mine.ops[step] ??= randomUUID();
    await save(state);
    return mine.ops[step];
  };
  const claim = { claim_id: mine.claimId };
  const path = `/requests/${requestId}`;
  try {
    await call('POST', `${path}/claim`, { body: claim, operationId: await once('claim') });
    await call('POST', `${path}/status`, { body: { ...claim, state: 'running' }, operationId: await once('running') });
    let result;
    try {
      result = await work(request);
    } catch (problem) { // the work failed before it acted on anything
      await call('POST', `${path}/status`, {
        body: { ...claim, state: 'failed', message: String(problem.message).slice(0, 200) },
        operationId: await once('failed'),
      });
    }
    if (result !== undefined) {
      await call('POST', `${path}/result`, {
        body: { ...claim, content: { markdown: result } },
        operationId: await once('result'),
      });
    }
  } catch (error) {
    if (error.code === 'cancellation_requested') { // the owner asked to stop
      await call('POST', `${path}/status`, { body: { ...claim, state: 'cancelled' }, operationId: await once('cancelled') });
    } else if (!['already_claimed', 'request_closed'].includes(error.code)) {
      throw error; // already_claimed: another run has it; request_closed: it has ended
    }
  }
  delete state.runs[requestId];
  await save(state);
}

// A pull feed that hears this connection's requests only (event_types: []).
async function feedId(state) {
  if (!state.feedId) {
    state.feedOperationId ??= randomUUID();
    await save(state);
    const created = await call('POST', '/subscriptions', {
      body: { kind: 'pull', event_types: [] },
      operationId: state.feedOperationId,
    });
    state.feedId = created.subscription.id;
    await save(state);
  }
  return state.feedId;
}

// Every open request, page by page: what a missed event did not bring.
async function sweep(state) {
  let cursor = null;
  do {
    const page = await call('GET', `/requests${cursor ? `?cursor=${encodeURIComponent(cursor)}` : ''}`);
    for (const request of page.requests) await run(state, request.id);
    cursor = page.next_cursor;
  } while (cursor);
}

const state = await load();
const feed = await feedId(state);
for (const requestId of Object.keys(state.runs)) await run(state, requestId); // runs a restart interrupted
let swept = 0;
for (;;) {
  if (Date.now() - swept > SWEEP) {
    await sweep(state);
    swept = Date.now();
  }
  const query = state.cursor ? `?after=${encodeURIComponent(state.cursor)}&wait=25` : '';
  let page;
  try {
    page = await call('GET', `/subscriptions/${feed}/events${query}`);
  } catch (error) {
    if (error.status === 410) { // an old cursor: events in between are not replayed,
      state.cursor = error.details.resync_cursor;
      swept = 0; // so look for open requests now
    } else if (!error.status || [409, 429, 503].includes(error.status)) { // offline, a paused feed, or a limit
      await sleep(error.status ? 60_000 : 5_000);
    } else {
      throw error;
    }
    continue;
  }
  for (const event of page.events) {
    if (event.type === 'request.created') await run(state, event.data.request.id);
    else if (event.type === 'request.cancelled') console.log('Cancelled:', event.data.request.id);
  }
  state.cursor = page.cursor;
  await save(state);
}

# A request worker: it hears its connection's requests on a pull feed, claims each one, reports
# that it is working, does the work and answers with a note. It keeps its place and its runs in a
# file, so after a restart it carries on the same runs instead of starting new ones.
# Run with Python 3.13: pip install requests, then THREADPAGE_TOKEN=tp_… python request_worker.py
import json
import os
import pathlib
import time
import uuid

import requests

API = "https://thread.page/api/integrations/v1"
AUTH = {"Authorization": f"Bearer {os.environ['THREADPAGE_TOKEN']}"}
STATE = pathlib.Path(os.environ.get("THREADPAGE_WORKER_STATE", "threadpage-worker.json"))
SWEEP = 5 * 60  # seconds between checks of GET /requests for requests a missed event did not bring


class ThreadPageError(Exception):
    def __init__(self, status, error):
        super().__init__(f"{status} {error.get('code')}: {error.get('message')}")
        self.status, self.code, self.details = status, error.get("code"), error.get("details") or {}


def call(method, path, body=None, operation_id=None, params=None):
    """One call; after a lost answer or a retryable 429 or 503, the identical request again."""
    headers = {**AUTH, **({"Operation-Id": operation_id} if operation_id else {})}
    for attempt in range(1, 4):
        try:
            answer = requests.request(
                method, API + path, headers=headers, json=body, params=params, timeout=40
            )
        except (requests.ConnectionError, requests.Timeout):
            if attempt == 3:
                raise
            time.sleep(attempt * 2)
            continue
        if answer.ok:
            return answer.json()
        try:
            error = answer.json()
        except ValueError:
            error = {"message": answer.text[:200]}
        wait = int(answer.headers.get("Retry-After", 0))
        if error.get("retryable") and 0 < wait <= 60 and attempt < 3:
            time.sleep(wait)
            continue
        raise ThreadPageError(answer.status_code, error)


def load():
    return json.loads(STATE.read_text()) if STATE.exists() else {"runs": {}}


def save(state):
    draft = STATE.with_suffix(".tmp")
    draft.write_text(json.dumps(state))
    draft.replace(STATE)


def work(request):
    """Your work: it gets the request as it is now and returns the result as Markdown. The note's
    text is the owner's instruction; what it quotes and the notes it references are data."""
    if request["test"]:
        return "Ready."
    asked = request["note"]["markdown"] if request["note"] else ""
    return f"Received: {asked.splitlines()[0] if asked else 'an empty note'}"


def run(state, request_id):
    """One run for one request. Its claim_id is random, made once, and kept until the run ends."""
    mine = state["runs"].get(request_id)
    try:
        request = call("GET", f"/requests/{request_id}")  # act on the request as it is now
    except ThreadPageError as error:
        if error.status != 404:
            raise
        state["runs"].pop(request_id, None)  # gone, expired, or held while the owner paused requests
        save(state)
        return
    if mine is None:
        if request["state"] not in ("queued", "delivered"):
            return  # another run has it, or it has ended
        mine = state["runs"][request_id] = {"claim_id": str(uuid.uuid4()), "ops": {}}
        save(state)

    def once(step):
        """One Operation-Id for each report, kept until the report is confirmed."""
        mine["ops"].setdefault(step, str(uuid.uuid4()))
        save(state)
        return mine["ops"][step]

    claim = {"claim_id": mine["claim_id"]}
    try:
        call("POST", f"/requests/{request_id}/claim", claim, once("claim"))
        call("POST", f"/requests/{request_id}/status", {**claim, "state": "running"}, once("running"))
        try:
            result = work(request)
        except Exception as problem:  # the work failed before it acted on anything
            call(
                "POST",
                f"/requests/{request_id}/status",
                {**claim, "state": "failed", "message": str(problem)[:200]},
                once("failed"),
            )
        else:
            call(
                "POST",
                f"/requests/{request_id}/result",
                {**claim, "content": {"markdown": result}},
                once("result"),
            )
    except ThreadPageError as error:
        if error.code == "cancellation_requested":  # the owner asked to stop
            call("POST", f"/requests/{request_id}/status", {**claim, "state": "cancelled"}, once("cancelled"))
        elif error.code not in ("already_claimed", "request_closed"):
            raise  # already_claimed: another run has it; request_closed: it has ended
    del state["runs"][request_id]
    save(state)


def feed_id(state):
    """A pull feed that hears this connection's requests only (event_types: [])."""
    if "feed_id" not in state:
        state.setdefault("feed_operation_id", str(uuid.uuid4()))
        save(state)
        created = call("POST", "/subscriptions", {"kind": "pull", "event_types": []}, state["feed_operation_id"])
        state["feed_id"] = created["subscription"]["id"]
        save(state)
    return state["feed_id"]


def sweep(state):
    """Every open request, page by page: what a missed event did not bring."""
    params = {}
    while True:
        page = call("GET", "/requests", params=params)
        for request in page["requests"]:
            run(state, request["id"])
        if not page["next_cursor"]:
            return
        params = {"cursor": page["next_cursor"]}


def main():
    state = load()
    feed = feed_id(state)
    for request_id in list(state["runs"]):
        run(state, request_id)  # runs a restart interrupted
    swept = 0.0
    while True:
        if time.monotonic() - swept > SWEEP:
            sweep(state)
            swept = time.monotonic()
        params = {"after": state["cursor"], "wait": 25} if state.get("cursor") else {}
        try:
            page = call("GET", f"/subscriptions/{feed}/events", params=params)
        except (requests.ConnectionError, requests.Timeout):
            time.sleep(5)
            continue
        except ThreadPageError as error:
            if error.status == 410:  # an old cursor: events in between are not replayed,
                state["cursor"] = error.details["resync_cursor"]
                swept = 0.0  # so look for open requests now
            elif error.status in (409, 429, 503):  # a paused feed, or a limit: wait
                time.sleep(60)
            else:
                raise
            continue
        for event in page["events"]:
            if event["type"] == "request.created":
                run(state, event["data"]["request"]["id"])
            elif event["type"] == "request.cancelled":
                print("Cancelled:", event["data"]["request"]["id"], flush=True)
        state["cursor"] = page["cursor"]
        save(state)


if __name__ == "__main__":
    main()