storebound/cli/lib/upsert.ts

// Idempotent writes against the stores. Every create in storebound goes through here or follows the same rules
// (docs/idempotency.md):
//   1. Look the thing up by its stable key first (product id, locale, version string, display type, name) and update or
//      leave it instead of creating it again.
//   2. Right before a create, look again with a fresh GET (findOrCreate), and ask a journal of recent creates
//      (.storebound/created.json) for an id the store's list may not show yet: a direct GET by id is not subject to the
//      minute of list lag right after a create.
//   3. A create answered with 409 / ALREADY_EXISTS is "found": look it up and use it, never create again (createOnce).
//   4. A create whose outcome is unknown (timeout, connection reset, 5xx on a POST) is never sent blindly again: look
//      first; only when the store itself refuses duplicates of that key (unique) is a second attempt safe (its 409
//      then finds the first one). Otherwise stop with a message: re-running reads the store and converges.
// Retries of 429 / 5xx in the API clients are only for requests that end the same when sent twice (retrySafe).
import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { log, warn } from './log';
import * as rt from './rt';

/** A write whose outcome is unknown: it may or may not have been applied (timeout, connection reset, 5xx answer to a
 *  POST). The API clients throw this instead of retrying such a request. */
export class AmbiguousWriteError extends Error {
  readonly ambiguous = true;
  constructor(message: string, readonly status: number | null = null) { super(message); this.name = 'AmbiguousWriteError'; }
}

export const isAmbiguous = (e: unknown): boolean => e instanceof AmbiguousWriteError || (e as { ambiguous?: boolean } | null)?.ambiguous === true;

/**
 * The store refused a create because the thing already exists: Google ALREADY_EXISTS / 409, App Store Connect 409 with
 * a "duplicate / already / unique / in use / taken" detail (Apple also answers 409 for plain validation errors, which
 * are not conflicts), RevenueCat 409. Pure.
 */
export function isConflict(e: unknown): boolean {
  const s = String((e as Error)?.message ?? e);
  if (/ALREADY_EXISTS/.test(s)) return true;
  // "already added to another reviewSubmission": the item is held elsewhere, not a duplicate of this create
  if (/ITEM_PART_OF_ANOTHER_SUBMISSION/.test(s)) return false;
  if (!/(-> |: )409\b/.test(s)) return false;
  return /duplicate|already|unique|in use|taken|exists/i.test(s.replace(/^.*?409/, ''));
}

/** POSTs that end the same when sent twice: relationship adds, Google custom actions that set a state, and the App
 *  Store "create or replace" singletons (price schedules, availability). */
const IDEMPOTENT_POST: RegExp[] = [
  /\/relationships\//,
  /:(validate|convertRegionPrices|batchUpdateStates|activate|deactivate|query|batchGet|reply)(\?|$)/,
  /\/edits(\?|$)/, // a new edit; an unused one expires on its own and changes nothing
  /\/dataSafety(\?|$)/,
  /^\/v1\/(inAppPurchasePriceSchedules|appPriceSchedules|inAppPurchaseAvailabilities|subscriptionAvailabilities)(\?|$)/,
  /^\/v2\/appAvailabilities(\?|$)/,
];

/** Whether a request may be sent again after a 5xx or a network error. GET, HEAD, PUT, PATCH and DELETE set a state;
 *  a POST creates unless its path is a known idempotent action. `path` may be a full URL. Pure. */
export function retrySafe(method: string, path: string): boolean {
  const m = method.toUpperCase();
  if (m !== 'POST') return true;
  const p = path.replace(/^https:\/\/[^/]+/, '').replace(/^\/(androidpublisher|upload\/androidpublisher)\/v3\/applications\/[^/]+/, '');
  return IDEMPOTENT_POST.some(re => re.test(p));
}

/** Waits between re-checks (STOREBOUND_UPSERT_WAIT_MS; the test preload sets 0). */
export const upsertTiming = { waitMs: Number(process.env.STOREBOUND_UPSERT_WAIT_MS ?? 3000), tries: 4, sleep: (ms: number): Promise<void> => (ms > 0 ? rt.sleep(ms) : Promise.resolve()) };

export interface CreateOnce<T> {
  /** what is being created, for log lines and errors: `in-app purchase com.example.gems` */
  what: string;
  /** the create request; returns the created thing */
  create: () => Promise<T>;
  /** a fresh lookup by the stable key (GET with a filter, or the list), null when it is not there */
  find: () => Promise<T | null | undefined>;
  /** the store refuses a second one with this key (409): a create may then be tried again after an unknown outcome */
  unique?: boolean;
  /** log tag */
  tag?: string;
}

async function recheck<T>(o: CreateOnce<T>): Promise<T | null> {
  for (let i = 0; i < upsertTiming.tries; i++) {
    const v = await o.find();
    if (v) return v;
    if (i < upsertTiming.tries - 1) await upsertTiming.sleep(upsertTiming.waitMs * (i + 1));
  }
  return null;
}

/** One create, never two: a conflict finds the existing one, an unknown outcome looks before (maybe) trying again. */
export async function createOnce<T>(o: CreateOnce<T>, attempt = 1): Promise<{ value: T; created: boolean }> {
  try {
    return { value: await o.create(), created: true };
  } catch (e) {
    const msg = (e as Error).message;
    if (isConflict(e)) {
      const v = await recheck(o);
      if (v) { log(o.tag ?? 'upsert', `${o.what}: already exists, using it (not created again)`); return { value: v, created: false }; }
      throw new Error(`${o.what}: the store says it already exists but does not list it yet (its lists lag right after a create). Nothing was created twice; run the command again in a minute.\n${msg}`);
    }
    if (!isAmbiguous(e)) throw e;
    warn(o.tag ?? 'upsert', `${o.what}: no clear answer to the create (${msg.slice(0, 160)}); checking whether it exists before anything else`);
    const v = await recheck(o);
    if (v) { log(o.tag ?? 'upsert', `${o.what}: the create went through`); return { value: v, created: true }; }
    if (o.unique && attempt < 2) return createOnce(o, attempt + 1);
    throw new Error(`${o.what}: the create got no clear answer and the store does not list it (yet). It was not sent again, so nothing is created twice: run the command again in a minute (it reads the store first).\n${msg}`);
  }
}

// ---------------------------------------------------------------------------------------------- journal of creates

/** Creates of the last day, keyed by store + kind + stable key: { id, at }. Lives in .storebound/created.json. */
type Journal = Record<string, { id: string; at: string }>;
const JOURNAL_MS = 24 * 3600_000;
const journalPath = (root: string): string => join(root, '.storebound', 'created.json');

export function readJournal(root: string, now = Date.now()): Journal {
  try {
    const j = JSON.parse(readFileSync(journalPath(root), 'utf8')) as Journal;
    return Object.fromEntries(Object.entries(j).filter(([, v]) => v && now - Date.parse(v.at) < JOURNAL_MS));
  } catch { return {}; }
}

export function remember(root: string, key: string, id: string, now = new Date()): void {
  try {
    const j = readJournal(root, now.getTime());
    j[key] = { id, at: now.toISOString() };
    mkdirSync(join(root, '.storebound'), { recursive: true });
    const tmp = journalPath(root) + `.${process.pid}.tmp`;
    writeFileSync(tmp, JSON.stringify(j, null, 2) + '\n');
    renameSync(tmp, journalPath(root));
  } catch (e) { warn('upsert', `could not note the create in .storebound/created.json: ${(e as Error).message}`); }
}

export interface FindOrCreate<T> extends CreateOnce<T> {
  /** journal of recent creates: guards against the list lag right after a create of an earlier run */
  journal?: { root: string; key: string; byId: (id: string) => Promise<T | null | undefined>; idOf: (v: T) => string };
}

/** Looks with a fresh GET, then in the journal (direct GET by id), and only then creates (createOnce). */
export async function findOrCreate<T>(o: FindOrCreate<T>): Promise<{ value: T; created: boolean }> {
  const found = await o.find();
  if (found) return { value: found, created: false };
  if (o.journal && existsSync(join(o.journal.root, '.storebound'))) {
    const hit = readJournal(o.journal.root)[o.journal.key];
    if (hit) {
      const v = await o.journal.byId(hit.id).catch(() => null);
      if (v) { log(o.tag ?? 'upsert', `${o.what}: created ${hit.at.slice(11, 16)} UTC by an earlier run (${hit.id}); the store's list does not show it yet, using it`); return { value: v, created: false }; }
    }
  }
  const r = await createOnce(o);
  if (o.journal && r.created) remember(o.journal.root, o.journal.key, o.journal.idOf(r.value));
  return r;
}

/** Apple: GET of one resource by id, null on 404. For FindOrCreate.journal.byId. */
export async function getById(asc: (m: string, p: string) => Promise<{ data?: unknown }>, path: string): Promise<unknown | null> {
  try { return (await asc('GET', path)).data ?? null; } catch (e) { if (/-> 404/.test(String(e))) return null; throw e; }
}