Concurrency and optimistic locking

import { randomUUID } from "node:crypto";

const base = process.env.FILMO_BASE_URL ?? "https://app.filmo.io";
const auth = { Authorization: `Bearer ${process.env.FILMO_API_KEY}` };
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

// Adds a phone number without overwriting a change someone else made in between.
export async function addPhone(contactId, number) {
  const url = base + "/api/v1/contacts/" + contactId;
  for (let attempt = 0; attempt < 5; attempt++) {
    const read = await fetch(url, { headers: auth });
    if (!read.ok) throw new Error("Read failed: HTTP " + read.status);
    const etag = read.headers.get("ETag");
    const { data } = await read.json();
    if (data.phoneNumbers.some((p) => p.number === number)) return data;

    // phoneNumbers replaces the whole list, so send the existing entries back too.
    const phoneNumbers = data.phoneNumbers.map((p) => ({
      number: p.number,
      isPrimary: p.isPrimary,
      ...(p.label ? { label: p.label } : {}),
      ...(p.countryCode ? { countryCode: p.countryCode } : {}),
    }));
    phoneNumbers.push({ number, isPrimary: false });

    const write = await fetch(url, {
      method: "PATCH",
      headers: {
        ...auth,
        "Content-Type": "application/json",
        "If-Match": etag,
        // A new key per attempt: each attempt sends a different body.
        "Idempotency-Key": randomUUID(),
      },
      body: JSON.stringify({ phoneNumbers }),
    });
    if (write.ok) return (await write.json()).data;

    const failure = await write.text();
    const cause = write.status === 409 ? JSON.parse(failure).errors?.[0]?.code : null;
    // 412: the contact changed since the read. 409 "lock_timeout": another write held the lock for 5 seconds.
    if (write.status === 412 || cause === "lock_timeout") {
      await sleep(100 * 2 ** attempt);
      continue;
    }
    throw new Error("Write failed: HTTP " + write.status + " " + failure);
  }
  throw new Error("Too many concurrent changes; try again later");
}

const contact = await addPhone(process.env.CONTACT_ID, "+34 600 000 000");
console.log(contact.phoneNumbers.length, "phone numbers");

import os
import time
import uuid

import requests

BASE = os.environ.get("FILMO_BASE_URL", "https://app.filmo.io")
AUTH = {"Authorization": f"Bearer {os.environ['FILMO_API_KEY']}"}


def add_phone(contact_id, number):
    """Adds a phone number without overwriting a change made in between."""
    url = BASE + "/api/v1/contacts/" + contact_id
    for attempt in range(5):
        read = requests.get(url, headers=AUTH, timeout=30)
        read.raise_for_status()
        etag = read.headers["ETag"]
        data = read.json()["data"]
        if any(p["number"] == number for p in data["phoneNumbers"]):
            return data

        # phoneNumbers replaces the whole list, so send the existing entries back too.
        phone_numbers = []
        for p in data["phoneNumbers"]:
            entry = {"number": p["number"], "isPrimary": p["isPrimary"]}
            if p.get("label"):
                entry["label"] = p["label"]
            if p.get("countryCode"):
                entry["countryCode"] = p["countryCode"]
            phone_numbers.append(entry)
        phone_numbers.append({"number": number, "isPrimary": False})

        write = requests.patch(
            url,
            json={"phoneNumbers": phone_numbers},
            headers={
                **AUTH,
                "If-Match": etag,
                # A new key per attempt: each attempt sends a different body.
                "Idempotency-Key": str(uuid.uuid4()),
            },
            timeout=30,
        )
        if write.ok:
            return write.json()["data"]

        # 412: the contact changed since the read. 409 "lock_timeout": another write held the lock for 5 seconds.
        cause = None
        if write.status_code == 409:
            cause = ((write.json().get("errors") or [{}])[0]).get("code")
        if write.status_code == 412 or cause == "lock_timeout":
            time.sleep(0.1 * 2**attempt)
            continue
        raise RuntimeError(f"Write failed: HTTP {write.status_code} {write.text}")
    raise RuntimeError("Too many concurrent changes; try again later")


contact = add_phone(os.environ["CONTACT_ID"], "+34 600 000 000")
print(len(contact["phoneNumbers"]), "phone numbers")