feat(channels): implement ChannelMapping, SyndicationJob models, transformation service and syndication API endpoints

2026-08-12 13:23:04 +05:30
parent 6cdef6da32
commit c5821f8015
7 changed files with 424 additions and 1 deletions
@@ -1,4 +1,6 @@
 import service from './channel.service.js';
+import channelMappingService from '../mappings/channelMapping.service.js';
+import syndicationService from '../syndication/syndication.service.js';

 export class ChannelController {
  async getAll(req, res, next) {
@@ -63,6 +65,53 @@ export class ChannelController {
      next(error);
    }
  }
+
+  // Channel Field Mappings
+  async getMappings(req, res, next) {
+    try {
+      const data = await channelMappingService.getByChannel(req.params.id, req.user);
+      return res.status(200).json({ success: true, data });
+    } catch (error) {
+      next(error);
+    }
+  }
+
+  async upsertMappings(req, res, next) {
+    try {
+      const data = await channelMappingService.upsertMappings(req.params.id, req.body.mappings || [], req.user);
+      return res.status(200).json({ success: true, data, message: 'Mapping rules updated successfully' });
+    } catch (error) {
+      next(error);
+    }
+  }
+
+  // Syndication Engine
+  async triggerSyndication(req, res, next) {
+    try {
+      const data = await syndicationService.triggerSyndication(req.params.id, req.user);
+      return res.status(200).json({ success: true, data, message: 'Syndication job triggered successfully' });
+    } catch (error) {
+      next(error);
+    }
+  }
+
+  async getJobs(req, res, next) {
+    try {
+      const data = await syndicationService.getJobsByChannel(req.params.id);
+      return res.status(200).json({ success: true, data });
+    } catch (error) {
+      next(error);
+    }
+  }
+
+  async getJobById(req, res, next) {
+    try {
+      const data = await syndicationService.getJobById(req.params.jobId);
+      return res.status(200).json({ success: true, data });
+    } catch (error) {
+      next(error);
+    }
+  }
 }

 export default new ChannelController();
@@ -143,4 +143,43 @@ router.post(
  controller.restore
 );

+// Channel Field Mappings
+router.get(
+  '/:id/mappings',
+  authenticate,
+  authorize(['settings.integrations']),
+  controller.getMappings
+);
+
+router.put(
+  '/:id/mappings',
+  authenticate,
+  authorize(['settings.integrations']),
+  audit('UPDATE_CHANNEL_MAPPINGS'),
+  controller.upsertMappings
+);
+
+// Syndication Jobs & Execution
+router.post(
+  '/:id/syndicate',
+  authenticate,
+  authorize(['settings.integrations']),
+  audit('TRIGGER_CHANNEL_SYNDICATION'),
+  controller.triggerSyndication
+);
+
+router.get(
+  '/:id/jobs',
+  authenticate,
+  authorize(['settings.integrations']),
+  controller.getJobs
+);
+
+router.get(
+  '/jobs/:jobId',
+  authenticate,
+  authorize(['settings.integrations']),
+  controller.getJobById
+);
+
 export default router;
@@ -0,0 +1,63 @@
+import { DataTypes } from 'sequelize';
+
+export default (sequelize) => {
+  const ChannelMapping = sequelize.define('ChannelMapping', {
+    id: {
+      type: DataTypes.UUID,
+      defaultValue: DataTypes.UUIDV4,
+      primaryKey: true,
+    },
+    tenant_id: {
+      type: DataTypes.INTEGER,
+      allowNull: true,
+      references: {
+        model: 'tenants',
+        key: 'id'
+      }
+    },
+    channel_id: {
+      type: DataTypes.UUID,
+      allowNull: false,
+      references: {
+        model: 'channels',
+        key: 'id'
+      },
+      onDelete: 'CASCADE'
+    },
+    pim_attribute_code: {
+      type: DataTypes.STRING(100),
+      allowNull: false,
+    },
+    channel_field_code: {
+      type: DataTypes.STRING(100),
+      allowNull: false,
+    },
+    transformation_rule: {
+      type: DataTypes.STRING(50),
+      defaultValue: 'none',
+      allowNull: false,
+      comment: 'none, uppercase, lowercase, currency_format, strip_html, default_if_null'
+    },
+    default_value: {
+      type: DataTypes.TEXT,
+      allowNull: true,
+    },
+    is_required: {
+      type: DataTypes.BOOLEAN,
+      defaultValue: false,
+    }
+  }, {
+    tableName: 'channel_mappings',
+    timestamps: true,
+    createdAt: 'created_at',
+    updatedAt: 'updated_at',
+  });
+
+  ChannelMapping.associate = (models) => {
+    if (models.Channel) {
+      ChannelMapping.belongsTo(models.Channel, { foreignKey: 'channel_id', as: 'channel' });
+    }
+  };
+
+  return ChannelMapping;
+};
@@ -0,0 +1,51 @@
+import { models } from '../../../shared/database/models.js';
+import { ApiError } from '../../../utils/helpers/ApiError.utils.js';
+import { AuditService } from '../../../shared/services/audit.service.js';
+
+export class ChannelMappingService {
+  async getByChannel(channelId, userContext = {}) {
+    const channel = await models.Channel.findByPk(channelId);
+    if (!channel) {
+      throw new ApiError(404, 'Channel not found');
+    }
+
+    return await models.ChannelMapping.findAll({
+      where: { channel_id: channelId },
+      order: [['created_at', 'ASC']]
+    });
+  }
+
+  async upsertMappings(channelId, mappingsArray, userContext = {}) {
+    const channel = await models.Channel.findByPk(channelId);
+    if (!channel) {
+      throw new ApiError(404, 'Channel not found');
+    }
+
+    // Delete existing mappings for this channel and bulk insert new rules
+    await models.ChannelMapping.destroy({ where: { channel_id: channelId } });
+
+    const records = mappingsArray.map(item => ({
+      tenant_id: userContext.tenantId || channel.tenant_id || null,
+      channel_id: channelId,
+      pim_attribute_code: item.pim_attribute_code,
+      channel_field_code: item.channel_field_code,
+      transformation_rule: item.transformation_rule || 'none',
+      default_value: item.default_value || null,
+      is_required: Boolean(item.is_required)
+    }));
+
+    const created = await models.ChannelMapping.bulkCreate(records);
+
+    await AuditService.log({
+      action: 'UPDATE_MAPPINGS',
+      resource: 'ChannelMapping',
+      resourceId: channelId,
+      userId: userContext.userId || 'system',
+      details: { count: created.length }
+    });
+
+    return created;
+  }
+}
+
+export default new ChannelMappingService();
@@ -0,0 +1,142 @@
+import { models } from '../../../shared/database/models.js';
+import { ApiError } from '../../../utils/helpers/ApiError.utils.js';
+import { AuditService } from '../../../shared/services/audit.service.js';
+
+export class SyndicationService {
+  applyTransformation(val, rule, defaultValue) {
+    if (val === null || val === undefined || val === '') {
+      return defaultValue !== undefined && defaultValue !== null ? defaultValue : '';
+    }
+    const str = String(val);
+    switch (rule) {
+      case 'uppercase':
+        return str.toUpperCase();
+      case 'lowercase':
+        return str.toLowerCase();
+      case 'currency_format':
+        const num = parseFloat(str) || 0;
+        return num.toFixed(2);
+      case 'strip_html':
+        return str.replace(/<[^>]*>?/gm, '');
+      case 'default_if_null':
+        return str || defaultValue || '';
+      case 'none':
+      default:
+        return str;
+    }
+  }
+
+  async triggerSyndication(channelId, userContext = {}) {
+    const channel = await models.Channel.findByPk(channelId);
+    if (!channel) {
+      throw new ApiError(404, 'Channel not found');
+    }
+
+    const mappings = await models.ChannelMapping.findAll({
+      where: { channel_id: channelId }
+    });
+
+    const tenantId = userContext.tenantId || channel.tenant_id || null;
+
+    // Fetch tenant products
+    const where = {};
+    if (tenantId) where.tenant_id = tenantId;
+
+    const products = await models.Product.findAll({
+      where,
+      limit: 100
+    });
+
+    const job = await models.SyndicationJob.create({
+      tenant_id: tenantId,
+      channel_id: channelId,
+      status: 'running',
+      triggered_by: userContext.userId || null,
+      total_products: products.length,
+      success_count: 0,
+      failed_count: 0,
+      error_log: [],
+      started_at: new Date()
+    });
+
+    let successCount = 0;
+    let failedCount = 0;
+    const errorLogs = [];
+
+    for (const prod of products) {
+      try {
+        const transformedPayload = {};
+        let hasError = false;
+
+        for (const mapItem of mappings) {
+          const rawVal = prod[mapItem.pim_attribute_code];
+          if (mapItem.is_required && (rawVal === null || rawVal === undefined || rawVal === '')) {
+            errorLogs.push({
+              productId: prod.id,
+              sku: prod.sku,
+              error: `Required attribute "${mapItem.pim_attribute_code}" is missing or null`
+            });
+            hasError = true;
+            break;
+          }
+          transformedPayload[mapItem.channel_field_code] = this.applyTransformation(
+            rawVal,
+            mapItem.transformation_rule,
+            mapItem.default_value
+          );
+        }
+
+        if (hasError) {
+          failedCount++;
+        } else {
+          successCount++;
+        }
+      } catch (err) {
+        failedCount++;
+        errorLogs.push({
+          productId: prod.id,
+          sku: prod.sku,
+          error: err.message
+        });
+      }
+    }
+
+    const finalStatus = failedCount > 0 ? (successCount > 0 ? 'completed' : 'failed') : 'completed';
+
+    await job.update({
+      status: finalStatus,
+      success_count: successCount,
+      failed_count: failedCount,
+      error_log: errorLogs,
+      completed_at: new Date()
+    });
+
+    await AuditService.log({
+      action: 'SYNDICATE_CHANNEL',
+      resource: 'Channel',
+      resourceId: channelId,
+      userId: userContext.userId || 'system',
+      details: { jobId: job.id, status: finalStatus, total: products.length, success: successCount, failed: failedCount }
+    });
+
+    return job;
+  }
+
+  async getJobsByChannel(channelId) {
+    return await models.SyndicationJob.findAll({
+      where: { channel_id: channelId },
+      order: [['created_at', 'DESC']],
+      limit: 50
+    });
+  }
+
+  async getJobById(jobId) {
+    const job = await models.SyndicationJob.findByPk(jobId);
+    if (!job) {
+      throw new ApiError(404, 'Syndication job not found');
+    }
+    return job;
+  }
+}
+
+export default new SyndicationService();
@@ -0,0 +1,75 @@
+import { DataTypes } from 'sequelize';
+
+export default (sequelize) => {
+  const SyndicationJob = sequelize.define('SyndicationJob', {
+    id: {
+      type: DataTypes.UUID,
+      defaultValue: DataTypes.UUIDV4,
+      primaryKey: true,
+    },
+    tenant_id: {
+      type: DataTypes.INTEGER,
+      allowNull: true,
+      references: {
+        model: 'tenants',
+        key: 'id'
+      }
+    },
+    channel_id: {
+      type: DataTypes.UUID,
+      allowNull: false,
+      references: {
+        model: 'channels',
+        key: 'id'
+      },
+      onDelete: 'CASCADE'
+    },
+    status: {
+      type: DataTypes.ENUM('pending', 'running', 'completed', 'failed'),
+      defaultValue: 'pending',
+      allowNull: false,
+    },
+    triggered_by: {
+      type: DataTypes.INTEGER,
+      allowNull: true,
+    },
+    total_products: {
+      type: DataTypes.INTEGER,
+      defaultValue: 0,
+    },
+    success_count: {
+      type: DataTypes.INTEGER,
+      defaultValue: 0,
+    },
+    failed_count: {
+      type: DataTypes.INTEGER,
+      defaultValue: 0,
+    },
+    error_log: {
+      type: DataTypes.JSONB,
+      allowNull: true,
+      defaultValue: [],
+    },
+    started_at: {
+      type: DataTypes.DATE,
+      allowNull: true,
+    },
+    completed_at: {
+      type: DataTypes.DATE,
+      allowNull: true,
+    }
+  }, {
+    tableName: 'syndication_jobs',
+    timestamps: true,
+    createdAt: 'created_at',
+    updatedAt: 'updated_at',
+  });
+
+  SyndicationJob.associate = (models) => {
+    if (models.Channel) {
+      SyndicationJob.belongsTo(models.Channel, { foreignKey: 'channel_id', as: 'channel' });
+    }
+  };
+
+  return SyndicationJob;
+};
@@ -177,6 +177,8 @@ import channelAssetModelInit from '../../features/channels/channels/channelAsset
 // Channel Management Model
 import channelModelInit from '../../features/channels/channels/channel.model.js';
 import channelTypeModelInit from '../../features/channels/channelTypes/channelType.model.js';
+import channelMappingModelInit from '../../features/channels/mappings/channelMapping.model.js';
+import syndicationJobModelInit from '../../features/channels/syndication/syndicationJob.model.js';
 import workflowModelInit from '../../features/workflows/workflow.model.js';
 import productAttributeValueModelInit from '../../features/products/products/productAttributeValue.model.js';
 import productVariantValueModelInit from '../../features/products/products/productVariantValue.model.js';
@@ -218,9 +220,11 @@ export const initializeDatabaseModels = () => {
  registerModel('FamilyAssetRequirement', familyAssetRequirementModelInit);
  registerModel('FamilyChannel', familyChannelModelInit);

-  // Channels
+  // Channels & Syndication
  registerModel('Channel', channelModelInit);
  registerModel('ChannelType', channelTypeModelInit);
+  registerModel('ChannelMapping', channelMappingModelInit);
+  registerModel('SyndicationJob', syndicationJobModelInit);
  registerModel('WorkflowRegistry', workflowModelInit);
  registerModel('ProductAttributeValue', productAttributeValueModelInit);
  registerModel('ProductVariantValue', productVariantValueModelInit);