CRM integration: adapter, field mapping, synchronisation

// Ядро ничего не знает про amoCRM: своя модель заявки и порт, который реализует адаптер
export type LeadRequest = {
  externalId: string; // id заявки на сайте — ключ идемпотентности
  name: string;
  phone: string;
  email?: string;
  comment?: string;
  source: "site-form" | "callback" | "landing";
};

export type LeadStatus = "new" | "in_progress" | "won" | "lost";

export type CrmLeadRef = {
  crmLeadId: number;
  crmContactId: number;
  contactReused: boolean;
};

export interface CrmPort {
  /* найти уже созданную сделку по id заявки — защита от дубля при повторе */
  findLeadByExternalId(externalId: string): Promise<number | null>;
  createLead(lead: LeadRequest): Promise<CrmLeadRef>;
}

export type Tokens = { accessToken: string; refreshToken: string; expiresAt: number };

// Хранилище токенов — своя БД или менеджер секретов, не файл в репозитории
export interface TokenStore {
  load(): Promise<Tokens>;
  save(tokens: Tokens): Promise<void>;
}

type OAuthApp = { clientId: string; clientSecret: string; redirectUri: string };

export class AmoTokenProvider {
  private refreshing: Promise<Tokens> | null = null;
  private baseUrl: string;
  private app: OAuthApp;
  private store: TokenStore;

  constructor(baseUrl: string, app: OAuthApp, store: TokenStore) {
    this.baseUrl = baseUrl;
    this.app = app;
    this.store = store;
  }

  async accessToken(): Promise<string> {
    const tokens = await this.store.load();
    if (tokens.expiresAt - 60_000 > Date.now()) return tokens.accessToken;
    return (await this.refresh()).accessToken;
  }

  // один обмен на процесс: refresh token одноразовый, параллельный второй обмен его «сожжёт»
  refresh(): Promise<Tokens> {
    this.refreshing ??= this.exchange().finally(() => {
      this.refreshing = null;
    });
    return this.refreshing;
  }

  private async exchange(): Promise<Tokens> {
    const { refreshToken } = await this.store.load();
    const res = await fetch(`${this.baseUrl}/oauth2/access_token`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        client_id: this.app.clientId,
        client_secret: this.app.clientSecret,
        grant_type: "refresh_token",
        refresh_token: refreshToken,
        redirect_uri: this.app.redirectUri,
      }),
      signal: AbortSignal.timeout(10_000),
    });
    if (!res.ok) throw new Error(`token refresh failed: HTTP ${res.status}`);
    const body = (await res.json()) as { access_token: string; refresh_token: string; expires_in: number };
    const tokens = {
      accessToken: body.access_token,
      refreshToken: body.refresh_token,
      expiresAt: Date.now() + body.expires_in * 1000,
    };
    await this.store.save(tokens); // сохранить до первого использования: старый refresh уже недействителен
    return tokens;
  }
}

export class AmoHttpError extends Error {
  status: number;
  body: string;
  constructor(status: number, body: string) {
    super(`amoCRM HTTP ${status}: ${body.slice(0, 300)}`);
    this.status = status;
    this.body = body;
  }
}

type TokenSource = { accessToken(): Promise<string>; refresh(): Promise<unknown> };

type Options = {
  baseUrl: string; // https://<subdomain>.amocrm.ru — адрес аккаунта, не общий домен
  tokens: TokenSource;
  minIntervalMs?: number; // лимит — 7 запросов в секунду на интеграцию: 150 мс между стартами
  maxRetries?: number;
  sleep?: (ms: number) => Promise<void>;
};

const wait = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));

export class AmoClient {
  private nextSlot = 0;
  private opts: Required<Options>;

  constructor(opts: Options) {
    this.opts = { minIntervalMs: 150, maxRetries: 3, sleep: wait, ...opts };
  }

  async request<T>(method: "GET" | "POST" | "PATCH", path: string, body?: unknown): Promise<T | null> {
    let refreshed = false;
    for (let attempt = 0; ; attempt++) {
      await this.throttle();
      const res = await fetch(this.opts.baseUrl + path, {
        method,
        headers: {
          Authorization: `Bearer ${await this.opts.tokens.accessToken()}`,
          "Content-Type": "application/json",
        },
        body: body === undefined ? undefined : JSON.stringify(body),
        signal: AbortSignal.timeout(15_000),
      });
      if (res.status === 204) return null; // нет данных: тело пустое
      if (res.ok) return (await res.json()) as T;

      const text = await res.text();
      if (res.status === 401 && !refreshed) {
        refreshed = true; // access token истёк — обменять refresh и повторить один раз
        await this.opts.tokens.refresh();
        continue;
      }
      // 429 — запрос не выполнен, повтор безопасен; 5xx повторяем только для чтения
      const retryable = res.status === 429 || (res.status >= 500 && method === "GET");
      if (retryable && attempt < this.opts.maxRetries) {
        const retryAfter = Number(res.headers.get("retry-after"));
        await this.opts.sleep(retryAfter > 0 ? retryAfter * 1000 : 1000 * 2 ** attempt);
        continue;
      }
      throw new AmoHttpError(res.status, text);
    }
  }

  private async throttle() {
    const now = Date.now();
    const slot = Math.max(now, this.nextSlot);
    this.nextSlot = slot + this.opts.minIntervalMs;
    if (slot > now) await wait(slot - now);
  }
}

// Формы запросов и ответов amoCRM API v4 — по разделам «Сделки», «Контакты», «Поля»
export type AmoFieldValue = { value: string; enum_code?: string };
export type AmoCustomField = { field_id?: number; field_code?: string; values: AmoFieldValue[] };

export type AmoContactInput = { first_name: string; custom_fields_values: AmoCustomField[] };

export type AmoComplexLead = {
  name: string;
  pipeline_id: number;
  request_id: string;
  custom_fields_values: AmoCustomField[];
  tags_to_add: { name: string }[];
  _embedded: { contacts: [{ id: number } | AmoContactInput] };
};

export type AmoComplexResult = {
  id: number;
  contact_id: number | null;
  company_id: number | null;
  request_id: string[];
  merged: boolean;
};

export type AmoContact = { id: number; name: string; custom_fields_values: AmoCustomField[] | null };
export type AmoLead = {
  id: number;
  status_id: number;
  pipeline_id: number;
  updated_at: number;
  custom_fields_values: AmoCustomField[] | null;
};

export type AmoPage<K extends string, T> = {
  _page?: number;
  _links: { next?: { href: string } };
  _embedded: Record<K, T[]>;
};

// id полей и воронки отличаются в каждом аккаунте: берите их из GET /api/v4/leads/custom_fields
export type AmoMapping = {
  pipelineId: number;
  firstStatusId: number;
  fields: { externalId: number; comment: number; source: number };
  tag: string;
};

import type { LeadRequest, LeadStatus } from "../core/lead.ts";
import { normalizeEmail, normalizePhone } from "../core/normalize.ts";
import type { AmoComplexLead, AmoCustomField, AmoMapping } from "./types.ts";

const SOURCE_LABEL: Record<LeadRequest["source"], string> = {
  "site-form": "Форма на сайте",
  callback: "Обратный звонок",
  landing: "Лендинг",
};

const text = (field_id: number, value: string | undefined): AmoCustomField[] =>
  value?.trim() ? [{ field_id, values: [{ value: value.trim() }] }] : [];

export function toAmoComplexLead(
  lead: LeadRequest,
  map: AmoMapping,
  existingContactId: number | null,
): AmoComplexLead {
  const phone = normalizePhone(lead.phone);
  if (!phone) throw new Error(`invalid phone for ${lead.externalId}`);
  const email = normalizeEmail(lead.email);

  const contact =
    existingContactId !== null
      ? { id: existingContactId }
      : {
          first_name: lead.name.trim(),
          custom_fields_values: [
            { field_code: "PHONE", values: [{ value: phone, enum_code: "MOB" }] },
            ...(email ? [{ field_code: "EMAIL", values: [{ value: email, enum_code: "WORK" }] }] : []),
          ],
        };

  return {
    name: `Заявка ${lead.externalId}`,
    pipeline_id: map.pipelineId,
    request_id: lead.externalId,
    custom_fields_values: [
      ...text(map.fields.externalId, lead.externalId),
      ...text(map.fields.comment, lead.comment),
      ...text(map.fields.source, SOURCE_LABEL[lead.source]),
    ],
    tags_to_add: [{ name: map.tag }],
    _embedded: { contacts: [contact] },
  };
}

// 142 и 143 — системные статусы любой воронки amoCRM: «успешно реализовано» и «закрыто и не реализовано»
export function toLeadStatus(statusId: number, map: AmoMapping): LeadStatus {
  if (statusId === 142) return "won";
  if (statusId === 143) return "lost";
  return statusId === map.firstStatusId ? "new" : "in_progress";
}

export function readText(fields: AmoCustomField[] | null, fieldId: number): string | null {
  return fields?.find((f) => f.field_id === fieldId)?.values[0]?.value ?? null;
}

// Нормализация ключей дедупликации. Правило для плана нумерации +7 (Казахстан, Россия);
// для международных номеров берите библиотеку уровня libphonenumber
export function normalizePhone(raw: string): string | null {
  const digits = raw.replace(/\D/g, "");
  if (digits.length === 11 && digits.startsWith("8")) return `+7${digits.slice(1)}`;
  if (digits.length === 11 && digits.startsWith("7")) return `+${digits}`;
  if (digits.length === 10) return `+7${digits}`;
  return null;
}

export function normalizeEmail(raw: string | undefined): string | null {
  const email = raw?.trim().toLowerCase();
  return email && /^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email) ? email : null;
}

import { normalizeEmail, normalizePhone } from "../core/normalize.ts";
import type { AmoContact } from "./types.ts";

const values = (c: AmoContact, code: string) =>
  (c.custom_fields_values ?? []).filter((f) => f.field_code === code).flatMap((f) => f.values.map((v) => v.value));

// query в amoCRM — полнотекстовый поиск: ответ перепроверяем точным сравнением нормализованных ключей
export function pickDuplicate(candidates: AmoContact[], phone: string | null, email: string | null): number | null {
  const matches = candidates.filter(
    (c) =>
      (phone !== null && values(c, "PHONE").some((v) => normalizePhone(v) === phone)) ||
      (email !== null && values(c, "EMAIL").some((v) => normalizeEmail(v) === email)),
  );
  // несколько совпадений — берём самый старый контакт (минимальный id), остальные склеивает менеджер
  return matches.length ? Math.min(...matches.map((c) => c.id)) : null;
}

import type { CrmLeadRef, CrmPort, LeadRequest } from "../core/lead.ts";
import { normalizeEmail, normalizePhone } from "../core/normalize.ts";
import type { AmoClient } from "./client.ts";
import { pickDuplicate } from "./dedup.ts";
import { readText, toAmoComplexLead } from "./mapping.ts";
import type { AmoComplexResult, AmoContact, AmoLead, AmoMapping, AmoPage } from "./types.ts";

export class AmoCrmAdapter implements CrmPort {
  private client: AmoClient;
  private map: AmoMapping;

  constructor(client: AmoClient, map: AmoMapping) {
    this.client = client;
    this.map = map;
  }

  async findLeadByExternalId(externalId: string): Promise<number | null> {
    const page = await this.client.request<AmoPage<"leads", AmoLead>>(
      "GET",
      `/api/v4/leads?query=${encodeURIComponent(externalId)}&limit=50`,
    );
    const hit = page?._embedded.leads.find(
      (l) => readText(l.custom_fields_values, this.map.fields.externalId) === externalId,
    );
    return hit?.id ?? null;
  }

  async createLead(lead: LeadRequest): Promise<CrmLeadRef> {
    const contactId = await this.findContact(normalizePhone(lead.phone), normalizeEmail(lead.email));
    const result = await this.client.request<AmoComplexResult[]>("POST", "/api/v4/leads/complex", [
      toAmoComplexLead(lead, this.map, contactId),
    ]);
    const created = result?.[0];
    if (!created) throw new Error(`amoCRM returned no lead for ${lead.externalId}`);
    return {
      crmLeadId: created.id,
      crmContactId: created.contact_id ?? contactId ?? 0,
      contactReused: contactId !== null || created.merged,
    };
  }

  private async findContact(phone: string | null, email: string | null): Promise<number | null> {
    for (const query of [phone, email]) {
      if (!query) continue;
      const page = await this.client.request<AmoPage<"contacts", AmoContact>>(
        "GET",
        `/api/v4/contacts?query=${encodeURIComponent(query)}&limit=50`,
      );
      const id = pickDuplicate(page?._embedded.contacts ?? [], phone, email);
      if (id !== null) return id;
    }
    return null;
  }
}

// Журнал обмена: по каждой заявке видно, дошла ли она до CRM, сколько было попыток и почему упала
export type ExchangeEntry = {
  externalId: string;
  status: "pending" | "synced" | "failed";
  attempts: number;
  crmLeadId?: number;
  lastError?: string;
  updatedAt: string;
};

export interface ExchangeLog {
  get(externalId: string): Promise<ExchangeEntry | null>;
  put(entry: ExchangeEntry): Promise<void>;
  failed(): Promise<ExchangeEntry[]>;
}

// Для тестов; в продакшене — таблица в своей БД с уникальным индексом по externalId
export class MemoryExchangeLog implements ExchangeLog {
  private entries = new Map<string, ExchangeEntry>();

  async get(externalId: string) {
    return this.entries.get(externalId) ?? null;
  }

  async put(entry: ExchangeEntry) {
    this.entries.set(entry.externalId, { ...entry });
  }

  async failed() {
    return [...this.entries.values()].filter((e) => e.status === "failed");
  }
}

import type { CrmPort, LeadRequest } from "./lead.ts";
import type { ExchangeEntry, ExchangeLog } from "./exchange-log.ts";

type Deps = { crm: CrmPort; log: ExchangeLog; now?: () => Date };

export async function submitLead(lead: LeadRequest, deps: Deps): Promise<ExchangeEntry> {
  const now = () => (deps.now?.() ?? new Date()).toISOString();
  const prev = await deps.log.get(lead.externalId);
  if (prev?.status === "synced") return prev; // повтор той же заявки — второй сделки не будет

  const attempts = (prev?.attempts ?? 0) + 1;
  await deps.log.put({ externalId: lead.externalId, status: "pending", attempts, updatedAt: now() });

  try {
    // прошлая попытка могла создать сделку, а ответ — потеряться: сначала ищем её
    const existing = prev ? await deps.crm.findLeadByExternalId(lead.externalId) : null;
    const crmLeadId = existing ?? (await deps.crm.createLead(lead)).crmLeadId;
    const entry: ExchangeEntry = {
      externalId: lead.externalId,
      status: "synced",
      attempts,
      crmLeadId,
      updatedAt: now(),
    };
    await deps.log.put(entry);
    return entry;
  } catch (err) {
    const entry: ExchangeEntry = {
      externalId: lead.externalId,
      status: "failed",
      attempts,
      lastError: err instanceof Error ? err.message : String(err),
      updatedAt: now(),
    };
    await deps.log.put(entry); // повторит фоновая задача по log.failed()
    return entry;
  }
}

import { timingSafeEqual } from "node:crypto";
import type { IncomingMessage, ServerResponse } from "node:http";

export type LeadStatusEvent = { leadId: number; statusId: number; oldStatusId: number | null };

// amoCRM шлёт x-www-form-urlencoded: leads[status][0][id]=…&leads[status][0][status_id]=…
export function parseLeadStatusWebhook(body: string): { subdomain: string | null; events: LeadStatusEvent[] } {
  const params = new URLSearchParams(body);
  const rows = new Map<string, Record<string, string>>();
  for (const [key, value] of params) {
    const m = /^leads\[status\]\[(\d+)\]\[(\w+)\]$/.exec(key);
    if (!m) continue;
    const row = rows.get(m[1]) ?? {};
    row[m[2]] = value;
    rows.set(m[1], row);
  }
  const events = [...rows.values()]
    .filter((r) => r.id && r.status_id)
    .map((r) => ({
      leadId: Number(r.id),
      statusId: Number(r.status_id),
      oldStatusId: r.old_status_id ? Number(r.old_status_id) : null,
    }));
  return { subdomain: params.get("account[subdomain]"), events };
}

const sameSecret = (a: string, b: string) =>
  a.length === b.length && timingSafeEqual(Buffer.from(a), Buffer.from(b));

// Подписи у вебхука нет: секрет в пути URL + сверка субдомена, а данные потом перечитываем через API
export function createWebhookHandler(opts: {
  secret: string;
  subdomain: string;
  onEvents: (events: LeadStatusEvent[]) => void; // только положить в очередь: на ответ есть 2 секунды
}) {
  return (req: IncomingMessage, res: ServerResponse) => {
    const secret = req.url?.match(/^\/hooks\/amocrm\/([\w-]+)$/)?.[1] ?? "";
    if (req.method !== "POST" || !sameSecret(secret, opts.secret)) {
      res.writeHead(404).end();
      return;
    }
    let body = "";
    req.setEncoding("utf8");
    req.on("data", (chunk: string) => {
      body += chunk;
      if (body.length > 1_000_000) req.destroy();
    });
    req.on("end", () => {
      const { subdomain, events } = parseLeadStatusWebhook(body);
      if (subdomain !== opts.subdomain) {
        res.writeHead(403).end();
        return;
      }
      res.writeHead(200).end(); // сначала ответ, обработка — асинхронно
      if (events.length) opts.onEvents(events);
    });
  };
}

import type { LeadStatus } from "../core/lead.ts";
import type { AmoClient } from "./client.ts";
import { toLeadStatus } from "./mapping.ts";
import type { AmoLead, AmoMapping, AmoPage } from "./types.ts";

// Сделки, изменённые после since (Unix, секунды): сортировка по updated_at и постраничное чтение.
// filter[updated_at] не используем: API-фильтрация в amoCRM — отдельный пакет (Alpha)
export async function* leadsUpdatedSince(client: AmoClient, since: number, pageSize = 250) {
  for (let page = 1; ; page++) {
    const res = await client.request<AmoPage<"leads", AmoLead>>(
      "GET",
      `/api/v4/leads?order[updated_at]=desc&limit=${pageSize}&page=${page}`,
    );
    for (const lead of res?._embedded.leads ?? []) {
      if (lead.updated_at < since) return;
      yield lead;
    }
    if (!res?._links.next) return;
  }
}

// Страховка к вебхукам: раз в N минут догоняем пропущенные смены статуса.
// apply обязан быть идемпотентным — одна сделка может прийти и вебхуком, и сверкой
export async function reconcile(
  client: AmoClient,
  map: AmoMapping,
  since: number,
  apply: (crmLeadId: number, status: LeadStatus) => Promise<void>,
  pageSize?: number,
): Promise<{ checked: number; nextSince: number }> {
  let checked = 0;
  let newest = since;
  for await (const lead of leadsUpdatedSince(client, since, pageSize)) {
    if (lead.pipeline_id !== map.pipelineId) continue;
    await apply(lead.id, toLeadStatus(lead.status_id, map));
    checked++;
    newest = Math.max(newest, lead.updated_at);
  }
  return { checked, nextSince: newest - 300 }; // перекрытие 5 минут: страницы сдвигаются во время чтения
}

npm init -y && npm pkg set type=module
npm install -D typescript@7.0.2 @types/node@24 vitest@5.0.3
npx tsc -p tsconfig.json   # types: strict, no emit
npx vitest run             # 23 tests against the local mock
node mock/amo-mock.ts      # mock on its own, for manual curl requests

{
  "compilerOptions": {
    "target": "es2023",
    "module": "nodenext",
    "moduleResolution": "nodenext",
    "strict": true,
    "noEmit": true,
    "allowImportingTsExtensions": true,
    "erasableSyntaxOnly": true,
    "verbatimModuleSyntax": true,
    "types": ["node"],
    "skipLibCheck": true
  },
  "include": ["src", "mock", "test"]
}

// Локальный мок amoCRM API v4: пути, коды и формы ответов — по примерам из документации разработчика.
// Не эмулирует права, контроль дублей, фильтры и прочее, что не нужно тестам адаптера.
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
import type { AddressInfo } from "node:net";

type Cf = { field_id?: number; field_code?: string; values: { value: string; enum_code?: string }[] };
type Contact = { id: number; name: string; custom_fields_values: Cf[] | null };
type Lead = {
  id: number;
  name: string;
  status_id: number;
  pipeline_id: number;
  updated_at: number;
  custom_fields_values: Cf[] | null;
  contactId: number;
};

export type MockState = {
  access: string;
  refresh: string;
  contacts: Contact[];
  leads: Lead[];
  knownLeadFields: number[];
  pipelineId: number;
  firstStatusId: number;
  fail429: number; // сколько следующих запросов к /api ответить 429
  loseNextComplexResponse: boolean; // создать сделку, но ответить 502 — «потерянный ответ»
  requests: string[];
};

const problem = (res: ServerResponse, status: number, extra: object = {}) => {
  res.writeHead(status, { "Content-Type": "application/problem+json" });
  res.end(JSON.stringify({ title: status === 401 ? "Unauthorized" : "Error", type: `https://httpstatus.es/${status}`, status, ...extra }));
};
const hal = (res: ServerResponse, body: unknown) => {
  res.writeHead(200, { "Content-Type": "application/hal+json" });
  res.end(JSON.stringify(body));
};
const readJson = async (req: IncomingMessage) => {
  let s = "";
  for await (const chunk of req) s += chunk;
  return JSON.parse(s || "null");
};
const digits = (s: string) => s.replace(/\D/g, "");
const cfText = (cfs: Cf[] | null) => (cfs ?? []).flatMap((f) => f.values.map((v) => v.value));
const matches = (q: string, haystack: string[]) =>
  haystack.some((h) =>
    digits(q).length >= 7 ? digits(h).includes(digits(q)) : h.toLowerCase().includes(q.toLowerCase()),
  );

export async function startAmoMock(init: Partial<MockState> = {}) {
  const state: MockState = {
    access: "access-1",
    refresh: "refresh-1",
    contacts: [],
    leads: [],
    knownLeadFields: [1001, 1002, 1003],
    pipelineId: 3383152,
    firstStatusId: 33929752,
    fail429: 0,
    loseNextComplexResponse: false,
    requests: [],
    ...init,
  };
  let seq = 19663100;
  const now = () => Math.floor(Date.now() / 1000);

  const server = createServer(async (req, res) => {
    const url = new URL(req.url ?? "/", "http://mock");
    state.requests.push(`${req.method} ${url.pathname}${url.search}`);

    if (req.method === "POST" && url.pathname === "/oauth2/access_token") {
      const body = await readJson(req);
      if (body?.grant_type !== "refresh_token" || body.refresh_token !== state.refresh) return problem(res, 400);
      const n = Number(state.refresh.split("-")[1]) + 1;
      state.access = `access-${n}`;
      state.refresh = `refresh-${n}`; // старый refresh больше не действует
      res.writeHead(200, { "Content-Type": "application/json" });
      return res.end(JSON.stringify({ token_type: "Bearer", expires_in: 86400, access_token: state.access, refresh_token: state.refresh }));
    }

    if (!url.pathname.startsWith("/api/v4/")) return problem(res, 404);
    if (req.headers.authorization !== `Bearer ${state.access}`) return problem(res, 401, { detail: "Unauthorized" });
    if (state.fail429 > 0) {
      state.fail429--;
      return problem(res, 429);
    }

    if (req.method === "GET" && url.pathname === "/api/v4/contacts") {
      const q = url.searchParams.get("query") ?? "";
      const found = state.contacts.filter((c) => matches(q, [c.name, ...cfText(c.custom_fields_values)]));
      if (!found.length) {
        res.writeHead(204);
        return res.end();
      }
      return hal(res, { _page: 1, _links: {}, _embedded: { contacts: found } });
    }

    if (req.method === "GET" && url.pathname === "/api/v4/leads") {
      const q = url.searchParams.get("query");
      const limit = Math.min(Number(url.searchParams.get("limit") ?? 50), 250);
      const page = Number(url.searchParams.get("page") ?? 1);
      let list = q ? state.leads.filter((l) => matches(q, [l.name, ...cfText(l.custom_fields_values)])) : [...state.leads];
      if (url.searchParams.get("order[updated_at]") === "desc") list.sort((a, b) => b.updated_at - a.updated_at || b.id - a.id);
      const slice = list.slice((page - 1) * limit, page * limit);
      if (!slice.length) {
        res.writeHead(204);
        return res.end();
      }
      const next = page * limit < list.length ? { next: { href: `/api/v4/leads?page=${page + 1}` } } : {};
      return hal(res, {
        _page: page,
        _links: { self: { href: url.pathname + url.search }, ...next },
        _embedded: {
          leads: slice.map(({ contactId, ...l }) => ({ ...l, _embedded: { contacts: [{ id: contactId, is_main: true }] } })),
        },
      });
    }

    if (req.method === "POST" && url.pathname === "/api/v4/leads/complex") {
      const items = await readJson(req);
      if (!Array.isArray(items) || items.length > 50) return problem(res, 400, { detail: "Request validation failed" });
      const errors = items.flatMap((item, i) =>
        (item.custom_fields_values ?? [])
          .map((f: Cf, j: number) => ({ f, j }))
          .filter(({ f }: { f: Cf }) => f.field_id !== undefined && !state.knownLeadFields.includes(f.field_id))
          .map(({ j }: { j: number }) => ({
            request_id: item.request_id ?? String(i),
            errors: [{ code: "NotSupportedChoice", path: `custom_fields_values.${j}.field_id`, detail: "The value you selected is not a valid choice." }],
          })),
      );
      if (errors.length) return problem(res, 400, { "validation-errors": errors, title: "Bad Request", detail: "Request validation failed" });

      const result = items.map((item, i) => {
        const embedded = item._embedded?.contacts?.[0];
        let contactId: number;
        if (embedded?.id) contactId = embedded.id;
        else {
          contactId = ++seq;
          state.contacts.push({ id: contactId, name: embedded?.first_name ?? "", custom_fields_values: embedded?.custom_fields_values ?? null });
        }
        const lead: Lead = {
          id: ++seq,
          name: item.name ?? "",
          status_id: item.status_id ?? state.firstStatusId,
          pipeline_id: item.pipeline_id ?? state.pipelineId,
          updated_at: now(),
          custom_fields_values: item.custom_fields_values ?? null,
          contactId,
        };
        state.leads.push(lead);
        return { id: lead.id, contact_id: contactId, company_id: null, request_id: [item.request_id ?? String(i)], merged: false };
      });
      if (state.loseNextComplexResponse) {
        state.loseNextComplexResponse = false;
        return problem(res, 502);
      }
      res.writeHead(200, { "Content-Type": "application/hal+json" });
      return res.end(JSON.stringify(result));
    }

    return problem(res, 404);
  });

  await new Promise<void>((r) => server.listen(0, "127.0.0.1", r));
  const { port } = server.address() as AddressInfo;
  return {
    url: `http://127.0.0.1:${port}`,
    state,
    setStatus(leadId: number, statusId: number, updatedAt: number) {
      const lead = state.leads.find((l) => l.id === leadId);
      if (lead) Object.assign(lead, { status_id: statusId, updated_at: updatedAt });
    },
    close: () => new Promise<void>((r) => server.close(() => r())),
  };
}

if (import.meta.main) {
  const mock = await startAmoMock();
  console.log(`amoCRM mock: ${mock.url}  (Bearer ${mock.state.access})`);
}

import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { startAmoMock } from "../mock/amo-mock.ts";
import { AmoCrmAdapter } from "../src/amocrm/adapter.ts";
import { AmoClient } from "../src/amocrm/client.ts";
import { AmoTokenProvider, type Tokens, type TokenStore } from "../src/amocrm/tokens.ts";
import type { AmoMapping } from "../src/amocrm/types.ts";
import { MemoryExchangeLog } from "../src/core/exchange-log.ts";
import type { LeadRequest } from "../src/core/lead.ts";
import { submitLead } from "../src/core/submit-lead.ts";

const map: AmoMapping = {
  pipelineId: 3383152,
  firstStatusId: 33929752,
  fields: { externalId: 1001, comment: 1002, source: 1003 },
  tag: "site",
};
const lead = (externalId: string, phone: string): LeadRequest => ({
  externalId,
  name: "Айгерим",
  phone,
  source: "site-form",
});

class MemoryTokenStore implements TokenStore {
  tokens: Tokens;
  constructor(tokens: Tokens) {
    this.tokens = tokens;
  }
  async load() {
    return this.tokens;
  }
  async save(t: Tokens) {
    this.tokens = t;
  }
}

let mock: Awaited<ReturnType<typeof startAmoMock>>;
let store: MemoryTokenStore;
let sleeps: number[];
let crm: AmoCrmAdapter;

beforeEach(async () => {
  mock = await startAmoMock();
  store = new MemoryTokenStore({ accessToken: "access-1", refreshToken: "refresh-1", expiresAt: Date.now() + 3_600_000 });
  const tokens = new AmoTokenProvider(mock.url, { clientId: "CLIENT_ID", clientSecret: "CLIENT_SECRET", redirectUri: "https://example.com/oauth" }, store);
  sleeps = [];
  const client = new AmoClient({
    baseUrl: mock.url,
    tokens,
    minIntervalMs: 0,
    sleep: async (ms) => {
      sleeps.push(ms);
    },
  });
  crm = new AmoCrmAdapter(client, map);
});
afterEach(() => mock.close());

describe("AmoCrmAdapter против мок-сервера", () => {
  it("второй лид с тем же телефоном в другом формате привязывается к существующему контакту", async () => {
    const first = await crm.createLead(lead("web-1", "+7 701 123 45 67"));
    const second = await crm.createLead(lead("web-2", "8 (701) 123-45-67"));
    expect(first.contactReused).toBe(false);
    expect(second).toMatchObject({ crmContactId: first.crmContactId, contactReused: true });
    expect(mock.state.contacts).toHaveLength(1);
    expect(mock.state.leads).toHaveLength(2);
  });

  it("повтор той же заявки не создаёт вторую сделку", async () => {
    const log = new MemoryExchangeLog();
    const a = await submitLead(lead("web-7", "87011234567"), { crm, log });
    const b = await submitLead(lead("web-7", "87011234567"), { crm, log });
    expect(a.status).toBe("synced");
    expect(b.crmLeadId).toBe(a.crmLeadId);
    expect(mock.state.leads).toHaveLength(1);
  });

  it("потерянный ответ: сделка создана, но пришёл 502 — повтор находит её по externalId", async () => {
    const log = new MemoryExchangeLog();
    mock.state.loseNextComplexResponse = true;
    const failed = await submitLead(lead("web-9", "87011234567"), { crm, log });
    expect(failed).toMatchObject({ status: "failed", attempts: 1 });
    expect(failed.lastError).toMatch(/HTTP 502/);
    expect(await log.failed()).toHaveLength(1);

    const retried = await submitLead(lead("web-9", "87011234567"), { crm, log });
    expect(retried).toMatchObject({ status: "synced", attempts: 2, crmLeadId: mock.state.leads[0].id });
    expect(mock.state.leads).toHaveLength(1);
  });

  it("429 — пауза и повтор", async () => {
    mock.state.fail429 = 2;
    await crm.createLead(lead("web-3", "87011234567"));
    expect(sleeps).toEqual([1000, 2000]);
    expect(mock.state.leads).toHaveLength(1);
  });

  it("401 — обмен refresh token, новая пара сохранена, запрос повторён", async () => {
    mock.state.access = "access-revoked"; // сервер больше не принимает выданный access token
    mock.state.refresh = "refresh-1";
    await crm.createLead(lead("web-4", "87011234567"));
    expect(store.tokens).toMatchObject({ accessToken: "access-2", refreshToken: "refresh-2" });
    expect(mock.state.requests.filter((r) => r.startsWith("POST /oauth2"))).toHaveLength(1);
    expect(mock.state.leads).toHaveLength(1);
  });

  it("ошибка валидации 400 доходит до журнала с телом ответа", async () => {
    const log = new MemoryExchangeLog();
    mock.state.knownLeadFields = [1001]; // поля 1003 в аккаунте нет — маппинг устарел
    const entry = await submitLead(lead("web-5", "87011234567"), { crm, log });
    expect(entry.status).toBe("failed");
    expect(entry.lastError).toContain("custom_fields_values.1.field_id");
  });
});