Tom Butcher
61d29a8f7e
This commit introduces new fields `syncHash` and `syncImageHash` to the `listing` and `listingVariant` schemas, enhancing the tracking of synchronization states. Additionally, a new module for sync fingerprinting is added, which includes functions to generate hashes for listings and their variants. The marketplace integration logic is updated to utilize these hashes for determining if updates are necessary, improving efficiency in syncing operations. Tests are also added to ensure the correctness of the new functionality and its integration with existing systems.
1278 lines
42 KiB
JavaScript
1278 lines
42 KiB
JavaScript
import config from '../config.js';
|
|
import log4js from 'log4js';
|
|
import { listingModel } from '../database/schemas/sales/listing.schema.js';
|
|
import { listingVarientModel } from '../database/schemas/sales/listingvarient.schema.js';
|
|
import { marketplaceModel } from '../database/schemas/sales/marketplace.schema.js';
|
|
import { shipmentModel } from '../database/schemas/inventory/shipment.schema.js';
|
|
import { paymentPolicyModel } from '../database/schemas/finance/paymentpolicy.schema.js';
|
|
import { returnPolicyModel } from '../database/schemas/sales/returnpolicy.schema.js';
|
|
import { fulfillmentPolicyModel } from '../database/schemas/sales/fulfillmentpolicy.schema.js';
|
|
import { taxRateModel } from '../database/schemas/management/taxrate.schema.js';
|
|
import { persistMarketplaceMapping } from './marketplaces/ebay/accountPolicies.js';
|
|
import { editObject } from '../database/database.js';
|
|
import { dbConnect } from '../database/mongo.js';
|
|
import { redisServer } from '../database/redis.js';
|
|
import { natsServer } from '../database/nats.js';
|
|
import * as tiktokShop from './marketplaces/tiktokShop.js';
|
|
import * as ebay from './marketplaces/ebay/index.js';
|
|
import {
|
|
marketplaceActor,
|
|
marketplaceSku,
|
|
upsertExternalOrder,
|
|
importExternalItems,
|
|
applyWebhookAction,
|
|
pushShipmentFulfillment,
|
|
} from './marketplaceSync.js';
|
|
import {
|
|
imageFilesHash,
|
|
isUnsyncedDraftListing,
|
|
listingImagesUnchanged,
|
|
listingSyncHash,
|
|
listingSyncUnchanged,
|
|
} from './marketplaces/syncFingerprint.js';
|
|
|
|
const logger = log4js.getLogger('Marketplace Worker');
|
|
logger.level = config.server.logLevel;
|
|
|
|
const providers = {
|
|
tiktokShop,
|
|
ebay,
|
|
};
|
|
|
|
function getProvider(marketplace) {
|
|
const provider = providers[marketplace.provider];
|
|
if (!provider) {
|
|
throw new Error(`No integration available for provider: ${marketplace.provider}`);
|
|
}
|
|
return provider;
|
|
}
|
|
|
|
export function hasIntegration(provider) {
|
|
return !!providers[provider];
|
|
}
|
|
|
|
export async function publishMarketplaceOfferForSku(marketplace, user, sku, listing, varient) {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const provider = getProvider(authenticatedMarketplace);
|
|
if (typeof provider.publishOfferForSku !== 'function') {
|
|
throw new Error(
|
|
`Marketplace provider "${marketplace.provider}" does not support publishing offers`
|
|
);
|
|
}
|
|
return provider.publishOfferForSku(authenticatedMarketplace, sku, listing, varient);
|
|
}
|
|
|
|
export async function ensureMarketplaceListingInventory(marketplace, user, listing, varients = []) {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const provider = getProvider(authenticatedMarketplace);
|
|
if (
|
|
typeof provider.updateItem !== 'function' ||
|
|
typeof provider.publishOfferForSku !== 'function'
|
|
) {
|
|
return null;
|
|
}
|
|
const fullListing = listing?._id ? await fetchFullListing(listing._id) : listing;
|
|
if (!fullListing) throw new Error('Listing not found');
|
|
const listingVarients =
|
|
varients.length > 0 ? varients : listing?._id ? await fetchListingVarients(listing._id) : [];
|
|
if (listingSyncUnchanged(fullListing, listingVarients)) {
|
|
logger.info(
|
|
`Listing "${fullListing._reference}" inventory unchanged — skipping eBay inventory sync`
|
|
);
|
|
return {
|
|
skipped: true,
|
|
url: fullListing.url,
|
|
externalReference: fullListing.externalReference,
|
|
};
|
|
}
|
|
return provider.updateItem(authenticatedMarketplace, fullListing, listingVarients);
|
|
}
|
|
|
|
export async function syncMarketplaceListingImages(marketplace, user, listing, varients = []) {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const provider = getProvider(authenticatedMarketplace);
|
|
if (typeof provider.syncListingImages !== 'function') {
|
|
return null;
|
|
}
|
|
const fullListing = listing?._id ? await fetchFullListing(listing._id) : listing;
|
|
if (!fullListing) throw new Error('Listing not found');
|
|
const listingVarients =
|
|
varients.length > 0 ? varients : listing?._id ? await fetchListingVarients(listing._id) : [];
|
|
if (listingImagesUnchanged(fullListing, listingVarients)) {
|
|
logger.debug(
|
|
`Listing "${fullListing._reference}" images unchanged — skipping eBay image sync`
|
|
);
|
|
return { skipped: true, listing: fullListing, varients: listingVarients };
|
|
}
|
|
return provider.syncListingImages(authenticatedMarketplace, fullListing, listingVarients);
|
|
}
|
|
|
|
export async function withdrawMarketplaceOfferForSku(marketplace, user, sku, listing) {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const provider = getProvider(authenticatedMarketplace);
|
|
if (typeof provider.withdrawOfferForSku !== 'function') {
|
|
throw new Error(
|
|
`Marketplace provider "${marketplace.provider}" does not support withdrawing offers`
|
|
);
|
|
}
|
|
return provider.withdrawOfferForSku(authenticatedMarketplace, sku, listing);
|
|
}
|
|
|
|
async function persistMarketplaceUpdate(marketplace, configUpdates, marketplaceUpdates, user) {
|
|
const updateData = { updatedAt: new Date() };
|
|
|
|
if (configUpdates && Object.keys(configUpdates).length > 0) {
|
|
updateData.config = {
|
|
...(marketplace.config || {}),
|
|
...configUpdates,
|
|
};
|
|
}
|
|
|
|
if (marketplaceUpdates && Object.keys(marketplaceUpdates).length > 0) {
|
|
Object.assign(updateData, marketplaceUpdates);
|
|
}
|
|
|
|
if (Object.keys(updateData).length <= 1) {
|
|
return marketplace;
|
|
}
|
|
|
|
return editObject({
|
|
model: marketplaceModel,
|
|
id: marketplace._id,
|
|
updateData,
|
|
user,
|
|
});
|
|
}
|
|
|
|
async function ensureMarketplaceAuth(marketplace, user) {
|
|
const provider = getProvider(marketplace);
|
|
|
|
if (!provider.ensureAuthenticatedMarketplace) {
|
|
return marketplace;
|
|
}
|
|
|
|
const authResult = await provider.ensureAuthenticatedMarketplace(marketplace);
|
|
if (!authResult?.configUpdates) {
|
|
return authResult?.marketplace || marketplace;
|
|
}
|
|
|
|
return persistMarketplaceUpdate(
|
|
marketplace,
|
|
authResult.configUpdates,
|
|
authResult.marketplaceUpdates,
|
|
user
|
|
);
|
|
}
|
|
|
|
export function canAuthorize(marketplace) {
|
|
const provider = getProvider(marketplace);
|
|
return (
|
|
typeof provider.createAuthorizationUrl === 'function' &&
|
|
typeof provider.exchangeAuthorizationCode === 'function'
|
|
);
|
|
}
|
|
|
|
export function getAuthorizationUrl(marketplace, { state } = {}) {
|
|
const provider = getProvider(marketplace);
|
|
if (!provider.createAuthorizationUrl) {
|
|
throw new Error(`Provider ${marketplace.provider} does not support marketplace authorization`);
|
|
}
|
|
|
|
return provider.createAuthorizationUrl(marketplace, { state });
|
|
}
|
|
|
|
export async function exchangeAuthorizationCode(marketplace, user, { code, state } = {}) {
|
|
const provider = getProvider(marketplace);
|
|
if (!provider.exchangeAuthorizationCode) {
|
|
throw new Error(`Provider ${marketplace.provider} does not support marketplace authorization`);
|
|
}
|
|
|
|
const authResult = await provider.exchangeAuthorizationCode(marketplace, { code, state });
|
|
const updatedMarketplace = await persistMarketplaceUpdate(
|
|
marketplace,
|
|
authResult?.configUpdates || {},
|
|
authResult?.marketplaceUpdates,
|
|
user
|
|
);
|
|
|
|
let marketplaceWithAuth = updatedMarketplace;
|
|
try {
|
|
marketplaceWithAuth = await syncMarketplaceMetadata(marketplaceWithAuth, user);
|
|
} catch (err) {
|
|
logger.warn(
|
|
`Failed to sync marketplace metadata after authorization for "${marketplace.name}": ${err.message}`
|
|
);
|
|
}
|
|
|
|
return {
|
|
marketplace: marketplaceWithAuth,
|
|
...(authResult?.data ? { data: authResult.data } : {}),
|
|
};
|
|
}
|
|
|
|
export async function refreshMarketplaceAuth(marketplace, user) {
|
|
const provider = getProvider(marketplace);
|
|
if (!provider.refreshAuth) {
|
|
throw new Error(`Provider ${marketplace.provider} does not support token refresh`);
|
|
}
|
|
|
|
const authResult = await provider.refreshAuth(marketplace);
|
|
const updatedMarketplace = await persistMarketplaceUpdate(
|
|
marketplace,
|
|
authResult?.configUpdates || {},
|
|
authResult?.marketplaceUpdates,
|
|
user
|
|
);
|
|
|
|
return {
|
|
marketplace: updatedMarketplace,
|
|
...(authResult?.data ? { data: authResult.data } : {}),
|
|
};
|
|
}
|
|
|
|
export function canVerifyWebhookSignature(marketplace) {
|
|
const provider = getProvider(marketplace);
|
|
if (
|
|
typeof provider.verifyWebhookSignature !== 'function' &&
|
|
typeof provider.verifyNotificationSignature !== 'function'
|
|
) {
|
|
return false;
|
|
}
|
|
|
|
if (typeof provider.canVerifyNotificationSignature === 'function') {
|
|
return provider.canVerifyNotificationSignature(marketplace);
|
|
}
|
|
|
|
if (typeof provider.canVerifyWebhookSignature === 'function') {
|
|
return provider.canVerifyWebhookSignature(marketplace);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
export function verifyWebhookSignature(marketplace, rawBody, signature) {
|
|
const provider = getProvider(marketplace);
|
|
if (typeof provider.verifyNotificationSignature === 'function') {
|
|
return provider.verifyNotificationSignature(marketplace, rawBody, signature);
|
|
}
|
|
if (!provider.verifyWebhookSignature) {
|
|
logger.warn(`Provider ${marketplace.provider} does not support webhook signature verification`);
|
|
return true;
|
|
}
|
|
|
|
return provider.verifyWebhookSignature(marketplace, rawBody, signature);
|
|
}
|
|
|
|
export async function handleWebhook(marketplace, event, { rawBody, signature } = {}) {
|
|
const provider = getProvider(marketplace);
|
|
const actor = marketplaceActor(marketplace);
|
|
|
|
if (signature && (await canVerifyWebhookSignature(marketplace))) {
|
|
const valid = await verifyWebhookSignature(
|
|
marketplace,
|
|
rawBody || JSON.stringify(event),
|
|
signature
|
|
);
|
|
if (!valid) {
|
|
const error = new Error('Invalid webhook signature');
|
|
error.status = 401;
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
const classified = await provider.handleWebhook(marketplace, event);
|
|
return applyWebhookAction(marketplace, provider, classified, actor);
|
|
}
|
|
|
|
export function buildWebhookChallengeResponse(marketplace, query) {
|
|
const provider = getProvider(marketplace);
|
|
if (typeof provider.buildWebhookChallengeResponse !== 'function') {
|
|
return null;
|
|
}
|
|
return provider.buildWebhookChallengeResponse(marketplace, query);
|
|
}
|
|
|
|
export async function syncMarketplaceMetadata(marketplace, user) {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const provider = getProvider(authenticatedMarketplace);
|
|
if (typeof provider.syncMarketplaceMetadata !== 'function') {
|
|
return authenticatedMarketplace;
|
|
}
|
|
const metadataUpdates = await provider.syncMarketplaceMetadata(authenticatedMarketplace);
|
|
if (!metadataUpdates || !Object.keys(metadataUpdates).length) {
|
|
return authenticatedMarketplace;
|
|
}
|
|
return persistMarketplaceUpdate(authenticatedMarketplace, {}, metadataUpdates, user);
|
|
}
|
|
|
|
export async function syncMarketplacePolicySet(marketplace, user, methodName) {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const provider = getProvider(authenticatedMarketplace);
|
|
if (typeof provider[methodName] !== 'function') {
|
|
throw new Error(
|
|
`Marketplace provider "${marketplace.provider}" does not support ${methodName}`
|
|
);
|
|
}
|
|
return provider[methodName](authenticatedMarketplace);
|
|
}
|
|
|
|
export async function syncFulfillmentPolicies(marketplace, user) {
|
|
return syncMarketplacePolicySet(marketplace, user, 'syncFulfillmentPolicies');
|
|
}
|
|
|
|
export async function syncPaymentPolicies(marketplace, user) {
|
|
return syncMarketplacePolicySet(marketplace, user, 'syncPaymentPolicies');
|
|
}
|
|
|
|
export async function syncReturnPolicies(marketplace, user) {
|
|
return syncMarketplacePolicySet(marketplace, user, 'syncReturnPolicies');
|
|
}
|
|
|
|
export async function syncTaxRates(marketplace, user) {
|
|
return syncMarketplacePolicySet(marketplace, user, 'syncTaxRates');
|
|
}
|
|
|
|
async function syncOutboundAccountPolicy(
|
|
marketplace,
|
|
user,
|
|
{ policyId, model, populate, methodName, label }
|
|
) {
|
|
const policy = await model.findById(policyId).populate(populate).lean();
|
|
if (!policy) throw new Error(`${label} not found`);
|
|
|
|
try {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const provider = getProvider(authenticatedMarketplace);
|
|
if (typeof provider[methodName] !== 'function') {
|
|
throw new Error(
|
|
`Marketplace provider "${marketplace.provider}" does not support ${methodName}`
|
|
);
|
|
}
|
|
return provider[methodName](authenticatedMarketplace, policy);
|
|
} catch (err) {
|
|
await persistMarketplaceMapping(model, policy, marketplace, {
|
|
stateType: 'failed',
|
|
message: err.message,
|
|
});
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
export async function syncPaymentPolicyOutbound(marketplace, user, policyId) {
|
|
return syncOutboundAccountPolicy(marketplace, user, {
|
|
policyId,
|
|
model: paymentPolicyModel,
|
|
populate: ['marketplaces.marketplace'],
|
|
methodName: 'ensurePaymentPolicySynced',
|
|
label: 'Payment policy',
|
|
});
|
|
}
|
|
|
|
export async function syncReturnPolicyOutbound(marketplace, user, policyId) {
|
|
return syncOutboundAccountPolicy(marketplace, user, {
|
|
policyId,
|
|
model: returnPolicyModel,
|
|
populate: ['marketplaces.marketplace'],
|
|
methodName: 'ensureReturnPolicySynced',
|
|
label: 'Return policy',
|
|
});
|
|
}
|
|
|
|
export async function syncFulfillmentPolicyOutbound(marketplace, user, policyId) {
|
|
return syncOutboundAccountPolicy(marketplace, user, {
|
|
policyId,
|
|
model: fulfillmentPolicyModel,
|
|
populate: [
|
|
{ path: 'courierServices', populate: ['marketplaces.marketplace'] },
|
|
'marketplaces.marketplace',
|
|
],
|
|
methodName: 'ensureFulfillmentPolicySynced',
|
|
label: 'Fulfillment policy',
|
|
});
|
|
}
|
|
|
|
export async function syncTaxRateOutbound(marketplace, user, policyId) {
|
|
return syncOutboundAccountPolicy(marketplace, user, {
|
|
policyId,
|
|
model: taxRateModel,
|
|
populate: ['marketplaces.marketplace'],
|
|
methodName: 'ensureTaxRateSynced',
|
|
label: 'Tax rate',
|
|
});
|
|
}
|
|
|
|
export async function ensureWebhookSubscriptions(marketplace, user) {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const provider = getProvider(authenticatedMarketplace);
|
|
if (typeof provider.ensureWebhookSubscriptions !== 'function') {
|
|
return { skipped: true };
|
|
}
|
|
return provider.ensureWebhookSubscriptions(authenticatedMarketplace);
|
|
}
|
|
|
|
export async function debugMarketplaceGet(marketplace, user, path, params = {}) {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const provider = getProvider(authenticatedMarketplace);
|
|
if (typeof provider.debugGet !== 'function') {
|
|
throw new Error(`Provider ${marketplace.provider} does not support debug GET`);
|
|
}
|
|
if (!String(path).startsWith('/sell/') && !String(path).startsWith('/commerce/')) {
|
|
throw new Error('Debug proxy only allows /sell and /commerce paths');
|
|
}
|
|
return provider.debugGet({
|
|
marketplace: authenticatedMarketplace,
|
|
path,
|
|
params,
|
|
});
|
|
}
|
|
|
|
function roundProgress(progress) {
|
|
const clamped = Math.min(1, Math.max(0, Number(progress) || 0));
|
|
return Math.round(clamped * 1000) / 1000;
|
|
}
|
|
|
|
async function setListingState(listingId, stateType, user, messageOrOptions) {
|
|
const options =
|
|
typeof messageOrOptions === 'string' ? { message: messageOrOptions } : messageOrOptions || {};
|
|
const state = { type: stateType };
|
|
if (options.message) state.message = options.message;
|
|
if (options.progress != null && Number.isFinite(Number(options.progress))) {
|
|
state.progress = roundProgress(options.progress);
|
|
}
|
|
return editObject({
|
|
model: listingModel,
|
|
id: listingId,
|
|
updateData: { state },
|
|
user,
|
|
recalculate: false,
|
|
});
|
|
}
|
|
|
|
async function setListingVarientState(varientId, stateType, user, message) {
|
|
const state = { type: stateType };
|
|
if (message) state.message = message;
|
|
return editObject({
|
|
model: listingVarientModel,
|
|
id: varientId,
|
|
updateData: { state },
|
|
user,
|
|
recalculate: false,
|
|
});
|
|
}
|
|
|
|
async function setMarketplaceState(marketplaceId, stateType, user, message) {
|
|
const state = { type: stateType };
|
|
if (message) state.message = message;
|
|
return editObject({
|
|
model: marketplaceModel,
|
|
id: marketplaceId,
|
|
updateData: { state },
|
|
user,
|
|
recalculate: false,
|
|
});
|
|
}
|
|
|
|
async function recalculateMarketplaceState(marketplace, user) {
|
|
await marketplaceModel.recalculate(marketplace, user);
|
|
}
|
|
|
|
async function fetchFullListing(listingId) {
|
|
return listingModel
|
|
.findById(listingId)
|
|
.populate([
|
|
'product',
|
|
'vendor',
|
|
'stockLocation',
|
|
'courierServices',
|
|
{
|
|
path: 'marketplace',
|
|
populate: ['defaultFulfillmentPolicy', 'defaultPaymentPolicy', 'defaultReturnPolicy'],
|
|
},
|
|
{
|
|
path: 'fulfillmentPolicy',
|
|
populate: ['courierServices', 'marketplaces.marketplace'],
|
|
},
|
|
{
|
|
path: 'paymentPolicy',
|
|
populate: ['marketplaces.marketplace'],
|
|
},
|
|
{
|
|
path: 'returnPolicy',
|
|
populate: ['marketplaces.marketplace'],
|
|
},
|
|
'listingImages',
|
|
])
|
|
.lean();
|
|
}
|
|
|
|
async function fetchListingVarients(listingId) {
|
|
return listingVarientModel.find({ listing: listingId }).populate('listingImages').lean();
|
|
}
|
|
|
|
function listingStateAfterSync(previousStateType) {
|
|
if (!previousStateType || previousStateType === 'syncing') return 'active';
|
|
if (previousStateType === 'draft') return 'draft';
|
|
return previousStateType;
|
|
}
|
|
|
|
function listingSyncUpdateData(listing, varients, result, stateType) {
|
|
const updateData = {
|
|
lastSyncedAt: new Date(),
|
|
syncHash: listingSyncHash(listing, varients),
|
|
syncImageHash: imageFilesHash(listing.listingImages),
|
|
};
|
|
if (stateType) updateData.state = { type: stateType };
|
|
if (result?.url) updateData.url = result.url;
|
|
if (result?.externalReference) updateData.externalReference = result.externalReference;
|
|
const imageUrls = result?.marketplaceImageUrls?.length
|
|
? result.marketplaceImageUrls
|
|
: listing.marketplaceImageUrls;
|
|
if (imageUrls?.length) updateData.marketplaceImageUrls = imageUrls;
|
|
return updateData;
|
|
}
|
|
|
|
function varientSyncUpdateData(varient, result, stateType) {
|
|
const fromResult = result?.varients?.find((item) => String(item._id) === String(varient._id));
|
|
const updateData = {
|
|
lastSyncedAt: new Date(),
|
|
syncHash: listingSyncHash({ title: varient._reference }, [varient]),
|
|
syncImageHash: imageFilesHash(varient.listingImages),
|
|
};
|
|
if (stateType) updateData.state = { type: stateType };
|
|
if (!varient.externalReference && marketplaceSku(varient)) {
|
|
updateData.externalReference = marketplaceSku(varient);
|
|
}
|
|
const imageUrls = fromResult?.marketplaceImageUrls?.length
|
|
? fromResult.marketplaceImageUrls
|
|
: varient.marketplaceImageUrls;
|
|
if (imageUrls?.length) updateData.marketplaceImageUrls = imageUrls;
|
|
return updateData;
|
|
}
|
|
|
|
async function persistListingSyncMetadata({
|
|
listingId,
|
|
listing,
|
|
varients,
|
|
result,
|
|
user,
|
|
stateType,
|
|
applyVarientState = true,
|
|
}) {
|
|
await editObject({
|
|
model: listingModel,
|
|
id: listingId,
|
|
updateData: listingSyncUpdateData(listing, varients, result, stateType),
|
|
user,
|
|
});
|
|
|
|
for (const varient of varients) {
|
|
await editObject({
|
|
model: listingVarientModel,
|
|
id: varient._id,
|
|
updateData: varientSyncUpdateData(
|
|
varient,
|
|
result,
|
|
applyVarientState ? stateType : undefined
|
|
),
|
|
user,
|
|
}).catch(() => {});
|
|
}
|
|
}
|
|
|
|
export async function createListing(marketplace, user, listingData) {
|
|
const provider = getProvider(marketplace);
|
|
if (!provider.createItem) {
|
|
logger.debug(`Provider ${marketplace.provider} does not support createItem — skipping`);
|
|
return;
|
|
}
|
|
|
|
const fullListing = listingData._id ? await fetchFullListing(listingData._id) : listingData;
|
|
if (!fullListing) throw new Error('Listing not found');
|
|
const varients = listingData._id ? await fetchListingVarients(listingData._id) : [];
|
|
const previousState = fullListing.state?.type || 'draft';
|
|
|
|
if (listingData._id) {
|
|
await setListingState(listingData._id, 'syncing', user).catch((err) =>
|
|
logger.warn(`Failed to set listing syncing state: ${err.message}`)
|
|
);
|
|
}
|
|
|
|
try {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
|
|
logger.info(`Creating listing on marketplace "${marketplace.name}" (${marketplace.provider})`);
|
|
const result = await provider.createItem(authenticatedMarketplace, fullListing, varients);
|
|
|
|
if (listingData._id) {
|
|
await persistListingSyncMetadata({
|
|
listingId: listingData._id,
|
|
listing: fullListing,
|
|
varients,
|
|
result,
|
|
user,
|
|
stateType: previousState === 'active' ? 'active' : 'draft',
|
|
});
|
|
}
|
|
|
|
logger.info(`Background createListing complete for marketplace "${marketplace.name}"`);
|
|
} catch (err) {
|
|
logger.error(
|
|
`Background createListing failed for marketplace "${marketplace.name}": ${err.message}`
|
|
);
|
|
if (listingData._id) {
|
|
await setListingState(listingData._id, 'draft', user, err.message).catch(() => {});
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
export async function updateListing(marketplace, user, listingData) {
|
|
const provider = getProvider(marketplace);
|
|
if (!provider.updateItem) {
|
|
logger.debug(`Provider ${marketplace.provider} does not support updateItem — skipping`);
|
|
return;
|
|
}
|
|
|
|
const fullListing = listingData._id ? await fetchFullListing(listingData._id) : listingData;
|
|
if (!fullListing) throw new Error('Listing not found');
|
|
const varients = listingData._id ? await fetchListingVarients(listingData._id) : [];
|
|
const previousState = fullListing.state?.type || 'active';
|
|
|
|
if (listingSyncUnchanged(fullListing, varients)) {
|
|
logger.info(
|
|
`Listing "${fullListing._reference}" unchanged — skipping marketplace update`
|
|
);
|
|
return { skipped: true };
|
|
}
|
|
|
|
if (listingData._id) {
|
|
await setListingState(listingData._id, 'syncing', user).catch((err) =>
|
|
logger.warn(`Failed to set listing syncing state: ${err.message}`)
|
|
);
|
|
}
|
|
|
|
try {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
|
|
logger.info(`Updating listing on marketplace "${marketplace.name}" (${marketplace.provider})`);
|
|
const result = await provider.updateItem(authenticatedMarketplace, fullListing, varients);
|
|
|
|
if (listingData._id) {
|
|
await persistListingSyncMetadata({
|
|
listingId: listingData._id,
|
|
listing: fullListing,
|
|
varients,
|
|
result,
|
|
user,
|
|
stateType: listingStateAfterSync(previousState),
|
|
});
|
|
}
|
|
|
|
logger.info(`Background updateListing complete for marketplace "${marketplace.name}"`);
|
|
} catch (err) {
|
|
logger.error(
|
|
`Background updateListing failed for marketplace "${marketplace.name}": ${err.message}`
|
|
);
|
|
if (listingData._id) {
|
|
await setListingState(
|
|
listingData._id,
|
|
listingStateAfterSync(previousState),
|
|
user,
|
|
err.message
|
|
).catch(() => {});
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
export async function deleteListing(marketplace, user, listingData) {
|
|
const provider = getProvider(marketplace);
|
|
if (!provider.deleteItem) {
|
|
logger.debug(`Provider ${marketplace.provider} does not support deleteItem — skipping`);
|
|
return;
|
|
}
|
|
|
|
try {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
logger.info(
|
|
`Deleting listing from marketplace "${marketplace.name}" (${marketplace.provider})`
|
|
);
|
|
const varients = listingData?._id
|
|
? await fetchListingVarients(listingData._id).catch(() => listingData.varients || [])
|
|
: listingData?.varients || [];
|
|
await provider.deleteItem(authenticatedMarketplace, listingData, varients);
|
|
logger.info(`Background deleteListing complete for marketplace "${marketplace.name}"`);
|
|
} catch (err) {
|
|
logger.error(
|
|
`Background deleteListing failed for marketplace "${marketplace.name}": ${err.message}`
|
|
);
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
export async function syncItems(marketplace, user) {
|
|
await setMarketplaceState(marketplace._id, 'syncing', user).catch((err) =>
|
|
logger.warn(`Failed to set marketplace syncing state: ${err.message}`)
|
|
);
|
|
|
|
try {
|
|
const provider = getProvider(marketplace);
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const actor = marketplaceActor(authenticatedMarketplace);
|
|
logger.info(
|
|
`Starting item sync for marketplace "${marketplace.name}" (${marketplace.provider})`
|
|
);
|
|
|
|
if (typeof provider.syncMarketplaceMetadata === 'function') {
|
|
try {
|
|
const metadataUpdates = await provider.syncMarketplaceMetadata(authenticatedMarketplace);
|
|
if (metadataUpdates && Object.keys(metadataUpdates).length) {
|
|
Object.assign(
|
|
authenticatedMarketplace,
|
|
await persistMarketplaceUpdate(authenticatedMarketplace, {}, metadataUpdates, user)
|
|
);
|
|
}
|
|
} catch (err) {
|
|
logger.warn(
|
|
`Failed to sync marketplace metadata for "${marketplace.name}": ${err.message}`
|
|
);
|
|
}
|
|
}
|
|
|
|
await importExternalItems(authenticatedMarketplace, provider, actor);
|
|
|
|
const existingListings = await listingModel
|
|
.find({
|
|
marketplace: authenticatedMarketplace._id,
|
|
'state.type': { $ne: 'deleted' },
|
|
})
|
|
.lean();
|
|
|
|
const existingVarients = await listingVarientModel
|
|
.find({
|
|
listing: { $in: existingListings.map((l) => l._id) },
|
|
})
|
|
.lean();
|
|
|
|
const results = [];
|
|
|
|
for (const listing of existingListings) {
|
|
try {
|
|
const listingVarients = existingVarients.filter(
|
|
(varient) => String(varient.listing) === String(listing._id)
|
|
);
|
|
|
|
if (isUnsyncedDraftListing(listing)) {
|
|
results.push({
|
|
_reference: listing._reference,
|
|
action: 'skipped',
|
|
reason: 'draft',
|
|
id: listing._id,
|
|
});
|
|
continue;
|
|
}
|
|
|
|
if (listingSyncUnchanged(listing, listingVarients)) {
|
|
results.push({
|
|
_reference: listing._reference,
|
|
action: 'skipped',
|
|
reason: 'unchanged',
|
|
id: listing._id,
|
|
});
|
|
continue;
|
|
}
|
|
|
|
if (!listingVarients.length) {
|
|
throw new Error('Listing has no varients to sync');
|
|
}
|
|
|
|
await setListingState(listing._id, 'syncing', user).catch(() => {});
|
|
for (const varient of listingVarients) {
|
|
await setListingVarientState(varient._id, 'syncing', user).catch(() => {});
|
|
}
|
|
|
|
const result = await provider.updateItem(
|
|
authenticatedMarketplace,
|
|
listing,
|
|
listingVarients
|
|
);
|
|
|
|
await persistListingSyncMetadata({
|
|
listingId: listing._id,
|
|
listing,
|
|
varients: listingVarients,
|
|
result: {
|
|
...result,
|
|
url: result?.url || listing.url,
|
|
externalReference: result?.externalReference || listing.externalReference,
|
|
},
|
|
user,
|
|
stateType: listing.state?.type || 'draft',
|
|
});
|
|
|
|
results.push({ _reference: listing._reference, action: 'synced', id: listing._id });
|
|
} catch (err) {
|
|
logger.warn(`Failed to sync listing ${listing._reference}: ${err.message}`);
|
|
await setListingState(listing._id, listing.state?.type || 'draft', user, err.message).catch(
|
|
() => {}
|
|
);
|
|
results.push({
|
|
_reference: listing._reference,
|
|
action: 'error',
|
|
error: err.message,
|
|
});
|
|
}
|
|
}
|
|
|
|
logger.info(
|
|
`Item sync complete for marketplace ${marketplace.name}: ${results.length} processed`
|
|
);
|
|
|
|
await recalculateMarketplaceState(marketplace, user);
|
|
} catch (err) {
|
|
logger.error(
|
|
`Background syncItems failed for marketplace "${marketplace.name}": ${err.message}`
|
|
);
|
|
await recalculateMarketplaceState(marketplace, user);
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
export async function syncOrders(marketplace, user, { startTime, endTime } = {}) {
|
|
await setMarketplaceState(marketplace._id, 'syncing', user).catch((err) =>
|
|
logger.warn(`Failed to set marketplace syncing state: ${err.message}`)
|
|
);
|
|
|
|
try {
|
|
const provider = getProvider(marketplace);
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const actor = marketplaceActor(authenticatedMarketplace);
|
|
logger.info(
|
|
`Starting order sync for marketplace "${marketplace.name}" (${marketplace.provider})`
|
|
);
|
|
|
|
const externalOrders = await provider.syncOrders(authenticatedMarketplace, {
|
|
startTime,
|
|
endTime,
|
|
});
|
|
const results = [];
|
|
|
|
for (const externalOrder of externalOrders) {
|
|
try {
|
|
const result = await upsertExternalOrder(
|
|
authenticatedMarketplace,
|
|
provider,
|
|
externalOrder,
|
|
actor
|
|
);
|
|
results.push({
|
|
externalReference: result.externalReference,
|
|
action: result.action,
|
|
id: result.salesOrder?._id,
|
|
});
|
|
} catch (err) {
|
|
logger.warn(`Failed to process order: ${err.message}`);
|
|
results.push({
|
|
externalId: externalOrder.id || externalOrder.orderId,
|
|
action: 'error',
|
|
error: err.message,
|
|
});
|
|
}
|
|
}
|
|
|
|
logger.info(
|
|
`Order sync complete for marketplace ${marketplace.name}: ${results.length} processed`
|
|
);
|
|
|
|
await recalculateMarketplaceState(marketplace, user);
|
|
} catch (err) {
|
|
logger.error(
|
|
`Background syncOrders failed for marketplace "${marketplace.name}": ${err.message}`
|
|
);
|
|
await recalculateMarketplaceState(marketplace, user);
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
export async function pushMarketplaceShipmentFulfillment(marketplace, user, shipment) {
|
|
const authenticatedMarketplace = await ensureMarketplaceAuth(marketplace, user);
|
|
const provider = getProvider(authenticatedMarketplace);
|
|
return pushShipmentFulfillment(
|
|
authenticatedMarketplace,
|
|
provider,
|
|
shipment,
|
|
marketplaceActor(authenticatedMarketplace)
|
|
);
|
|
}
|
|
|
|
function userRef(userId) {
|
|
if (!userId) return null;
|
|
return { _id: userId };
|
|
}
|
|
|
|
async function loadMarketplace(marketplaceId) {
|
|
if (!marketplaceId) throw new Error('Marketplace id is required');
|
|
const marketplace = await marketplaceModel.findById(marketplaceId);
|
|
if (!marketplace) throw new Error('Marketplace not found');
|
|
return marketplace;
|
|
}
|
|
|
|
export async function publishListingOffers({
|
|
listingId,
|
|
userId,
|
|
varientIds,
|
|
restoreStateType = 'draft',
|
|
varientRestoreStateType = 'draft',
|
|
}) {
|
|
const user = userRef(userId);
|
|
const listing = await fetchFullListing(listingId);
|
|
if (!listing) throw new Error('Listing not found');
|
|
|
|
const marketplace = listing.marketplace?._id
|
|
? listing.marketplace
|
|
: await loadMarketplace(listing.marketplace);
|
|
if (!marketplace) throw new Error('Listing has no marketplace');
|
|
|
|
const allVarients = await fetchListingVarients(listingId);
|
|
const selectedIds =
|
|
Array.isArray(varientIds) && varientIds.length > 0
|
|
? new Set(varientIds.map((id) => String(id)))
|
|
: null;
|
|
const toPublish = allVarients.filter((varient) => {
|
|
if (!varient._reference) return false;
|
|
if (selectedIds && !selectedIds.has(String(varient._id))) return false;
|
|
return varient.state?.type !== 'active';
|
|
});
|
|
|
|
if (toPublish.length === 0) {
|
|
throw new Error('No variants to publish (all are already active or missing SKU).');
|
|
}
|
|
|
|
await setListingState(listingId, 'publishing', user, { progress: 0.05 });
|
|
for (const varient of toPublish) {
|
|
await setListingVarientState(varient._id, 'publishing', user).catch(() => {});
|
|
}
|
|
|
|
try {
|
|
await ensureMarketplaceListingInventory(marketplace, user, listing, allVarients);
|
|
await setListingState(listingId, 'publishing', user, { progress: 0.15 });
|
|
|
|
let publishedListingId = listing.externalReference;
|
|
let publishedUrl = listing.url;
|
|
for (let i = 0; i < toPublish.length; i += 1) {
|
|
const varient = toPublish[i];
|
|
const apiResult = await publishMarketplaceOfferForSku(
|
|
marketplace,
|
|
user,
|
|
marketplaceSku(varient),
|
|
listing,
|
|
varient
|
|
);
|
|
await editObject({
|
|
model: listingVarientModel,
|
|
id: varient._id,
|
|
updateData: {
|
|
updatedAt: new Date(),
|
|
state: { type: 'active' },
|
|
lastSyncedAt: new Date(),
|
|
},
|
|
user,
|
|
recalculate: false,
|
|
});
|
|
if (apiResult?.listingId) {
|
|
publishedListingId = apiResult.listingId;
|
|
listing.externalReference = apiResult.listingId;
|
|
} else if (apiResult?.externalReference) {
|
|
publishedListingId = apiResult.externalReference;
|
|
listing.externalReference = apiResult.externalReference;
|
|
}
|
|
if (apiResult?.url) {
|
|
publishedUrl = apiResult.url;
|
|
listing.url = apiResult.url;
|
|
}
|
|
await setListingState(listingId, 'publishing', user, {
|
|
progress: 0.15 + (0.8 * (i + 1)) / toPublish.length,
|
|
});
|
|
}
|
|
|
|
await persistListingSyncMetadata({
|
|
listingId,
|
|
listing,
|
|
varients: allVarients,
|
|
result: {
|
|
url: publishedUrl,
|
|
externalReference: publishedListingId,
|
|
},
|
|
user,
|
|
stateType: 'active',
|
|
applyVarientState: false,
|
|
});
|
|
logger.info(`Background publishListing complete for listing ${listingId}`);
|
|
} catch (err) {
|
|
logger.error(`Background publishListing failed for listing ${listingId}: ${err.message}`);
|
|
await setListingState(listingId, restoreStateType, user, err.message).catch(() => {});
|
|
for (const varient of toPublish) {
|
|
await setListingVarientState(varient._id, varientRestoreStateType, user, err.message).catch(
|
|
() => {}
|
|
);
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
export async function unpublishListingOffers({
|
|
listingId,
|
|
userId,
|
|
varientIds,
|
|
restoreStateType = 'active',
|
|
varientRestoreStateType = 'active',
|
|
}) {
|
|
const user = userRef(userId);
|
|
const listing = await fetchFullListing(listingId);
|
|
if (!listing) throw new Error('Listing not found');
|
|
|
|
const marketplace = listing.marketplace?._id
|
|
? listing.marketplace
|
|
: await loadMarketplace(listing.marketplace);
|
|
if (!marketplace) throw new Error('Listing has no marketplace');
|
|
|
|
const allVarients = await fetchListingVarients(listingId);
|
|
const selectedIds =
|
|
Array.isArray(varientIds) && varientIds.length > 0
|
|
? new Set(varientIds.map((id) => String(id)))
|
|
: null;
|
|
const toUnpublish = allVarients.filter((varient) => {
|
|
if (!varient._reference) return false;
|
|
if (selectedIds && !selectedIds.has(String(varient._id))) return false;
|
|
return varient.state?.type === 'active' || varient.state?.type === 'unpublishing';
|
|
});
|
|
|
|
if (toUnpublish.length === 0) {
|
|
throw new Error('No active variants to unpublish.');
|
|
}
|
|
|
|
await setListingState(listingId, 'unpublishing', user, { progress: 0.05 });
|
|
for (const varient of toUnpublish) {
|
|
await setListingVarientState(varient._id, 'unpublishing', user).catch(() => {});
|
|
}
|
|
|
|
try {
|
|
await syncMarketplaceListingImages(marketplace, user, listing, allVarients);
|
|
await setListingState(listingId, 'unpublishing', user, { progress: 0.15 });
|
|
|
|
for (let i = 0; i < toUnpublish.length; i += 1) {
|
|
const varient = toUnpublish[i];
|
|
await withdrawMarketplaceOfferForSku(marketplace, user, marketplaceSku(varient), listing);
|
|
await editObject({
|
|
model: listingVarientModel,
|
|
id: varient._id,
|
|
updateData: {
|
|
updatedAt: new Date(),
|
|
state: { type: 'draft' },
|
|
lastSyncedAt: new Date(),
|
|
},
|
|
user,
|
|
recalculate: false,
|
|
});
|
|
await setListingState(listingId, 'unpublishing', user, {
|
|
progress: 0.15 + (0.8 * (i + 1)) / toUnpublish.length,
|
|
});
|
|
}
|
|
|
|
const remainingActive = await listingVarientModel.exists({
|
|
listing: listingId,
|
|
'state.type': 'active',
|
|
});
|
|
await editObject({
|
|
model: listingModel,
|
|
id: listingId,
|
|
updateData: {
|
|
updatedAt: new Date(),
|
|
state: { type: remainingActive ? 'active' : 'draft' },
|
|
lastSyncedAt: new Date(),
|
|
},
|
|
user,
|
|
});
|
|
logger.info(`Background unpublishListing complete for listing ${listingId}`);
|
|
} catch (err) {
|
|
logger.error(`Background unpublishListing failed for listing ${listingId}: ${err.message}`);
|
|
await setListingState(listingId, restoreStateType, user, err.message).catch(() => {});
|
|
for (const varient of toUnpublish) {
|
|
await setListingVarientState(varient._id, varientRestoreStateType, user, err.message).catch(
|
|
() => {}
|
|
);
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
export async function runJob(action, payload = {}) {
|
|
const user = userRef(payload.userId);
|
|
|
|
switch (action) {
|
|
case 'createListing': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return createListing(marketplace, user, { _id: payload.listingId });
|
|
}
|
|
case 'updateListing': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return updateListing(marketplace, user, { _id: payload.listingId });
|
|
}
|
|
case 'deleteListing': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return deleteListing(marketplace, user, payload.listing);
|
|
}
|
|
case 'publishListing':
|
|
return publishListingOffers(payload);
|
|
case 'unpublishListing':
|
|
return unpublishListingOffers(payload);
|
|
case 'syncItems': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncItems(marketplace, user);
|
|
}
|
|
case 'syncOrders': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncOrders(marketplace, user, {
|
|
startTime: payload.startTime,
|
|
endTime: payload.endTime,
|
|
});
|
|
}
|
|
case 'syncMarketplaceMetadata': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncMarketplaceMetadata(marketplace, user);
|
|
}
|
|
case 'syncFulfillmentPolicies': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncFulfillmentPolicies(marketplace, user);
|
|
}
|
|
case 'syncPaymentPolicies': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncPaymentPolicies(marketplace, user);
|
|
}
|
|
case 'syncReturnPolicies': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncReturnPolicies(marketplace, user);
|
|
}
|
|
case 'syncTaxRates': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncTaxRates(marketplace, user);
|
|
}
|
|
case 'syncPaymentPolicy': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncPaymentPolicyOutbound(marketplace, user, payload.policyId);
|
|
}
|
|
case 'syncReturnPolicy': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncReturnPolicyOutbound(marketplace, user, payload.policyId);
|
|
}
|
|
case 'syncFulfillmentPolicy': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncFulfillmentPolicyOutbound(marketplace, user, payload.policyId);
|
|
}
|
|
case 'syncTaxRate': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return syncTaxRateOutbound(marketplace, user, payload.policyId);
|
|
}
|
|
case 'ensureWebhookSubscriptions': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return ensureWebhookSubscriptions(marketplace, user);
|
|
}
|
|
case 'pushShipmentFulfillment': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
const shipment = payload.shipmentId
|
|
? await shipmentModel.findById(payload.shipmentId).populate('courierService').lean()
|
|
: payload.shipment;
|
|
if (!shipment) throw new Error('Shipment not found');
|
|
return pushMarketplaceShipmentFulfillment(marketplace, user, shipment);
|
|
}
|
|
case 'getAuthorizationUrl': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return getAuthorizationUrl(marketplace, { state: payload.state });
|
|
}
|
|
case 'exchangeAuthorizationCode': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return exchangeAuthorizationCode(marketplace, user, {
|
|
code: payload.code,
|
|
state: payload.state,
|
|
});
|
|
}
|
|
case 'refreshMarketplaceAuth': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return refreshMarketplaceAuth(marketplace, user);
|
|
}
|
|
case 'handleWebhook': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return handleWebhook(marketplace, payload.event, {
|
|
rawBody: payload.rawBody,
|
|
signature: payload.signature,
|
|
});
|
|
}
|
|
case 'buildWebhookChallengeResponse': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return buildWebhookChallengeResponse(marketplace, payload.query);
|
|
}
|
|
case 'canAuthorize': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return canAuthorize(marketplace);
|
|
}
|
|
case 'canVerifyWebhookSignature': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return canVerifyWebhookSignature(marketplace);
|
|
}
|
|
case 'verifyWebhookSignature': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return verifyWebhookSignature(marketplace, payload.rawBody, payload.signature);
|
|
}
|
|
case 'debugMarketplaceGet': {
|
|
const marketplace = await loadMarketplace(payload.marketplaceId);
|
|
return debugMarketplaceGet(marketplace, user, payload.path, payload.params);
|
|
}
|
|
case 'hasIntegration':
|
|
return hasIntegration(payload.provider);
|
|
default:
|
|
throw new Error(`Unknown marketplace worker action: ${action}`);
|
|
}
|
|
}
|
|
|
|
async function startWorkerProcess() {
|
|
logger.info('Starting marketplace worker process...');
|
|
await dbConnect();
|
|
await redisServer.connect();
|
|
await natsServer.connect();
|
|
|
|
process.on('message', (message) => {
|
|
if (!message || message.type === 'ready') return;
|
|
const { id, action, payload, wait } = message;
|
|
if (!action) return;
|
|
|
|
const job = runJob(action, payload || {});
|
|
if (wait) {
|
|
job
|
|
.then((result) => {
|
|
process.send?.({ id, ok: true, result: result ?? null });
|
|
})
|
|
.catch((err) => {
|
|
logger.error(`Marketplace job "${action}" failed: ${err.message}`);
|
|
process.send?.({ id, ok: false, error: err.message });
|
|
});
|
|
return;
|
|
}
|
|
|
|
job.catch((err) => {
|
|
logger.error(`Background marketplace job "${action}" failed: ${err.message}`);
|
|
});
|
|
});
|
|
|
|
process.send?.({ type: 'ready' });
|
|
logger.info('Marketplace worker process ready');
|
|
}
|
|
|
|
if (process.env.MARKETPLACE_WORKER === '1') {
|
|
startWorkerProcess().catch((err) => {
|
|
logger.error('Marketplace worker failed to start:', err);
|
|
process.exit(1);
|
|
});
|
|
}
|
|
|
|
export { marketplaceSku, marketplaceActor };
|