diff --git a/packages/apps/bling-erp/src/integration/after-bling-queue.ts b/packages/apps/bling-erp/src/integration/after-bling-queue.ts index b839b485d..f00463f30 100644 --- a/packages/apps/bling-erp/src/integration/after-bling-queue.ts +++ b/packages/apps/bling-erp/src/integration/after-bling-queue.ts @@ -10,6 +10,18 @@ export default async ( payload: any, ) => { const isError = payload instanceof Error; + if (isError && (payload as any).isConcurrentExport) { + /* Outra execução segura o lock de exportação deste pedido por tempo demais. + Rejeitar faz o evento ser reentregue (`failurePolicy`) também na exportação + automática (fora da fila), e a nova tentativa acha o pedido exportado ou assume + o lock, em vez de terminar como falso sucesso sem nada no Bling. */ + logger.warn(payload.message); + return new Promise((resolve, reject) => { + setTimeout(() => { + reject(payload); + }, 2000); + }); + } const isImportation = !!queueEntry.action?.endsWith('importation'); const isQueued = !queueEntry.isNotQueued; const logs = appData.logs || []; diff --git a/packages/apps/bling-erp/src/integration/export-order-to-bling.ts b/packages/apps/bling-erp/src/integration/export-order-to-bling.ts index 3b76b7488..aeffcfd6c 100644 --- a/packages/apps/bling-erp/src/integration/export-order-to-bling.ts +++ b/packages/apps/bling-erp/src/integration/export-order-to-bling.ts @@ -3,6 +3,10 @@ import { URLSearchParams } from 'url'; import ecomUtils from '@ecomplus/utils'; import api from '@cloudcommerce/api'; import { logger } from '@cloudcommerce/firebase/lib/config'; +import { + type ExportLock, + acquireExportLock, +} from '@cloudcommerce/firebase/lib/helpers/export-lock'; import { createBlingClient } from '../bling-auth/client'; import parseOrder from './parsers/order-to-bling'; import parseStatusToBling from './parsers/status-to-bling'; @@ -11,6 +15,14 @@ import getProductsBling from './helpers/get-products-bling'; import { getPaymentBling } from './helpers/payment-method'; import getStatusBling from './helpers/get-status-bling'; +const EXPORT_LOCKS_COLLECTION = 'blingExportLocks'; +/* Maior que o timeout de 300s do `onStoreEvent`: uma execução lenta, mas viva, não +pode ter o lock tomado no meio da criação. */ +const EXPORT_LOCK_STALE_MS = 1000 * 60 * 6; +/* Depois de pegar o lock o vencedor ainda busca contato, forma de pagamento e +produtos antes do POST, com o client limitado a ~1 req/s. */ +const CONCURRENT_EXPORT_WAIT_MS = 1000 * 60; + /* `canCreateNew` is a tri-state: - `true`: always create the order on Bling when not found; @@ -148,69 +160,108 @@ const exportOrderToBling = async ( : String(order.number); } - const customerIdBling = await getCustomerBling(bling, appData, order); - if (!customerIdBling) { - throw new Error('Bling Customer not found'); + /* Criar não é idempotente: o mesmo pedido recebe um evento no checkout e outro + no pagamento, e as duas execuções podem rodar ao mesmo tempo. As duas buscaram + no Bling antes de qualquer uma gravar, então sem o lock as duas criam. */ + let exportLock: ExportLock | undefined; + let isExportedConcurrently = false; + if (!blingOrderId) { + const lockResult = await acquireExportLock(EXPORT_LOCKS_COLLECTION, orderId, { + staleLockMs: EXPORT_LOCK_STALE_MS, + maxWaitMs: CONCURRENT_EXPORT_WAIT_MS, + }); + if (lockResult.isLocked) { + exportLock = lockResult; + } else { + /* Criado por outra execução enquanto esta esperava. O id vem do lock porque + a busca acima não acha pedido com número aleatório nem com o `apiDoc` + lido antes de o vencedor gravar os metafields. */ + isExportedConcurrently = true; + blingOrderId = lockResult.exported.id; + logger.info(`${orderId} exported by a concurrent execution`, { blingOrderId }); + } } - /* Buscado só quando o pedido vai ser criado/atualizado de fato: mudança de - status de pedido já exportado não usa forma de pagamento nem itens. */ - const paymentTypeId = transaction - ? await getPaymentBling(bling, transaction, appData.parse_payment, order.payment_method_label) - : null; - const itemsBling = await getProductsBling(bling, order); - const blingOrder = parseOrder( - order, - blingOrderNumber, - blingStore, - appData, - customerIdBling, - paymentTypeId, - itemsBling, - blingSavedOrder, - ); - const endpoint = `/pedidos/vendas${blingOrderId ? `/${blingOrderId}` : ''}`; - const method = blingOrderId ? 'put' : 'post'; - logger.info(`[${method}]: ${endpoint} for ${order._id}`, { blingOrder, blingStatuses }); - try { - const { data: { data: savedData } } = await bling[method](endpoint, blingOrder); - logger.info(`Bling order ${method === 'put' ? 'upd' : 'cre'}ated successfully`); - if (savedData?.id) { - blingSavedOrder = blingOrder; - blingOrderId = savedData.id; - if (metafieldId) { - metafieldId.value = String(blingOrderId); - } else { - metafields.push({ - _id: ecomUtils.randomObjectId(), - namespace: 'bling', - field: 'bling:id', - value: String(blingOrderId), - }); + + if (!isExportedConcurrently) { + let blingOrder: Record | undefined; + try { + const customerIdBling = await getCustomerBling(bling, appData, order); + if (!customerIdBling) { + throw new Error('Bling Customer not found'); } - /* Guardado para a busca por número funcionar quando o número do Bling - não é o da loja (`random_order_number` ou colisão de numeração). */ - if ( - blingOrderNumber - && !appData.disable_order_number - && !metafields.some(({ field }) => field === 'bling:numero') - ) { - metafields.push({ - _id: ecomUtils.randomObjectId(), - namespace: 'bling', - field: 'bling:numero', - value: String(blingOrderNumber), + /* Buscado só quando o pedido vai ser criado/atualizado de fato: mudança de + status de pedido já exportado não usa forma de pagamento nem itens. */ + const paymentTypeId = transaction + ? await getPaymentBling( + bling, + transaction, + appData.parse_payment, + order.payment_method_label, + ) + : null; + const itemsBling = await getProductsBling(bling, order); + blingOrder = parseOrder( + order, + blingOrderNumber, + blingStore, + appData, + customerIdBling, + paymentTypeId, + itemsBling, + blingSavedOrder, + ); + const endpoint = `/pedidos/vendas${blingOrderId ? `/${blingOrderId}` : ''}`; + const method = blingOrderId ? 'put' : 'post'; + logger.info(`[${method}]: ${endpoint} for ${order._id}`, { blingOrder, blingStatuses }); + exportLock?.setSending(); + const { data: { data: savedData } } = await bling[method](endpoint, blingOrder); + logger.info(`Bling order ${method === 'put' ? 'upd' : 'cre'}ated successfully`); + if (savedData?.id) { + blingSavedOrder = blingOrder; + blingOrderId = savedData.id; + if (metafieldId) { + metafieldId.value = String(blingOrderId); + } else { + metafields.push({ + _id: ecomUtils.randomObjectId(), + namespace: 'bling', + field: 'bling:id', + value: String(blingOrderId), + }); + } + /* Guardado para a busca por número funcionar quando o número do Bling + não é o da loja (`random_order_number` ou colisão de numeração). */ + if ( + blingOrderNumber + && !appData.disable_order_number + && !metafields.some(({ field }) => field === 'bling:numero') + ) { + metafields.push({ + _id: ecomUtils.randomObjectId(), + namespace: 'bling', + field: 'bling:numero', + value: String(blingOrderNumber), + }); + } + await api.patch(`orders/${order._id}`, { metafields } as any).catch(logger.error); + } + } catch (err: any) { + /* O lock é liberado para a próxima tentativa quando o pedido certamente não + foi criado: falha antes do POST (contato, forma de pagamento, produtos) ou + erro respondido pelo Bling. Sem resposta (timeout) o pedido pode ter sido + criado, e o lock fica até expirar. */ + await exportLock?.fail(err); + if (err.response) { + logger.warn(`Failed exporting order ${order._id}`, { + blingOrder, + response: err.response.data, }); } - await api.patch(`orders/${order._id}`, { metafields } as any).catch(logger.error); + throw err; } - } catch (err: any) { - if (err.response) { - logger.warn(`Failed exporting order ${order._id}`, { - blingOrder, - response: err.response.data, - }); + if (exportLock && blingOrderId) { + await exportLock.complete({ id: blingOrderId, numero: blingOrderNumber }); } - throw err; } } diff --git a/packages/apps/tiny-erp/src/integration/after-tiny-queue.ts b/packages/apps/tiny-erp/src/integration/after-tiny-queue.ts index 4f566f4d2..8bc40024f 100644 --- a/packages/apps/tiny-erp/src/integration/after-tiny-queue.ts +++ b/packages/apps/tiny-erp/src/integration/after-tiny-queue.ts @@ -9,6 +9,18 @@ export default async ( payload: any, ) => { const isError = payload instanceof Error; + if (isError && (payload as any).isConcurrentExport) { + /* Another execution holds the export lock of this order for too long. Rejecting + gets the event redelivered (`failurePolicy`) even for automatic exports (not + queued), and the retry finds the order exported or takes over the lock, instead + of ending as a false success with nothing sent to Tiny. */ + logger.warn(payload.message); + return new Promise((resolve, reject) => { + setTimeout(() => { + reject(payload); + }, 2000); + }); + } const isImportation = !!queueEntry.action?.endsWith('importation'); const isQueued = !queueEntry.isNotQueued; const logs = appData.logs || []; diff --git a/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts b/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts index 8662df869..d2007d21e 100644 --- a/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts +++ b/packages/apps/tiny-erp/src/integration/export-order-to-tiny.ts @@ -1,11 +1,25 @@ import type { Orders } from '@cloudcommerce/types'; import { logger } from '@cloudcommerce/firebase/lib/config'; +import { acquireExportLock } from '@cloudcommerce/firebase/lib/helpers/export-lock'; import api from '@cloudcommerce/api'; import postTiny from './post-tiny-erp'; import parseStatusToTiny from './parsers/status-to-tiny'; import parseTinyStatus from './parsers/status-from-tiny'; import parseOrder from './parsers/order-to-tiny'; +const EXPORT_LOCKS_COLLECTION = 'tinyErpExportLocks'; +/* Longer than anything holding the lock can live: the events function times out +in 60s, and the cron holds it for a single order (the POST times out in 30s). */ +const EXPORT_LOCK_STALE_MS = 1000 * 60 * 2; +const CONCURRENT_EXPORT_WAIT_MS = 20000; + +/* `pedido.incluir.php` returns the created order in `registros.registro` */ +const getCreatedTinyId = (response: Record) => { + const { registros } = response; + const registro = (Array.isArray(registros) ? registros[0] : registros)?.registro; + return registro?.id as string | number | undefined; +}; + export default async (apiDoc, queueEntry, appData, canCreateNew) => { const orderId = queueEntry.nextId; let order: Orders; @@ -52,39 +66,38 @@ export default async (apiDoc, queueEntry, appData, canCreateNew) => { logger.info(`${orderId} skipped with last fulfillment status from Tiny`); return null; } - logger.info(`${orderId} searching order ${order.number}`); - - let tinyData: { pedidos?: any }; - try { - tinyData = await postTiny('/pedidos.pesquisa.php', { - numeroEcommerce: String(order.number), - }); - } catch (err: any) { - const status = err.response && err.response.status; - if (status === 404) { - tinyData = {}; - } else { - logger.info(`${orderId} search on tiny ends with status ${status}`); - if (!status) { - logger.error(err); + const searchTinyOrder = async () => { + logger.info(`${orderId} searching order ${order.number}`); + let tinyData: { pedidos?: any }; + try { + tinyData = await postTiny('/pedidos.pesquisa.php', { + numeroEcommerce: String(order.number), + }); + } catch (err: any) { + const status = err.response && err.response.status; + if (status === 404) { + tinyData = {}; + } else { + logger.info(`${orderId} search on tiny ends with status ${status}`); + if (!status) { + logger.error(err); + } + throw err; } - throw err; } - } - const { pedidos } = tinyData; - let originalTinyOrder: undefined | Record; - if (Array.isArray(pedidos)) { - originalTinyOrder = pedidos + const { pedidos } = tinyData; + if (!Array.isArray(pedidos)) return undefined; + const foundEntry = pedidos .sort((a: { pedido: Record }, b: { pedido: Record }) => { return a.pedido.id < b.pedido.id ? 1 : -1; }) .find(({ pedido }) => { return order.number === Number(pedido.numero_ecommerce); }); - if (originalTinyOrder) { - originalTinyOrder = originalTinyOrder.pedido; - } - } + return foundEntry?.pedido as undefined | Record; + }; + + let originalTinyOrder = await searchTinyOrder(); if (!originalTinyOrder) { if (!canCreateNew) { @@ -113,24 +126,52 @@ export default async (apiDoc, queueEntry, appData, canCreateNew) => { break; } } - const tinyOrder = await parseOrder(order, appData); - try { - logger.info(`Posting ${orderId}`, { tinyOrder }); - } catch { - logger.info(`Posting ${orderId} (stringify failed)`); - } - return postTiny('/pedido.incluir.php', { - pedido: { - pedido: tinyOrder, - }, - }).then((response) => { + const exportLock = await acquireExportLock(EXPORT_LOCKS_COLLECTION, orderId, { + staleLockMs: EXPORT_LOCK_STALE_MS, + maxWaitMs: queueEntry.concurrentExportWaitMs ?? CONCURRENT_EXPORT_WAIT_MS, + }); + if (!exportLock.isLocked) { + /* Created by a concurrent execution while this one waited, the search above + could not see it: falls through to the status update on that order. */ + const { exported } = exportLock; + originalTinyOrder = exported.id + ? { id: exported.id, situacao: exported.situacao } + : await searchTinyOrder(); + if (!originalTinyOrder) { + logger.warn(`${orderId} exported by a concurrent execution but not found`); + return null; + } + } else { + let tinyOrder: Record; + let response: Record; + try { + tinyOrder = await parseOrder(order, appData); + try { + logger.info(`Posting ${orderId}`, { tinyOrder }); + } catch { + logger.info(`Posting ${orderId} (stringify failed)`); + } + exportLock.setSending(); + response = await postTiny('/pedido.incluir.php', { + pedido: { + pedido: tinyOrder, + }, + }); + } catch (err) { + await exportLock.fail(err); + throw err; + } logger.info(`Created ${orderId} on Tiny`, { response }); + await exportLock.complete({ + id: getCreatedTinyId(response), + situacao: tinyOrder.situacao, + }); const tinyErpOnNewOrder = global.$tinyErpOnNewOrder; if (tinyErpOnNewOrder && typeof tinyErpOnNewOrder === 'function') { return tinyErpOnNewOrder({ response, order, postTiny }); } return response; - }); + } } if (originalTinyOrder) { diff --git a/packages/apps/tiny-erp/src/tiny-erp-send-orders.ts b/packages/apps/tiny-erp/src/tiny-erp-send-orders.ts index 8156d915b..0559b5946 100644 --- a/packages/apps/tiny-erp/src/tiny-erp-send-orders.ts +++ b/packages/apps/tiny-erp/src/tiny-erp-send-orders.ts @@ -63,7 +63,9 @@ export const sendWaitingOrders = async () => { const order = orders[i] as Orders; if (!order.number) continue; try { - const queueEntry = { nextId: order._id }; + /* An order locked by a concurrent export is left to it (and to the next + run) instead of holding the sequential loop while it finishes. */ + const queueEntry = { nextId: order._id, concurrentExportWaitMs: 0 }; // eslint-disable-next-line no-await-in-loop const result = await exportOrderToTiny(order, queueEntry, appData, true); if (result && typeof result === 'object' && 'registros' in result) { @@ -72,13 +74,18 @@ export const sendWaitingOrders = async () => { skippedCount += 1; } } catch (error: any) { - const err = new Error(`Failed sending order ${order.number} to Tiny`); - logger.error(err, { - orderId: order._id, - orderNumber: order.number, - request: error.config, - response: error.response?.data, - }); + if (error.isConcurrentExport) { + logger.info(`Order ${order.number} skipped while exported concurrently`); + skippedCount += 1; + } else { + const err = new Error(`Failed sending order ${order.number} to Tiny`); + logger.error(err, { + orderId: order._id, + orderNumber: order.number, + request: error.config, + response: error.response?.data, + }); + } } // Rate limit: 150ms between requests // eslint-disable-next-line no-await-in-loop diff --git a/packages/apps/tiny-erp/src/tiny-erp.ts b/packages/apps/tiny-erp/src/tiny-erp.ts index f99c6e0f2..11d725fbe 100644 --- a/packages/apps/tiny-erp/src/tiny-erp.ts +++ b/packages/apps/tiny-erp/src/tiny-erp.ts @@ -15,7 +15,10 @@ export const tinyerp = { onStoreEvent: createAppEventsFunction( 'tinyErp', handleApiEvent, - { memory: '512MB' }, + /* With the default max age (60s), an order event redelivered after waiting on a + concurrent export (`ConcurrentExportError`) would be dropped before an abandoned + export lock gets stale (2 min) and can be taken over. */ + { memory: '512MB', eventMaxAgeMs: 1000 * 60 * 10 }, ), webhook: functions diff --git a/packages/firebase/package.json b/packages/firebase/package.json index 961a1c6cc..12a0340be 100644 --- a/packages/firebase/package.json +++ b/packages/firebase/package.json @@ -13,6 +13,7 @@ "./lib/env": "./lib/env.js", "./lib/helpers/pubsub": "./lib/helpers/pubsub.js", "./lib/helpers/firestore": "./lib/helpers/firestore.js", + "./lib/helpers/export-lock": "./lib/helpers/export-lock.js", "./lib/helpers/update-app-data": "./lib/helpers/update-app-data.js", "./lib/helpers/get-app-data": "./lib/helpers/get-app-data.js" }, diff --git a/packages/firebase/src/helpers/export-lock.ts b/packages/firebase/src/helpers/export-lock.ts new file mode 100644 index 000000000..ea029ba80 --- /dev/null +++ b/packages/firebase/src/helpers/export-lock.ts @@ -0,0 +1,192 @@ +import { randomUUID } from 'node:crypto'; +import { getFirestore, Timestamp } from 'firebase-admin/firestore'; +import { logger } from '../config'; + +/* +Locks the creation of an external record (such as an ERP order) to one execution. + +Resource events are published per modified fields, so the same order usually gets +two events within seconds (checkout, then payment). Each event runs on its own +function execution, and nothing stops two of them from running at the same time: +`maxInstances` is best-effort and is exceeded on concurrent cold starts. Checking +whether the record already exists on the ERP is not enough to dedupe, because +creating it takes seconds and both executions read "not found" before either writes. + +The lock doc is the source of truth for the executions that lose it: they wait on +it, not on the ERP, until the winner either saves what it created (`complete`) or +releases the lock because nothing was created (`fail`), in which case the next one +takes the lock and creates the record itself. + +Docs carry `expiresAt` for a Firestore TTL policy, they are useless after a day. +*/ + +const DEFAULT_POLL_INTERVAL_MS = 2000; +const LOCK_TTL_MS = 1000 * 60 * 60 * 24; + +export type ExportLockOptions = { + /* + A lock without a finished export older than this is taken over, so a record is + not left unexported forever when the winning execution dies between taking the + lock and creating it (timeout, instance eviction). It must be longer than an + execution can hold the lock (the timeout of the function taking it), otherwise + a slow but live execution is taken over and the record is duplicated. + A finished export older than this is taken over as well, it is how a record + deleted on the ERP on purpose can be exported again. + */ + staleLockMs: number, + // How long to wait on a concurrent execution before giving up + maxWaitMs: number, + pollIntervalMs?: number, +}; + +export type ExportLock = { + isLocked: true, + // To call right before sending the request that creates the record + setSending: () => void, + // Saves what concurrent executions need to go on without creating (the ERP id) + complete: (exported: Record) => Promise, + /* + Releases the lock so the next execution can create the record, unless the create + request was sent and failed without a response (timeout, network): the record may + have been created anyway, so the lock is kept until stale to avoid a duplicate. + */ + fail: (err: any) => Promise, +}; + +export type ConcurrentExport = { + isLocked: false, + // Saved by the concurrent execution that created the record + exported: Record, +}; + +/* +Thrown when a concurrent execution holds the lock for longer than `maxWaitMs`. +It is meant to be retried later (event redelivery), never handled as success. +*/ +export class ConcurrentExportError extends Error { + isConcurrentExport = true; + + constructor(lockPath: string) { + super(`Exportação de ${lockPath} em andamento em outra execução`); + this.name = 'ConcurrentExportError'; + } +} + +const sleep = (ms: number) => new Promise((resolve) => { + setTimeout(resolve, ms); +}); + +export const acquireExportLock = async ( + collectionName: string, + resourceId: string, + { + staleLockMs, + maxWaitMs, + pollIntervalMs = DEFAULT_POLL_INTERVAL_MS, + }: ExportLockOptions, +): Promise => { + const firestore = getFirestore(); + const docRef = firestore.collection(collectionName).doc(resourceId); + const lockPath = `${collectionName}/${resourceId}`; + const lockId = randomUUID(); + /* Without Firestore the export goes on unlocked, as before the lock existed: + a possible duplicate is better than an order that never reaches the ERP. */ + let isTracked = true; + let isSending = false; + + const lock: ExportLock = { + isLocked: true, + setSending() { + isSending = true; + }, + async complete(exported) { + if (!isTracked) return; + const exportedData: Record = {}; + Object.keys(exported).forEach((field) => { + if (exported[field] !== undefined) exportedData[field] = exported[field]; + }); + try { + await docRef.set({ + exported: exportedData, + exportedAt: Timestamp.now(), + }, { merge: true }); + } catch (err: any) { + logger.warn(`Failed saving export lock ${lockPath}`, { err }); + } + }, + async fail(err) { + if (!isTracked) return; + if (isSending && !err?.response) { + logger.warn(`Keeping export lock ${lockPath}, record may have been created`, { + err, + }); + return; + } + try { + await firestore.runTransaction(async (transaction) => { + const snapshot = await transaction.get(docRef); + // The lock may have been taken over meanwhile, it is not ours to release + if (snapshot.get('lockId') === lockId) { + transaction.delete(docRef); + } + }); + } catch (_err: any) { + logger.warn(`Failed releasing export lock ${lockPath}`, { err: _err }); + } + }, + }; + + const startedAt = Date.now(); + let isWaitLogged = false; + for (;;) { + let state: 'locked' | 'takenOver' | 'busy' | { exported: Record }; + try { + // eslint-disable-next-line no-await-in-loop + state = await firestore.runTransaction(async (transaction) => { + const snapshot = await transaction.get(docRef); + const now = Date.now(); + const lockData = { + lockId, + lockedAt: Timestamp.fromMillis(now), + expiresAt: Timestamp.fromMillis(now + LOCK_TTL_MS), + }; + if (!snapshot.exists) { + transaction.create(docRef, lockData); + return 'locked'; + } + const exported = snapshot.get('exported') as Record | undefined; + const lockedAt = (snapshot.get('exportedAt') || snapshot.get('lockedAt')) as + Timestamp | undefined; + if (!lockedAt || now - lockedAt.toMillis() >= staleLockMs) { + transaction.set(docRef, lockData); + return 'takenOver'; + } + return exported ? { exported } : 'busy'; + }); + } catch (err: any) { + logger.error(err); + logger.warn(`Exporting ${lockPath} unlocked, export lock is unavailable`); + isTracked = false; + return lock; + } + if (state === 'locked') { + return lock; + } + if (state === 'takenOver') { + logger.warn(`Took over stale export lock ${lockPath}`); + return lock; + } + if (state !== 'busy') { + return { isLocked: false, exported: state.exported }; + } + if (Date.now() - startedAt + pollIntervalMs > maxWaitMs) { + throw new ConcurrentExportError(lockPath); + } + if (!isWaitLogged) { + logger.info(`Waiting concurrent export ${lockPath}`); + isWaitLogged = true; + } + // eslint-disable-next-line no-await-in-loop + await sleep(pollIntervalMs); + } +}; diff --git a/packages/firebase/tests/export-lock.test.ts b/packages/firebase/tests/export-lock.test.ts new file mode 100644 index 000000000..ce5f569d0 --- /dev/null +++ b/packages/firebase/tests/export-lock.test.ts @@ -0,0 +1,212 @@ +import { + describe, + test, + expect, + vi, + beforeEach, +} from 'vitest'; +import { Timestamp } from 'firebase-admin/firestore'; +import { + acquireExportLock, + ConcurrentExportError, + type ExportLock, +} from '../src/helpers/export-lock'; + +/* +In-memory Firestore with the subset used by the lock. Transactions run one at a +time, like the pessimistic locking of the server SDK, and their writes are only +applied when the callback resolves. +*/ +const { db } = vi.hoisted(() => { + const docs = new Map>(); + const state = { isDown: false }; + let queue: Promise = Promise.resolve(); + const checkUp = () => { + if (state.isDown) throw new Error('14 UNAVAILABLE'); + }; + const docRef = (path: string) => ({ + path, + async set(data: Record, opts?: { merge?: boolean }) { + checkUp(); + docs.set(path, opts?.merge ? { ...docs.get(path), ...data } : { ...data }); + }, + }); + const firestore = { + collection: (name: string) => ({ doc: (id: string) => docRef(`${name}/${id}`) }), + runTransaction(fn: (transaction: any) => Promise) { + const run = queue.then(async () => { + checkUp(); + const writes: Array<() => void> = []; + const result = await fn({ + async get({ path }: { path: string }) { + const data = docs.get(path); + return { exists: Boolean(data), get: (field: string) => data?.[field] }; + }, + create({ path }: { path: string }, data: Record) { + writes.push(() => { + if (docs.has(path)) throw new Error('6 ALREADY_EXISTS'); + docs.set(path, { ...data }); + }); + }, + set({ path }: { path: string }, data: Record) { + writes.push(() => docs.set(path, { ...data })); + }, + delete({ path }: { path: string }) { + writes.push(() => docs.delete(path)); + }, + }); + writes.forEach((write) => write()); + return result; + }); + queue = run.catch(() => null); + return run; + }, + }; + return { db: { docs, state, firestore } }; +}); + +vi.mock('firebase-admin/firestore', async (importOriginal) => ({ + ...(await importOriginal()), + getFirestore: () => db.firestore, +})); +vi.mock('../src/config', () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +const coll = 'testExportLocks'; +const docPath = `${coll}/order1`; +const opts = { staleLockMs: 60000, maxWaitMs: 400, pollIntervalMs: 10 }; +const acquire = (options: Partial = {}) => { + return acquireExportLock(coll, 'order1', { ...opts, ...options }); +}; +const minutesAgo = (minutes: number) => Timestamp.fromMillis(Date.now() - minutes * 60000); +const sleep = (ms: number) => new Promise((resolve) => { + setTimeout(resolve, ms); +}); + +const acquireLocked = async (options?: Partial) => { + const lock = await acquire(options); + if (!lock.isLocked) throw new Error('Expected to take the lock'); + return lock as ExportLock; +}; + +describe('Export lock', () => { + beforeEach(() => { + db.docs.clear(); + db.state.isDown = false; + }); + + test('takes a free lock, with expiresAt for the TTL policy', async () => { + const lock = await acquire(); + expect(lock.isLocked).toBe(true); + const doc = db.docs.get(docPath); + expect(doc?.lockedAt).toBeInstanceOf(Timestamp); + expect(doc?.expiresAt.toMillis()).toBeGreaterThan(Date.now() + 60 * 60000); + }); + + test('locks per resource', async () => { + await acquireLocked(); + const other = await acquireExportLock(coll, 'order2', opts); + expect(other.isLocked).toBe(true); + }); + + test('concurrent execution waits and gets what the winner exported', async () => { + const first = acquire(); + const second = acquire(); + const winner = await first; + if (!winner.isLocked) throw new Error('First execution should win'); + await sleep(30); + await winner.complete({ id: 123, numero: undefined }); + expect(await second).toEqual({ isLocked: false, exported: { id: 123 } }); + }); + + test('concurrent execution takes the lock when the winner fails before sending', async () => { + const winner = await acquireLocked(); + const second = acquire(); + await sleep(30); + await winner.fail(new Error('Bling Customer not found')); + expect((await second).isLocked).toBe(true); + }); + + test('released when the ERP answered the create request with an error', async () => { + const winner = await acquireLocked(); + winner.setSending(); + await winner.fail(Object.assign(new Error('Tiny error 6'), { response: { status: 503 } })); + expect(db.docs.has(docPath)).toBe(false); + expect((await acquire()).isLocked).toBe(true); + }); + + test('kept when the create request failed with no response', async () => { + const winner = await acquireLocked(); + winner.setSending(); + await winner.fail(Object.assign(new Error('timeout of 10000ms exceeded'), { + code: 'ECONNABORTED', + })); + expect(db.docs.has(docPath)).toBe(true); + await expect(acquire({ maxWaitMs: 50 })).rejects.toBeInstanceOf(ConcurrentExportError); + }); + + test('gives up with a retryable error after maxWaitMs', async () => { + await acquireLocked(); + const startedAt = Date.now(); + const err = await acquire({ maxWaitMs: 100 }).catch((_err) => _err); + expect(err).toBeInstanceOf(ConcurrentExportError); + expect(err.isConcurrentExport).toBe(true); + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(80); + }); + + test('does not wait at all with maxWaitMs 0', async () => { + await acquireLocked(); + const startedAt = Date.now(); + await expect(acquire({ maxWaitMs: 0 })).rejects.toBeInstanceOf(ConcurrentExportError); + expect(Date.now() - startedAt).toBeLessThan(50); + }); + + test('takes over a stale lock, then the renewed lock is busy', async () => { + db.docs.set(docPath, { lockId: 'dead', lockedAt: minutesAgo(2) }); + expect((await acquire()).isLocked).toBe(true); + expect(db.docs.get(docPath)?.lockId).not.toBe('dead'); + await expect(acquire({ maxWaitMs: 0 })).rejects.toBeInstanceOf(ConcurrentExportError); + }); + + test('takes over a doc without lockedAt instead of locking forever', async () => { + db.docs.set(docPath, { foo: 'bar' }); + expect((await acquire()).isLocked).toBe(true); + }); + + test('recent export is returned, old export is taken over to export again', async () => { + db.docs.set(docPath, { + lockId: 'done', + lockedAt: minutesAgo(0.5), + exported: { id: 1 }, + exportedAt: Timestamp.now(), + }); + expect(await acquire()).toEqual({ isLocked: false, exported: { id: 1 } }); + db.docs.set(docPath, { + lockId: 'done', + lockedAt: minutesAgo(5), + exported: { id: 1 }, + exportedAt: minutesAgo(2), + }); + expect((await acquire()).isLocked).toBe(true); + }); + + test('former holder of a taken over lock does not release the new one', async () => { + const formerHolder = await acquireLocked(); + db.docs.get(docPath)!.lockedAt = minutesAgo(2); + await acquireLocked(); + await formerHolder.fail(new Error('Bling Customer not found')); + expect(db.docs.has(docPath)).toBe(true); + await expect(acquire({ maxWaitMs: 0 })).rejects.toBeInstanceOf(ConcurrentExportError); + }); + + test('exports unlocked when Firestore is unavailable', async () => { + db.state.isDown = true; + const lock = await acquireLocked(); + lock.setSending(); + await expect(lock.complete({ id: 1 })).resolves.toBeUndefined(); + await expect(lock.fail(new Error('timeout'))).resolves.toBeUndefined(); + db.state.isDown = false; + expect(db.docs.size).toBe(0); + }); +});