profile-sync.ts

// A small client for profile sync (contracts/profile-sync.md; the routes in
// ../../reference/server-api.md), with nothing but Node's own fetch. Your backend calls it with
// the site's secret key, which never goes to a browser, an app or a log.
//
// What it teaches:
// - Every request is safe to send again: it carries the same version and content, so a retry
//   cannot apply twice (contracts/profile-sync.md, "Versions"). Profile sync has no idempotency key.
// - Retry only 429 (after Retry-After), 5xx and no answer, with a growing wait. Any other error
//   comes back the same until you fix the request.
// - Send many changes as batches of up to 100, never one call per profile, and never a call per
//   page view.
// - Read the platform's list a page at a time to reconcile.
// - Send the secret key only over https (plain http only to this machine), and never follow a
//   redirect, which could carry the key to another address.

/** Who may see a field's value (contracts/profile-sync.md, "Fields and visibility"). */
export type Visibility = "public" | "ai_only" | "private";

export type Field = Readonly<{ value: string; visibility: Visibility }>;

/** A whole profile, as an upsert sends it. Fields left out are gone from the platform's copy once it applies. */
export type Profile = Readonly<{
  /** Grows with every change to the profile, deletes included. */
  version: number;
  /** Your user ID for the profile's owner: the `sub` you sign them in with. */
  owner: string;
  fields: Readonly<Record<string, Field>>;
}>;

export type Rejection = Readonly<{
  outcome: "rejected";
  /** `owner-missing`, `invalid-field`, `version-reused` or `too-large`. */
  reason: string;
  /** The first field that broke a rule, when the reason concerns one. */
  field?: string;
}>;

/** What an upsert or a delete did: applied, unchanged (a retry) or stale (a newer change came first), or why it was rejected. */
export type Outcome =
  | Readonly<{ outcome: "applied" | "unchanged" | "stale"; version: number }>
  | Rejection;

export type UpsertItem = Readonly<
  { action: "upsert"; external_id: string } & Profile
>;
export type DeleteItem = Readonly<{
  action: "delete";
  external_id: string;
  version: number;
}>;
export type BatchItem = UpsertItem | DeleteItem;

export type BatchResult = Readonly<{ external_id: string }> & Outcome;

export type ListedProfile = Readonly<{
  external_id: string;
  version: number;
  deleted: boolean;
}>;

/** The platform's copy of one profile: the public and AI-only values, and only the names of the private fields. */
export type ProfileCopy = Readonly<{
  external_id: string;
  version: number;
  synced_at: string;
  deleted: boolean;
  fields: Readonly<
    Record<
      string,
      Readonly<{ value: string; visibility: "public" | "ai_only" }>
    >
  >;
  private_fields: readonly string[];
  unflagged_fields: readonly string[];
}>;

export type SyncClient = Readonly<{
  /** The platform's API address, such as https://api.messenger.example. Plain http only to this machine: localhost, 127.0.0.1 or [::1]. */
  apiUrl: string;
  /** The site's secret key, `sk_...`. */
  secretKey: string;
  /** How many times to send a request in all before giving up. Default 6. */
  attempts?: number;
  /** The first wait before a retry, doubling each time up to 30 seconds. Default 1,000 ms. */
  firstDelayMs?: number;
}>;

/**
 * A request that did not go through: either the platform will not take it as it is (fix it
 * first), or no attempt got an answer (keep the change queued and send it later). `status` is
 * the last answer's, or 0 when none came.
 */
export class SyncError extends Error {
  readonly status: number;
  constructor(status: number, message: string) {
    super(message);
    this.status = status;
  }
}

/** The client's own wait for an answer: longer than the platform's 10 seconds per request. */
const REQUEST_TIMEOUT_MS = 15_000;
const MAX_DELAY_MS = 30_000;

/** Hosts where plain http never leaves this machine. */
const LOOPBACK = new Set(["localhost", "127.0.0.1", "[::1]"]);

/** The address of `path`. Refuses plain http to any other host, so the secret key never crosses a network unencrypted. */
function addressOf(client: SyncClient, path: string): URL {
  const url = new URL(path, client.apiUrl);
  const local = url.protocol === "http:" && LOOPBACK.has(url.hostname);
  if (url.protocol !== "https:" && !local)
    throw new Error(
      "The API address must start with https://, so the secret key is never sent unencrypted. Plain http works only to this machine (localhost, 127.0.0.1 or [::1]).",
    );
  return url;
}

function isObject(value: unknown): value is Readonly<Record<string, unknown>> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}

function wait(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

/** The wait before attempt `attempt + 1`: Retry-After when the platform sent one, otherwise doubling with jitter. */
function delay(
  client: SyncClient,
  attempt: number,
  retryAfter: string | null,
): number {
  const seconds =
    retryAfter === null ? Number.NaN : Number.parseInt(retryAfter, 10);
  if (Number.isFinite(seconds) && seconds >= 0) return seconds * 1000;
  const base = Math.min(
    MAX_DELAY_MS,
    (client.firstDelayMs ?? 1000) * 2 ** (attempt - 1),
  );
  return base / 2 + Math.random() * (base / 2);
}

/** The answer's JSON body, or undefined when it has none or it is not JSON. */
async function jsonOf(response: Response): Promise<unknown> {
  const text = await response.text();
  try {
    const parsed: unknown = JSON.parse(text);
    return parsed;
  } catch {
    return undefined;
  }
}

type Answer = Readonly<{ status: number; body: unknown }>;

/**
 * Sends one request, and the same request again after no answer, a 429 or a 5xx, waiting longer
 * each time. Returns the status and the parsed body of any other answer.
 */
async function send(
  client: SyncClient,
  method: string,
  path: string,
  body?: unknown,
  attempt = 1,
): Promise<Answer> {
  const address = addressOf(client, path);
  let response: Response | undefined;
  try {
    response = await fetch(address, {
      method,
      headers: {
        Authorization: `Bearer ${client.secretKey}`,
        Accept: "application/json",
        ...(body === undefined ? {} : { "Content-Type": "application/json" }),
      },
      ...(body === undefined ? {} : { body: JSON.stringify(body) }),
      // Never follow a redirect: it would send the secret key on. A 3xx comes back as an answer
      // to fix (see `failure`), not as an error that looks like no answer and gets retried.
      redirect: "manual",
      signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
    });
  } catch {
    response = undefined;
  }
  if (
    response !== undefined &&
    response.status !== 429 &&
    response.status < 500
  )
    return { status: response.status, body: await jsonOf(response) };
  const attempts = client.attempts ?? 6;
  if (attempt >= attempts) {
    const what = response === undefined ? "no answer" : String(response.status);
    throw new SyncError(
      response?.status ?? 0,
      `${method} ${path}: ${what} after ${String(attempts)} attempts. Keep the change queued and send it again later.`,
    );
  }
  await wait(
    delay(client, attempt, response?.headers.get("Retry-After") ?? null),
  );
  return send(client, method, path, body, attempt + 1);
}

function isAccepted(
  outcome: unknown,
): outcome is "applied" | "unchanged" | "stale" {
  return (
    outcome === "applied" || outcome === "unchanged" || outcome === "stale"
  );
}

function rejection(reason: string, field: unknown): Rejection {
  return typeof field === "string"
    ? { outcome: "rejected", reason, field }
    : { outcome: "rejected", reason };
}

/** The problem type of a 422 that names a rejection, such as `/problems/version-reused`. */
const REJECTION = /^\/problems\/([a-z-]+)$/;

/** An upsert's or a delete's answer: its outcome, or its rejection, or an error for a request to fix. */
function outcomeOf(method: string, path: string, answer: Answer): Outcome {
  const { status, body } = answer;
  if (status === 200 && isObject(body)) {
    const { outcome, version } = body;
    if (isAccepted(outcome) && typeof version === "number")
      return { outcome, version };
  }
  if (status === 422 && isObject(body) && typeof body["type"] === "string") {
    const reason = REJECTION.exec(body["type"])?.[1];
    if (reason !== undefined) return rejection(reason, body["field"]);
  }
  throw failure(method, path, status);
}

function failure(method: string, path: string, status: number): SyncError {
  if (status === 401)
    return new SyncError(
      status,
      `${method} ${path}: the platform refused the secret key. Check it, or rotate it in the partner console.`,
    );
  if (status >= 300 && status < 400)
    return new SyncError(
      status,
      `${method} ${path}: the answer was a redirect, which this client never follows, so the secret key went nowhere else. Set the API address to the one the platform team gave you.`,
    );
  return new SyncError(
    status,
    `${method} ${path}: the platform answered ${String(status)}. Fix the request before sending it again.`,
  );
}

function profilePath(externalId: string): string {
  return `/v1/profiles/${encodeURIComponent(externalId)}`;
}

/** Creates or replaces one profile. */
export async function putProfile(
  client: SyncClient,
  externalId: string,
  profile: Profile,
): Promise<Outcome> {
  const path = profilePath(externalId);
  return outcomeOf("PUT", path, await send(client, "PUT", path, profile));
}

/** Deletes one profile. `version` is higher than the last upsert's: a delete is a change too. */
export async function deleteProfile(
  client: SyncClient,
  externalId: string,
  version: number,
): Promise<Outcome> {
  const path = `${profilePath(externalId)}?version=${String(version)}`;
  return outcomeOf("DELETE", path, await send(client, "DELETE", path));
}

/** The most items one batch takes. */
export const BATCH_SIZE = 100;

/**
 * Sends up to 100 upserts and deletes in one request; one result each, in the same order. `total`
 * is how many profiles your site lists in all, for the console: send it during an import.
 */
export async function sendBatch(
  client: SyncClient,
  items: readonly BatchItem[],
  total?: number,
): Promise<readonly BatchResult[]> {
  if (items.length === 0 || items.length > BATCH_SIZE)
    throw new Error(`A batch holds 1 to ${String(BATCH_SIZE)} items.`);
  const path = "/v1/profiles/batch";
  const { status, body } = await send(client, "POST", path, {
    items,
    ...(total === undefined ? {} : { total }),
  });
  const results = isObject(body) ? body["results"] : undefined;
  if (
    status !== 200 ||
    !Array.isArray(results) ||
    results.length !== items.length
  )
    throw failure("POST", path, status);
  return items.map((item, index) => {
    const result: unknown = results[index];
    if (!isObject(result)) throw failure("POST", path, status);
    const { outcome, version, reason, field } = result;
    if (isAccepted(outcome) && typeof version === "number")
      return { external_id: item.external_id, outcome, version };
    if (outcome === "rejected" && typeof reason === "string")
      return { external_id: item.external_id, ...rejection(reason, field) };
    throw failure("POST", path, status);
  });
}

/**
 * Each batch's results, one batch at a time: `yield` waits for a batch's answer, and the next
 * batch goes out only when the loop asks for it.
 */
async function* batchesOf(
  client: SyncClient,
  items: readonly BatchItem[],
  total: number | undefined,
): AsyncGenerator<readonly BatchResult[]> {
  for (let start = 0; start < items.length; start += BATCH_SIZE)
    yield sendBatch(client, items.slice(start, start + BATCH_SIZE), total);
}

/** Sends any number of changes as batches of 100, one batch after the other. */
export async function sendInBatches(
  client: SyncClient,
  items: readonly BatchItem[],
  total?: number,
): Promise<readonly BatchResult[]> {
  const results: BatchResult[] = [];
  for await (const batch of batchesOf(client, items, total))
    for (const result of batch) results.push(result);
  return results;
}

/** One page of the platform's list, and the `after` of the next page, if there is one. */
async function readPage(
  client: SyncClient,
  pageSize: number,
  after: string | undefined,
): Promise<Readonly<{ profiles: readonly ListedProfile[]; next?: string }>> {
  const query = new URLSearchParams({
    limit: String(pageSize),
    ...(after === undefined ? {} : { after }),
  });
  const path = `/v1/profiles?${query.toString()}`;
  const { status, body } = await send(client, "GET", path);
  const listed = isObject(body) ? body["profiles"] : undefined;
  if (status !== 200 || !isObject(body) || !Array.isArray(listed))
    throw failure("GET", path, status);
  const profiles = listed.flatMap((profile: unknown) => {
    if (!isObject(profile)) return [];
    const { external_id, version, deleted } = profile;
    return typeof external_id === "string" &&
      typeof version === "number" &&
      typeof deleted === "boolean"
      ? [{ external_id, version, deleted }]
      : [];
  });
  const next = body["next"];
  return typeof next === "string" ? { profiles, next } : { profiles };
}

/**
 * The platform's list for the site, a page at a time, for `for await`: each page's `next` asks
 * for the page after it, and the last page has no `next`. Reading the list keeps the site's copy
 * fresh for the AI assistant.
 */
export function pagesOf(
  client: SyncClient,
  pageSize = 1000,
): AsyncIterable<readonly ListedProfile[]> {
  return {
    [Symbol.asyncIterator]: () => {
      let after: string | undefined;
      let finished = false;
      return {
        next: async (): Promise<
          IteratorResult<readonly ListedProfile[], undefined>
        > => {
          if (finished) return { done: true, value: undefined };
          const page = await readPage(client, pageSize, after);
          after = page.next;
          finished = after === undefined;
          return { done: false, value: page.profiles };
        },
      };
    },
  };
}

/** Every profile the platform holds for the site, deleted ones included, from every page. */
export async function listProfiles(
  client: SyncClient,
  pageSize = 1000,
): Promise<readonly ListedProfile[]> {
  const profiles: ListedProfile[] = [];
  for await (const page of pagesOf(client, pageSize))
    for (const profile of page) profiles.push(profile);
  return profiles;
}

/**
 * What the reconciliation sends (contracts/profile-sync.md, "Reconciliation"), comparing your
 * site's list with the platform's as sets, by external id:
 * - `behind`: your profiles the platform does not list, or holds at a lower version, deleted or
 *   not. Send them again. One it holds at your version or higher would only come back
 *   `unchanged` or `stale`.
 * - `gone`: profiles the platform holds, not deleted, that your site no longer lists. Delete each
 *   with a version higher than the one listed.
 */
export function reconcile(
  site: readonly UpsertItem[],
  held: readonly ListedProfile[],
): Readonly<{ behind: readonly UpsertItem[]; gone: readonly ListedProfile[] }> {
  const platform = new Map(
    held.map((profile) => [profile.external_id, profile]),
  );
  const listed = new Set(site.map((item) => item.external_id));
  return {
    behind: site.filter((item) => {
      const copy = platform.get(item.external_id);
      return copy === undefined || copy.version < item.version;
    }),
    gone: held.filter(
      (profile) => !profile.deleted && !listed.has(profile.external_id),
    ),
  };
}

/** The platform's copy of one profile, to check what arrived; undefined when it never had it. */
export async function getProfile(
  client: SyncClient,
  externalId: string,
): Promise<ProfileCopy | undefined> {
  const path = profilePath(externalId);
  const { status, body } = await send(client, "GET", path);
  if (status === 404) return undefined;
  if (status !== 200 || !isObject(body)) throw failure("GET", path, status);
  const {
    external_id,
    version,
    synced_at,
    deleted,
    fields,
    private_fields,
    unflagged_fields,
  } = body;
  if (
    typeof external_id !== "string" ||
    typeof version !== "number" ||
    typeof synced_at !== "string" ||
    typeof deleted !== "boolean" ||
    !isObject(fields) ||
    !Array.isArray(private_fields) ||
    !Array.isArray(unflagged_fields)
  ) {
    throw failure("GET", path, status);
  }
  const kept: Record<
    string,
    Readonly<{ value: string; visibility: "public" | "ai_only" }>
  > = {};
  for (const [name, copied] of Object.entries(fields)) {
    if (!isObject(copied)) continue;
    const { value, visibility } = copied;
    if (
      typeof value === "string" &&
      (visibility === "public" || visibility === "ai_only")
    )
      kept[name] = { value, visibility };
  }
  return {
    external_id,
    version,
    synced_at,
    deleted,
    fields: kept,
    private_fields: strings(private_fields),
    unflagged_fields: strings(unflagged_fields),
  };
}

function strings(list: readonly unknown[]): readonly string[] {
  return list.filter((item): item is string => typeof item === "string");
}