CRM example

cd examples/starters
AGENTDB_WORKSPACE=ws_... AGENTDB_TOKEN=<owner key> node setup.mjs --seed --agent crm-app --role editor

const API = (process.env.AGENTDB_API_URL ?? "https://app.agentdb.link").replace(/\/+$/, "");
const WORKSPACE = process.env.AGENTDB_WORKSPACE ?? "";
const TOKEN = process.env.AGENTDB_TOKEN ?? ""; // an agent key with role editor

export interface AgentdbRecord<T> {
  id: string;
  object: string;
  version: number;
  created_at: string;
  updated_at: string;
  data: T;
}

export class AgentdbError extends Error {
  status: number;
  code: string | undefined;
  constructor(status: number, code: string | undefined, message: string) {
    super(message);
    this.status = status;
    this.code = code;
  }
}

export async function agentdb<T>(method: string, path: string, body?: unknown, idempotencyKey?: string): Promise<T> {
  const res = await fetch(`${API}/v1/workspaces/${WORKSPACE}${path}`, {
    method,
    headers: {
      authorization: `Bearer ${TOKEN}`,
      "content-type": "application/json",
      ...(idempotencyKey ? { "idempotency-key": idempotencyKey } : {}),
    },
    body: body === undefined ? undefined : JSON.stringify(body),
  });
  const json = (await res.json()) as { data?: T; error?: { code?: string; message?: string } };
  if (!res.ok) throw new AgentdbError(res.status, json.error?.code, json.error?.message ?? `AgentDB answered ${res.status}`);
  return json.data as T;
}

final class AgentDb
{
    public function __construct(
        private readonly string $apiUrl,
        private readonly string $workspace,
        private readonly string $token,
    ) {
    }

    public static function fromEnv(): self
    {
        $workspace = getenv('AGENTDB_WORKSPACE') ?: '';
        $token = getenv('AGENTDB_TOKEN') ?: '';
        if ($workspace === '' || $token === '') {
            throw new RuntimeException('Set AGENTDB_WORKSPACE and AGENTDB_TOKEN (see .env.example).');
        }
        return new self(rtrim(getenv('AGENTDB_API_URL') ?: 'https://app.agentdb.link', '/'), $workspace, $token);
    }

    /** @return array<string, mixed> the "data" part of the answer */
    public function call(string $method, string $path, ?array $body = null, ?string $idempotencyKey = null): array
    {
        $headers = ["Authorization: Bearer {$this->token}", 'Content-Type: application/json', 'Accept: application/json'];
        if ($idempotencyKey !== null) {
            $headers[] = "Idempotency-Key: {$idempotencyKey}";
        }
        $context = stream_context_create(['http' => [
            'method' => $method,
            'header' => implode("\r\n", $headers),
            'content' => $body === null ? '' : json_encode($body, JSON_THROW_ON_ERROR),
            'ignore_errors' => true, // read the JSON error body on 4xx/5xx too
            'timeout' => 15,
        ]]);
        $raw = @file_get_contents("{$this->apiUrl}/v1/workspaces/{$this->workspace}{$path}", false, $context);
        if ($raw === false) {
            throw new AgentDbException(0, null, 'Could not reach AgentDB.');
        }
        $status = 0;
        // PHP 8.4+ has http_get_last_response_headers(); older versions fill $http_response_header.
        $responseHeaders = function_exists('http_get_last_response_headers') ? http_get_last_response_headers() : ($http_response_header ?? []);
        foreach ($responseHeaders ?? [] as $line) {
            if (preg_match('#^HTTP/\S+ (\d{3})#', $line, $m)) {
                $status = (int) $m[1]; // the last status line wins (after redirects)
            }
        }
        $json = json_decode($raw, true);
        if ($status < 200 || $status >= 300 || !is_array($json)) {
            throw new AgentDbException($status, $json['error']['code'] ?? null, $json['error']['message'] ?? "AgentDB answered {$status}");
        }
        return $json['data'];
    }
}

async function listContacts(q: string, companyId: string | null) {
  if (companyId) {
    // Contacts of one company: follow the works_at links into the company.
    const rel = await agentdb<Related<Contact>>("GET", `/records/${companyId}/related?relationship=works_at&direction=incoming&limit=100`);
    const needle = q.toLowerCase();
    return rel.records.filter((r) => !needle || r.data.name.toLowerCase().includes(needle)).map(contactOut);
  }
  const res = await agentdb<QueryResult<Contact>>("POST", "/query", {
    object: "contact",
    where: { all: q ? [{ field: "name", op: "contains", value: q }] : [] },
    sort: [{ field: "name", order: "asc" }],
    limit: 100,
  });
  return res.records.map(contactOut);
}

function contacts_page(AgentDb $db): void
{
    $q = mb_substr(trim((string) ($_GET['q'] ?? '')), 0, 100);
    $result = $db->call('POST', '/query', [
        'object' => 'contact',
        'where' => ['all' => $q === '' ? [] : [['field' => 'name', 'op' => 'contains', 'value' => $q]]],
        'sort' => [['field' => 'name', 'order' => 'asc']],
        'limit' => 100,
    ]);
    $rows = '';
    foreach ($result['records'] as $r) {
        $rows .= '<li><a href="/contact?id=' . e($r['id']) . '">' . e($r['data']['name']) . '</a> <span class="muted">' . e($r['data']['email'] ?? '') . '</span></li>';
    }
    $empty = $q === '' ? 'No contacts yet. <a href="/contacts/new">Add the first one</a>.' : 'No contacts found for “' . e($q) . '”.';
    page('Contacts', '<form method="get" role="search"><label>Search <input type="search" name="q" value="' . e($q) . '" maxlength="100"></label> <button>Search</button></form>'
        . ($rows === '' ? "<p class=\"empty\">{$empty}</p>" : "<ul class=\"list\">{$rows}</ul>")
        . '<p><a class="button" href="/contacts/new">New contact</a></p>');
}

async function createContact(body: Record<string, unknown>, submitId: string) {
  const data = contactInput(body);
  const companyId = recordId(body.company_id);
  const rec = await agentdb<AgentdbRecord<Contact>>(
    "POST",
    "/records",
    // The record and its link are written together, in one call.
    { object: "contact", data, links: companyId ? [{ relationship: "works_at", target_record_id: companyId }] : [] },
    `crm-ts.contact.${submitId}`,
  );
  return contactOut(rec);
}

async function updateContact(id: string, body: Record<string, unknown>) {
  const version = Number(body.version);
  if (!Number.isInteger(version) || version < 1) throw new HttpError(400, "The edit form is missing the record version.");
  const data = contactInput(body);
  // Optimistic concurrency: AgentDB refuses the write with VERSION_CONFLICT if someone saved first.
  const rec = await agentdb<AgentdbRecord<Contact>>("PATCH", `/records/${id}`, {
    version,
    data: { name: data.name, email: data.email ?? null, role: data.role ?? null },
  });
  if ("company_id" in body) {
    // Move the contact: remove the old works_at link, then add the new one.
    const wanted = recordId(body.company_id);
    const current = await companyOf(id);
    if (current && current.id !== wanted) await agentdb("DELETE", `/links/${current.link_id}`);
    if (wanted && current?.id !== wanted) await agentdb("POST", "/links", { relationship: "works_at", source_record_id: id, target_record_id: wanted });
  }
  return contactOut(rec);
}

function contact_form_page(AgentDb $db, ?string $id, string $method): void
{
    $error = null;
    $status = 200;
    $current = $id === null ? null : $db->call('GET', "/records/{$id}");
    $company = $id === null ? null : company_of($db, $id);
    $values = [
        'name' => $current['data']['name'] ?? '',
        'email' => $current['data']['email'] ?? '',
        'role' => $current['data']['role'] ?? '',
        'company_id' => $company['id'] ?? '',
        'version' => (string) ($current['version'] ?? ''),
    ];

    if ($method === 'POST') {
        $values = ['name' => field('name', 200), 'email' => field('email', 254), 'role' => field('role', 200),
            'company_id' => record_id($_POST['company_id'] ?? null) ?? '', 'version' => field('version', 20)];
        $data = ['name' => $values['name'], 'email' => $values['email'] ?: null, 'role' => $values['role'] ?: null];
        try {
            if ($values['name'] === '') {
                throw new AgentDbException(422, 'VALIDATION_ERROR', 'A contact needs a name.');
            }
            if ($id === null) {
                $links = $values['company_id'] === '' ? [] : [['relationship' => 'works_at', 'target_record_id' => $values['company_id']]];
                $rec = $db->call('POST', '/records', ['object' => 'contact', 'data' => array_filter($data), 'links' => $links],
                    'crm-php.' . hash('sha256', session_id() . $_SESSION['csrf'] . $values['name'] . $values['email']));
                $id = $rec['id'];
            } else {
                // The version from the form: AgentDB refuses the write if someone saved in between.
                $db->call('PATCH', "/records/{$id}", ['version' => (int) $values['version'], 'data' => $data]);
                if (($company['id'] ?? '') !== $values['company_id']) {
                    if ($company !== null) {
                        $db->call('DELETE', "/links/{$company['link_id']}");
                    }
                    if ($values['company_id'] !== '') {
                        $db->call('POST', '/links', ['relationship' => 'works_at', 'source_record_id' => $id, 'target_record_id' => $values['company_id']]);
                    }
                }
            }
            header('Location: /contact?id=' . rawurlencode($id) . '&saved=1', true, 303);
            return;
        } catch (AgentDbException $e) {
            [$status, $error] = match (true) {
                $e->errorCode === 'VERSION_CONFLICT' => [409, 'Someone changed it first. Reload the page to see the latest version, then edit again.'],
                in_array($e->status, [400, 422], true) => [422, $e->getMessage()],
                default => throw $e,
            };
        }
    }
    http_response_code($status);
    page($id === null ? 'New contact' : 'Edit contact', contact_form($values, all_companies($db), $company, $error, isset($_GET['saved'])));
}

function errorAnswer(e: unknown): [number, string] {
  if (e instanceof HttpError) return [e.status, e.message];
  if (e instanceof AgentdbError) {
    if (e.code === "VERSION_CONFLICT") return [409, "Someone changed this contact first. Reload it and try again."];
    if (e.code === "NOT_FOUND") return [404, "Not found."];
    if (e.status === 400 || e.status === 422) return [400, e.message]; // validation errors are safe to show
    console.error("AgentDB call failed:", e.status, e.code, e.message);
  } else console.error(e);
  return [502, "The data service is not available. Try again in a moment."];
}

/** One place for loading, empty and error states: the list is replaced, the status line says why. */
async function show<T>(list: HTMLElement, status: HTMLElement, load: () => Promise<T[]>, row: (item: T) => HTMLElement, empty: string) {
  status.textContent = "Loading…";
  status.className = "status";
  list.setAttribute("aria-busy", "true");
  try {
    const items = await load();
    list.replaceChildren(...items.map(row));
    status.textContent = items.length ? "" : empty;
  } catch (e) {
    list.replaceChildren();
    status.textContent = `Could not load: ${(e as Error).message}`;
    status.className = "status error";
  } finally {
    list.removeAttribute("aria-busy");
  }
}

/** Calls the backend and turns an error answer into an Error with the backend's message. */
async function api<T>(method: string, path: string, body?: unknown): Promise<T> {
  const res = await fetch(path, {
    method,
    headers: body === undefined ? {} : { "content-type": "application/json", "x-submit-id": crypto.randomUUID() },
    body: body === undefined ? undefined : JSON.stringify(body),
  });
  const json = (await res.json().catch(() => ({}))) as T & { error?: string };
  if (!res.ok) throw new Error(json.error ?? `Request failed (${res.status}).`);
  return json;
}

/** One loop for the whole server: wait for changes, then send a fresh snapshot to every viewer. */
async function follow() {
  // Start at the newest change: the first snapshot already holds everything before it.
  let after = (await agentdb("GET", "/changes?limit=1")).marker;
  current = await snapshot();
  for (;;) {
    try {
      // wait=1: AgentDB answers as soon as something changes, or after about 25 seconds.
      const feed = await agentdb("GET", `/changes?after=${after}&limit=100&wait=1`, undefined, AbortSignal.timeout(40_000));
      if (feed.changes.length === 0) continue;
      after = feed.next_after ?? feed.changes.at(-1).seq;
      if (feed.changes.some((c) => c.action.startsWith("record.") || c.action.startsWith("schema."))) {
        current = await snapshot();
        broadcast("contacts", current);
      }
    } catch (e) {
      console.error("Change feed:", e.code ?? e.name, e.message);
      broadcast("problem", { message: "Live updates paused. Retrying…" });
      await new Promise((r) => setTimeout(r, 5_000)); // back off, then resume where we left off
    }
  }
}

function events(req, res) {
  res.writeHead(200, {
    "content-type": "text/event-stream; charset=utf-8",
    "cache-control": "no-store",
    "x-accel-buffering": "no", // tells nginx not to buffer this response
    ...SECURITY_HEADERS,
  });
  res.write("retry: 3000\n\n"); // the browser reconnects after 3 s if the connection drops
  if (current) res.write(`event: contacts\ndata: ${JSON.stringify(current)}\n\n`);
  viewers.add(res);
  // A comment line every 20 s keeps proxies from closing an idle connection.
  const ping = setInterval(() => res.write(": ping\n\n"), 20_000);
  req.on("close", () => {
    clearInterval(ping);
    viewers.delete(res);
  });
}

const source = new EventSource("/events");

source.addEventListener("open", () => {
  state.textContent = "Live";
  state.className = "state live";
});

source.addEventListener("contacts", (e) => {
  const data = JSON.parse(e.data);
  document.getElementById("count").textContent = String(data.count);
  document.getElementById("at").textContent = new Date(data.at).toLocaleTimeString();
  document.getElementById("empty").hidden = data.latest.length > 0;
  document.getElementById("latest").replaceChildren(
    ...data.latest.map((c) => {
      const li = document.createElement("li");
      li.textContent = c.email ? `${c.name} (${c.email})` : c.name; // text, never HTML
      return li;
    }),
  );
  state.textContent = "Live";
  state.className = "state live";
});

source.addEventListener("problem", (e) => {
  state.textContent = JSON.parse(e.data).message;
  state.className = "state warn";
});

// EventSource reconnects by itself; this only tells the viewer.
source.addEventListener("error", () => {
  state.textContent = "Connection lost. Reconnecting…";
  state.className = "state warn";
});