Как я подружил бэкенд с API AmoCRM: связь сущностей, отправка данных, защита от дублей / Хабр

Интеграция простой формы с AmoCRM на «бумаге» выглядит просто. Кажется, что можно просто отправить контакт, создать лид, прикрепить товары к сделке — и готово. На практике всё наоборот.

Честно говоря, документация AmoCRM сначала меня запутала. Я полез гуглить по моей ситуации (связка формы с CRM), но не нашел почти ничего. Посмотрел ролик на YouTube про библиотеку. Понял основы, но всё равно оставалось куча вопросов.

Дело в том, что AmoCRM в упор не видит дубликаты контактов и товаров. При очистке дублей из админки ничего не удаляется. Все из-за уникальных ID, которые назначаются при отправке данных.

После множества экспериментов, я все таки смог подружить небольшой бэкенд и API AmoCRM.

В данной статье я разберу два модуля. Для соединения с API использовал библиотеку amocrm-js. Ссылка на библиотеку

amoCRM.ts — интеграционный слой. Связываем сущности, проверяем дубли, отправляем данные.

orderProcessor.ts — воркер на BullMQ/Redis для очереди сообщений и кеша.

Все данные по сделкам подтягиваются из небольшого интернет магазина.

Разберемся для начала как устроена механика.

Каркас интеграции

Создаем внешнюю интеграцию в AmoCRM
Указываем ссылку на сайт (это нужно для аутентификации)
Предоставляем доступ: Все
Пишем название и описание
Получаем секретный ключ и долгосрочный токен

// Логирует сообщения с единым префиксом AmoCRM для удобной фильтрации.
const amoLog = (...args: unknown[]) => console.log("[AmoCRM]", ...args);

// Ваши данные из интеграции
const client = new Client({
  domain: "rolloffstore",
  auth: {
    client_id: process.env.AMO_CLIENT_ID, // ID интеграции
    client_secret: process.env.AMO_CLIENT_SECRET, // Секретный ключ
    redirect_uri: process.env.AMO_REDIRECT_URI, // Ссылка на перенаправление
    bearer: process.env.AMO_TOKEN, // Долгосрочный токен
  },
});

const connectionPromise = client.connection
  .connect()
  .then(() => {
    amoLog("connection:established");
  })
  .catch((error) => {
    console.error("Failed to connect to AmoCRM", error);
    throw error;
  });

// Гарантирует установку соединения с AmoCRM перед выполнением запросов.
async function ensureConnected() {
  await connectionPromise;
}

Для чего это:

Один ленивый промис на всё приложение убирает гонки «кто первый подключится» и дублирование кода.
Все функции upsert* / createLead* будут начинаться с await ensureConnected() для проверки соединения с API.

«Корзины ссылок» (link buckets): копим привязки и контролируем поведение

Проблема

Нам нужно было брать данные заказов и отправлять их в CRM. Звучит просто, но на деле всплывает множество других проблем.

При отправке данных контакта, AmoCRM создает новый контакт каждый раз, назначая уникальный ID. В итоге мы получаем дубли. Amo в упор не видит дубли и не удаляет.

Такая же история с товарами. Если связывать эти уникальные контакты и товары со сделкой, то мы не увидим историю заказов клиента.

Можно конечно использовать интеграцию по контролю дублей из маркетплейса, но я не горел желанием покупать 3-4 интеграции по 10 000 рублей.

Задача

Чтобы каждый раз не обращаться к API при связке контактов и товаров к сделке, создать временное хранилище для этих ссылок.

Пока мы не создали сделку, накапливаем привязки (контакты/товары) в этом временном хранилище и группируем по ключу заказа.

Потом одним запросом /leads/{id}/link связываем их со сделкой.

// amoCrm.ts

import { Client } from "amocrm-js";
import { Order } from "../types/order"; // Импорт типа (проверка получаемых данных).
import { toString } from "lodash";

type LinkPayload = {
  to_entity_id: number;
  to_entity_type: "contacts" | "catalog_elements";
  metadata?: {
    quantity: number;
    catalog_id: number;
  };
};

/**
 * Временное хранилище связей (контакты/товары), сгруппированных по ключу заказа.
 Ключ заказа — ID задачи в BullMQ. Если его нет, то назначается randomUUID.
 * Пока сделка не создана, ссылки собираются здесь, после чего одной пачкой
 * отправляются в AmoCRM через "createLeadInAmo".
 */
 
const linkBuckets = new Map<string, LinkPayload[]>();

/**
 * Возвращает корзину ссылок для заданного ключа, создавая её при необходимости.
 */
const getBucket = (linkKey: string) => {
  if (!linkBuckets.has(linkKey)) {
    linkBuckets.set(linkKey, []);
  }

  return linkBuckets.get(linkKey);
};


/**
 * Откладываем привязку элемента каталога к сделке, сохраняя количество товаров и ID каталога.
 */
export const pushProduct = (
  linkKey: string,
  prodId: number,
  quantity: number,
  catalogId: number,
	) => {
	  if (!prodId) {
	    amoLog("pushProduct:skip", { linkKey, prodId, reason: "missing id" });
	    return;
	  }
	
  // Получаем корзину ссылок по ключу заказа
  const bucket = getBucket(linkKey);
  amoLog("pushProduct", { linkKey, prodId, quantity, catalogId });

  // Пушим товары в корзину
  bucket.push({
    to_entity_id: prodId,
    to_entity_type: "catalog_elements",
    metadata: {
      quantity,
      catalog_id: catalogId,
    },
  });
};

/**
 * Откладывает привязку контакта к сделке, избегая повторных связей.
 * Сделка создается после того, как контакт будет добавлен/обновлен в CRM.
 */
export const pushContact = (linkKey: string, contactId: number) => {
  if (!contactId) {
    amoLog("pushContact:skip", { linkKey, contactId, reason: "missing id" });
    return;
  }
	
  // Получаем корзину ссылок по ключу заказа
  const bucket = getBucket(linkKey);
  amoLog("pushContact", { linkKey, contactId });

  const alreadyLinked = bucket.some(
    (link) => link.to_entity_type === "contacts" && link.to_entity_id === contactId,
  );

  if (!alreadyLinked) {
    bucket.push({
      to_entity_id: contactId,
      to_entity_type: "contacts",
    });
  } else {
    amoLog("pushContact:skipped", { linkKey, contactId, reason: "duplicate" });
  }
};

pushProduct и pushContact складывают будущие связи (товары и контакт) в «корзину» по ключу заказа (linkKey). Позже, когда лид создан, весь накопленный список уходит одним запросом на линковку сущностей. Это ускоряет поток, упрощает контроль дублей и уменьшает количество API-вызовов.

// amoCrm.ts

/**
 * Извлекает и очищает накопленные связи для передачи в AmoCRM одной пачкой.
 */
export const consumeLinks = (linkKey: string) => {
  const bucket = linkBuckets.get(linkKey) ?? [];
  amoLog("consumeLinks", { linkKey, count: bucket.length });
  
  // Удаление ключа из Map не очищает уже возвращённый массив — у нас остаётся валидная ссылка (linkPayload ниже), которую и возвращаем.
  linkBuckets.delete(linkKey);
  return bucket;
};

/**
 * Сбрасывает корзину связей по ключу после завершения операции или ошибки. Принудительная очистка корзины
 */
export const resetLinks = (linkKey: string) => {
  const removed = linkBuckets.delete(linkKey);
  amoLog("resetLinks", { linkKey, removed });
};

consumeLinks(linkKey) — забирает накопленные связи и для подготовки к отправке

Берёт «корзину» связей (контакты и товары), накопленную по linkKey.

Удаляет запись из Map, чтобы корзина больше не висела в памяти.

В случае если HTTP запрос упадет, то данные потеряются, так как корзина уже удалена. Поэтому лучше создать функционал с заменой, а не удалением.

resetLinks(linkKey) — принудительно удаляет корзину.

Ни в коем случае не вызывать его до отправки данных, иначе всё удалите.

Как выглядит процесс с корзиной:

Приходят данные заказа — вы копите связи через pushContact/pushProduct.
Создаёте лид.
Запускаете consumeLinks → получаете массив ссылок → отправляете одним запросом.
В finally — запускаете resetLinks, чтобы гарантированно не осталось мусора.

Если важно не терять ссылки при сбоях запросов:

Вместо удаления корзины по ключу, подмените на пустую и возвращайте старую со ссылками.

Если линковка упала, сложите неотправленные ссылки обратно в пустую корзину.

// Пример

type LinkPayload = {
  to_entity_id: number;
  to_entity_type: "contacts" | "catalog_elements";
  metadata?: { quantity: number; catalog_id: number };
};

export const consumeSwap = (linkKey: string) => {
  const bucket = linkBuckets.get(linkKey) ?? [];
  amoLog("consumeLinks", { linkKey, count: bucket.length });
  
  // На место кладём пустой массив
  linkBuckets.set(linkKey, []);
  return bucket;
};

export const rollBack = (linkKey: string, notSent: LinkPayload[]) => {
	if (notSent.length === 0) return;
  // Берём то, что уже накопилось в корзине. В случае undefined возвращаем пустой массив
  const pending = linkBuckets.get(linkKey) ?? [];
  
  // Склеиваем массивы и кладём обратно в Map. Сначала помещаем те, которые не отправились, а потом то, что накопилось позже
  linkBuckets.set(linkKey, notSent.concat(pending));

}

rollBack нужно будет поместить в блок try catch. При ошибке, вызываем rollBack, передавая payload.

Поиск, создание, обновление контактов

При поиске контактов номер телефона является ключевой переменной, а имя — вторичная проверка.

// amoCrm.ts

export async function upsertContactInAmo(orders: Order, linkKey: string): Promise<number> {
  // Расставляем вывод логов, чтобы видеть где крэшится
  amoLog("upsertContact:ensuringConnection");
  await ensureConnected();
  amoLog("upsertContact:start", { linkKey, phone: orders.phone, name: orders.name });
	
  // Пробуем найти контакт
  try {
    amoLog("upsertContact:searching", { linkKey, phone: orders.phone });
    const searchRes = await client.request.get(
      `/api/v4/contacts?query=${encodeURIComponent(orders.phone)}`
    );

    const searchData = searchRes.data as { _embedded?: { contacts?: any[] } };
    const rawHits = searchData._embedded?.contacts ?? [];
    amoLog("upsertContact:searchResults", { linkKey, count: rawHits.length });
		
		
    let contact: InstanceType<typeof client.Contact> | null = null;
    
  	// Ищем контакт по номеру телефона
    for (const raw of rawHits) {
      const c = raw;
      const cf: Array<{ field_code?: string; values?: any[] }> = c.custom_fields_values;

      const phoneField = cf.find(
        (f: { field_code?: string; values?: any[] }) => f.field_code === "PHONE",
      );
      const values: string[] = phoneField && Array.isArray(phoneField.values)
        ? phoneField.values.map((v: any) => v.value)
        : [];
        
      // Если есть мэтч, то присваиваем к переменной contact
      if (c.name === orders.name && values.includes(orders.phone)) {
        contact = c;
        amoLog("upsertContact:existingMatch", { linkKey, contactId: contact.id });
        break;
      }
    }
		
	// Если контакта не существует, то создаем новый и связываем с номером телефона из заказа
    if (!contact) {
      const nc = new client.Contact();
      nc.name = orders.name;
      nc.custom_fields_values = [
        {
          field_id: amoContactIds.phoneId, // Phone field ID
          field_code: "PHONE",
          values: [{ value: orders.phone }],
        },
        {
          field_id: amoContactIds.addressId, // Address field ID
          values: [{ value: orders.address ?? "" }],
        },
      ];
      
      // Сохраняем контакт
      await nc.save();
      contact = nc;
      amoLog("upsertContact:created", { linkKey, contactId: contact.id });
    } else {
      amoLog("upsertContact:updateExisting", { linkKey, contactId: contact.id });
      
      // Обновляем данные существующего контакта
      contact.custom_fields_values = [
        {
          field_id: amoContactIds.phoneId,
          field_code: "PHONE",
          values: [{ value: orders.phone }],
        },
        {
          field_id: amoContactIds.addressId,
          values: [{ value: orders.address ?? "" }],
        },
      ];
      
      // Отправляем данные в AmoCRM
      await client.contacts.update([contact]);
      amoLog("upsertContact:updated", { linkKey, contactId: contact.id });
    }
    
    // Добавляем контакт в корзину ссылок
    pushContact(linkKey, contact.id);
    amoLog("upsertContact:success", { linkKey, contactId: contact.id });
    
    // Возвращаем ID сохраненного контакта. Это нужно для кеширования и создания сделки (разберем ниже).
    return contact.id;
  } catch (error) {
    console.error("Failed to upsert contact in AmoCRM", error);
    amoLog("upsertContact:error", { linkKey, message: (error as Error)?.message });
    throw error;
  }
}

Поиск, создание, обновление товаров

Логика такая же, но уже взаимодействуем с каталогом.

Для поиска товаров нам нужно найти ID каталога.

// amoCrm.ts

export async function upsertProductInAmo(orders: Order, linkKey: string): Promise<number> {
  await ensureConnected();
  amoLog("upsertProduct:start", { linkKey, itemCount: orders.items.length });

	
  try {
	  // Смотрим какие каталоги у нас есть
    const catalogsRes = await client.request.get("/api/v4/catalogs");
    const catalogsData = catalogsRes.data as { _embedded?: { catalogs?: any[] } };
    const catalogs = catalogsData._embedded?.catalogs || [];
    
    // Ищем каталог с названием "Товары"
    // У меня был только один каталог с товарами. Если у вас несколько, то нужно проверять и по имени каталога
    const productsCatalog = catalogs.find((c: any) => c.type === "products");

    if (!productsCatalog) {
      throw new Error("Products catalog not found in AmoCRM");
    }
		
	// После поиска присваиваем ID к переменной
    const catalogId = productsCatalog.id;
    
    // amoProductIds.priceId — поле из группы полей в сделке.
    // Можете создать объект и внести туда ID своих полей
    const PRICE_FIELD = amoProductIds.priceId;
    
    // Переменная для последнего созданного/обновленного контакта
    let lastElementId: number;
		
	// Я не стал делать сложную логику по массовой проверке товаров из ответа.
	// Так как в заказе товаров обычно 5-6, то можно проверить каждый товар отдельно.
    for (const p of orders.items) {
      amoLog("upsertProduct:item", { linkKey, name: p.name, quantity: p.quantity });
			
			
      const filterUrl = `/api/v4/catalogs/${catalogId}/elements?filter[name]=${encodeURIComponent(
        p.name,
      )}`;
      amoLog("upsertProduct:searching", { linkKey, name: p.name, url: filterUrl });

      // Нужно найти товар и определить id полей
      const response = await client.request.get(filterUrl);
      const data = response.data as { _embedded?: { elements?: any[] } };
      const hits = data._embedded?.elements || [];

      let elementId: number;
      // Мэтч по имени
      const match = hits.find((el: any) => el.name === p.name);
			
			
      // При мэтче, просто обновляем данные товара
      if (match) {
        elementId = match.id;
        amoLog("upsertProduct:match", { linkKey, elementId, name: p.name });

        const payload = [
          {
            id: match.id,
            name: p.name,
            custom_fields_values: [
              {
                field_id: amoProductIds.priceId,
                values: [{ value: p.price }],
              },
            ],
          },
        ];
				
				
        // Обновляем данные товара
        await client.request.patch(`/api/v4/catalogs/${catalogId}/elements`, payload);
        amoLog("upsertProduct:updated", { linkKey, elementId, name: p.name });
      } else {
        amoLog("upsertProduct:create", { linkKey, name: p.name });

        const payload = [
          {
            name: p.name,
            custom_fields_values: [
              {
                field_id: amoProductIds.priceId,
                values: [{ value: p.price }],
              },
            ],
          },
        ];
				
				
        // Создаем новый товар в каталоге
        const created = await client.request.post(`/api/v4/catalogs/${catalogId}/elements`, payload);
        const createdData = created.data as { _embedded: { elements: Array<{ id: number }> } };
        // Вытаскиваем ID созданного товара
        const createdProductId = createdData._embedded.elements?.[0]?.id;

        if (!createdProductId) {
          throw new Error("Failed to create product in AmoCRM");
        }
				
				
        elementId = createdProductId;
        amoLog("upsertProduct:created", { linkKey, elementId, name: p.name });
      }
			
			
      // Добавляем в корзину ссылок
      pushProduct(linkKey, elementId, p.quantity, catalogId);
      amoLog("upsertProduct:linked", {
        linkKey,
        elementId,
        quantity: p.quantity,
        catalogId,
      });
      lastElementId = elementId;
    }

    amoLog("upsertProduct:complete", { linkKey, lastElementId });
    
    // Возвращаем ID. Нужно для создания сделки (разберем ниже).
    return lastElementId;
  } catch (error) {
    console.error("Failed to upsert product in AmoCRM", error);
    amoLog("upsertProduct:error", { linkKey, message: (error as Error)?.message });
    throw error;
  }
}

Создание сделки, привязка контактов и товаров

// amoCrm.ts

export async function createLeadInAmo(orders: Order, linkKey: string): Promise<number> {
  await ensureConnected();
  amoLog("createLead:start", { linkKey, total: orders.totalPrice, name: orders.name });

  // amoIds — это объект с полями из сделки. Можете создать кастомную группу полей и скопировать ID.
  try {
    const createdLeads = await client.leads.create([
      {
        name: orders.name,
        custom_fields_values: [
          {
            field_id: amoIds.phoneId,
            values: [{ value: toString(orders.phone) }],
          },
          {
            field_id: amoIds.numberOfPeopleId,
            values: [{ value: Number(orders.numberOfPeople) }],
          },
          {
            field_id: amoIds.deliveryMethodId,
            values: [{ value: toString(orders.deliveryMethod) }],
          },
          {
            field_id: amoIds.locationId,
            values: [{ value: toString(orders.location) }],
          },
          {
            field_id: amoIds.rayonId,
            values: [{ value: toString(orders.rayon) }],
          },
          {
            field_id: amoIds.addressId,
            values: [{ value: toString(orders.address) }],
          },
          {
            field_id: amoIds.paymentMethodId,
            values: [{ value: toString(orders.paymentMethod) }],
          },
          {
            field_id: amoIds.changeNeededId,
            values: [{ value: toString(orders.changeNeeded) }],
          },
          {
            field_id: amoIds.sumId,
            values: [{ value: toString(orders.totalPrice) }],
          },
          {
            field_id: amoIds.changeNeededId,
            values: [{ value: toString(orders.changeNeeded) }],
          },
          {
            field_id: amoIds.promocode,
            values: [{ value: toString(orders.promocode) }],
          },
          {
            field_id: amoIds.commentId,
            values: [{ value: toString(orders.comment) }],
          },
        ],
      },
    ]);
		
		
    // Создаем сделку и вытаскиваем ID
    const leadId = createdLeads[0]?.id;

    if (!leadId) {
      throw new Error("Lead was not created in AmoCRM");
    }
		
		
    // Получаем корзину ссылок (контакты и товары)
    const linkPayload = consumeLinks(linkKey);
    amoLog("createLead:links", { linkKey, count: linkPayload.length });
		
    if (linkPayload.length > 0) {
	    
      // Пачкой привязываем контакты и товары к сделке
      await client.request.post(`/api/v4/leads/${leadId}/link`, linkPayload);
      amoLog("createLead:linked", { linkKey, leadId, linkCount: linkPayload.length });
    }

    amoLog("createLead:success", { linkKey, leadId });
    return leadId;
  } catch (error) {
    console.error("Failed to create lead in AmoCRM", error);
    
    // Здесь можно вставить rollBack, но linkPayload нужно объявить снаружи try блока
    // rollbackLinks(linkKey, linkPayload);
    amoLog("createLead:error", { linkKey, message: (error as Error)?.message });
    throw error;
  } finally {
	  
    // Очищаем корзину ссылок
    resetLinks(linkKey);
  }
}

Очередь сообщений, кеширование, лимитирование, ретраи

Кеш контакта по телефону в Redis → меньше запросов к API Rate limiter (limiter.schedule) — значит укладываемся в лимиты API.

Импортируем Bottleneck для ограничения количества параллельных запросов. Подключаем Redis для хранения состояний.

// rateLimiter.ts

import Bottleneck from "bottleneck";

export const limiter = new Bottleneck({
  maxConcurrent: 5,
  minTime: 50,
  datastore: "ioredis",
  clientOptions: {
    host: process.env.REDIS_HOST,
    port: Number(process.env.REDIS_PORT),
    password: process.env.REDIS_PASSWORD,
  },
});

// orderProcessor.ts

import { randomUUID } from "crypto";
import { Worker, Job } from "bullmq";
import pRetry from "p-retry";
import { redis } from "../lib/redis";
import "dotenv/config"
import { limiter } from "../lib/rateLimiter";
import {
  upsertContactInAmo,
  upsertProductInAmo,
  createLeadInAmo,
  pushContact,
  resetLinks,
} from "amoCrm";
import type { Order } from "../types/order";
import { Client } from "amocrm-js";

const workerLog = (...args: unknown[]) => console.log("[OrderProcessor]", ...args);

const client = new Client({
  domain: "rolloffstore",
  auth: {
    client_id: process.env.AMO_CLIENT_ID!,
    client_secret: process.env.AMO_CLIENT_SECRET!,
    redirect_uri: process.env.AMO_REDIRECT_URI!,
    bearer: process.env.AMO_TOKEN!,
  },
});

// Проверяем подключение
async function connectAmo() {
  workerLog("connect:start");

  try {
    const res = await client.connection.connect();
    if (!res) {
      throw new Error("Failed to connect to AmoCRM");
    }

    workerLog("connect:success");
  } catch (error) {
    console.error("Error connecting to AmoCRM");
    workerLog("connect:error", { message: (error as Error)?.message });
    throw error;
  }
}

// orderProcessor.ts

// Ищем контакт в кеше или помещаем новый. Помним, что функции возвращают ID.
const getOrCreateContact = async (order: Order, linkKey: string): Promise<number> => {
  workerLog("contact:lookup", { linkKey, phone: order.phone });

  const key = `contact:${order.phone}`;
  const cached = await redis.get(key);

  if (cached) {
    const contactId = Number(cached);
    workerLog("contact:cacheHit", { linkKey, contactId });
    pushContact(linkKey, contactId);
    return contactId;
  }
    workerLog("contact:notFound, upserting", { linkKey, phone: order.phone });
    
    // Получаем ID контакта из AmoCRM
    const contactId = await limiter.schedule(() => upsertContactInAmo(order, linkKey));
    workerLog("contact:created", { linkKey, contactId });

  await redis.set(key, String(contactId));
  // Возвращаем ID
  return contactId;
};

// orderProcessor.ts

const processOrder = async (order: Order, linkKey: string) => {
  workerLog("job:start", { linkKey, total: order.totalPrice, itemCount: order.items.length });
  resetLinks(linkKey);

  try {
	// Получаем ID контакта
    const contactId = await getOrCreateContact(order, linkKey);
    workerLog("job:contactReady", { linkKey, contactId });
    
	// Получаем ID товара
    const productId = await limiter.schedule(() => upsertProductInAmo(order, linkKey));
    workerLog("job:productsReady", { linkKey, productId });
		
	// Получаем ID созданной сделки
    const leadId = await limiter.schedule(() => createLeadInAmo(order, linkKey));
    workerLog("job:leadReady", { linkKey, leadId });

    const result = {
      contactId,
      productId,
      leadId,
      status: "AmoCRM lead processed successfully",
    };

    workerLog("job:complete", { linkKey, leadId });
    return result;
  } catch (error) {
    console.error("Failed to process AmoCRM order workflow", error);
    workerLog("job:error", { linkKey, message: (error as Error)?.message });
    resetLinks(linkKey);
    throw error;
  }
};

// orderProcessor.ts

(async () => {
  try {
    console.log("Starting AmoCRM order processor worker...");
    await connectAmo();
  } catch (error) {
    console.error("Initial AmoCRM connection failed", error);
    workerLog("startup:error", { message: (error as Error)?.message });
    process.exit(1);
  }
	
  // Создаем новый воркер
  const worker = new Worker<Order>(
    "amoQueue",
    async (job: Job<Order>) => {
      workerLog("worker:jobReceived", { jobId: job.id });
      const linkKey = job.id?.toString() ?? randomUUID();
      workerLog("worker:linkKey", { jobId: job.id, linkKey });
      
      // Запускаем процесс с 2 ретраями
      return await pRetry(() => processOrder(job.data, linkKey), { retries: 2 });
    },
    {
      connection: {
        host: process.env.REDIS_HOST,
        port: parseInt(process.env.REDIS_PORT),
        password: process.env.REDIS_PASSWORD,
      },
      concurrency: 5,
    }
  );

  worker.on("active", (job) => {
    workerLog("worker:active", { jobId: job.id });
  });

  worker.on("completed", (job) => {
    workerLog("worker:completed", { jobId: job.id, result: job.returnvalue });
  });

  worker.on("failed", (job, err) => {
    console.error(`Job ${job?.id} failed:`, err);
    workerLog("worker:failed", { jobId: job?.id, message: err?.message });
  });

  worker.on("error", (err) => {
    console.error("Worker error:", err);
    workerLog("worker:error", { message: err?.message });
  });
})();

Заключение

Конечно, можно было бы улучшить много чего: перейти на массовые проверки наличия товаров вместо серии одиночных запросов к API, активнее кешировать данные и состояния, добавить вспомогательный функционал. Но это уже следующий шаг, если вы хотите больше скорости, меньше нагрузки на amoCRM и стабильную работу при масштабировании.

Что конкретно можно улучшить или добавить:

Массовые операции. Ищем/обновляем контакты, сделки и товары пачками, линковку отправляем батчами.
Кешируем товаров по SKU. Чтобы не отправлять одиночные запросы.
Замена и rollBack вместо удаления корзины.
Пагинацию, если в базе много товаров или контактов.
Разделить Redis: db0 — BullMQ, db1 — кеш.
Токены и секреты: авто-обновление bearer.
Градация приоритетов в очереди (важные заказы вперёд).

Я лишь написал рабочую схему. Все дальнейшие доработки снизят число запросов и улучшат устойчивость под нагрузкой.

Если возникли вопросы, напишите в комментариях.