admin/bodyshop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

6feature/IO-3556-Chattr-Integration - Move to BULLMQ stack

Browse Source
This commit is contained in:
Dave
2026-02-12 12:56:17 -05:00
parent 188a7b47b1
commit 34fe0cc3bf
3 changed files with 267 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
server/data/chatter-api.js
View File

@@ -40,7 +40,6 @@ const logger = require("../utils/logger");
const { ChatterApiClient, getChatterApiToken, CHATTER_BASE_URL } = require("../chatter/chatter-client");
const client = require("../graphql-client/graphql-client").client;
const { sendServerEmail } = require("../email/sendemail");
const CHATTER_EVENT = process.env.NODE_ENV === "production" ? "delivery" : "TEST_INTEGRATION";
const MAX_CONCURRENCY = Number(process.env.CHATTER_API_CONCURRENCY || 5);
@@ -53,74 +52,98 @@ const clientCache = new Map(); // companyId -> ChatterApiClient
const tokenInFlight = new Map(); // companyId -> Promise<string> (for in-flight deduplication)
const companyRateLimiters = new Map(); // companyId -> rate limiter
/**
* Core processing function for Chatter API jobs.
* This can be called by the HTTP handler or the BullMQ worker.
*
* @param {Object} options - Processing options
* @param {string} options.start - Start date for the delivery window
* @param {string} options.end - End date for the delivery window
* @param {Array<string>} options.bodyshopIds - Optional specific shops to process
* @param {boolean} options.skipUpload - Dry-run flag
* @param {Object} options.sessionUtils - Optional session utils for token caching
* @returns {Promise<Object>} Result with totals, allShopSummaries, and allErrors
*/
async function processChatterApiJob({ start, end, bodyshopIds, skipUpload, sessionUtils }) {
logger.log("chatter-api-start", "DEBUG", "api", null, null);
const allErrors = [];
const allShopSummaries = [];
// Shops that DO have chatter_company_id
const { bodyshops } = await client.request(queries.GET_CHATTER_SHOPS_WITH_COMPANY);
const shopsToProcess =
bodyshopIds?.length > 0 ? bodyshops.filter((shop) => bodyshopIds.includes(shop.id)) : bodyshops;
logger.log("chatter-api-shopsToProcess-generated", "DEBUG", "api", null, { count: shopsToProcess.length });
if (shopsToProcess.length === 0) {
logger.log("chatter-api-shopsToProcess-empty", "DEBUG", "api", null, null);
return {
totals: { shops: 0, jobs: 0, sent: 0, duplicates: 0, failed: 0 },
allShopSummaries: [],
allErrors: []
};
}
await processBatchApi({
shopsToProcess,
start,
end,
skipUpload,
allShopSummaries,
allErrors,
sessionUtils
});
const totals = allShopSummaries.reduce(
(acc, s) => {
acc.shops += 1;
acc.jobs += s.jobs || 0;
acc.sent += s.sent || 0;
acc.duplicates += s.duplicates || 0;
acc.failed += s.failed || 0;
return acc;
},
{ shops: 0, jobs: 0, sent: 0, duplicates: 0, failed: 0 }
);
logger.log("chatter-api-end", "DEBUG", "api", null, totals);
return { totals, allShopSummaries, allErrors };
}
exports.default = async (req, res) => {
if (process.env.NODE_ENV !== "production") return res.sendStatus(403);
if (req.headers["x-imex-auth"] !== process.env.AUTOHOUSE_AUTH_TOKEN) return res.sendStatus(401);
res.status(202).json({
success: true,
message: "Processing Chatter-API Cron request ...",
message: "Chatter API job queued for processing",
timestamp: new Date().toISOString()
});
try {
logger.log("chatter-api-start", "DEBUG", "api", null, null);
const { dispatchChatterApiJob } = require("./queues/chatterApiQueue");
const { start, end, bodyshopIds, skipUpload } = req.body;
const allErrors = [];
const allShopSummaries = [];
// Shops that DO have chatter_company_id
const { bodyshops } = await client.request(queries.GET_CHATTER_SHOPS_WITH_COMPANY);
const specificShopIds = req.body.bodyshopIds;
const { start, end, skipUpload } = req.body; // keep same flag; now acts like "dry run"
const shopsToProcess =
specificShopIds?.length > 0 ? bodyshops.filter((shop) => specificShopIds.includes(shop.id)) : bodyshops;
logger.log("chatter-api-shopsToProcess-generated", "DEBUG", "api", null, { count: shopsToProcess.length });
if (shopsToProcess.length === 0) {
logger.log("chatter-api-shopsToProcess-empty", "DEBUG", "api", null, null);
return;
}
await processBatchApi({
shopsToProcess,
await dispatchChatterApiJob({
start,
end,
skipUpload,
allShopSummaries,
allErrors,
sessionUtils: req.sessionUtils
bodyshopIds,
skipUpload
});
const totals = allShopSummaries.reduce(
(acc, s) => {
acc.shops += 1;
acc.jobs += s.jobs || 0;
acc.sent += s.sent || 0;
acc.duplicates += s.duplicates || 0;
acc.failed += s.failed || 0;
return acc;
},
{ shops: 0, jobs: 0, sent: 0, duplicates: 0, failed: 0 }
);
await sendServerEmail({
subject: `Chatter API Report ${moment().format("MM-DD-YY")}`,
text:
`Totals:\n${JSON.stringify(totals, null, 2)}\n\n` +
`Shop summaries:\n${JSON.stringify(allShopSummaries, null, 2)}\n\n` +
`Errors:\n${JSON.stringify(allErrors, null, 2)}\n`
});
logger.log("chatter-api-end", "DEBUG", "api", null, totals);
} catch (error) {
logger.log("chatter-api-error", "ERROR", "api", null, { error: error.message, stack: error.stack });
logger.log("chatter-api-queue-dispatch-error", "ERROR", "api", null, {
error: error.message,
stack: error.stack
});
}
};
exports.processChatterApiJob = processChatterApiJob;
async function processBatchApi({ shopsToProcess, start, end, skipUpload, allShopSummaries, allErrors, sessionUtils }) {
for (const bodyshop of shopsToProcess) {
const summary = {