From 9a3df81ea0ac5ebeb825a0059183e1362a81aba3 Mon Sep 17 00:00:00 2001 From: Mohammed-Fardeen-02 Date: Mon, 31 Aug 2026 19:31:39 +0530 Subject: [PATCH] implemented shopify integration --- package-lock.json | 468 +++++++++++ package.json | 3 + .../integrations/adapters/adapter.registry.js | 12 + .../contracts/channel-adapter.contract.js | 21 + .../adapters/shopify/shopify.adapter.js | 70 ++ .../adapters/shopify/shopify.client.js | 65 ++ .../adapters/shopify/shopify.mapper.js | 59 ++ .../adapters/shopify/shopify.oauth.service.js | 275 +++++++ .../adapters/shopify/shopify.publisher.js | 188 +++++ .../controllers/integration.controller.js | 281 +++++++ .../domain/integration.service.js | 116 +++ .../integrations/domain/rule.service.js | 37 + .../integrations/domain/sync.service.js | 89 +++ src/features/integrations/index.js | 39 +- .../models/channelMapping.model.js | 72 ++ .../models/externalResource.model.js | 73 ++ .../integrations/models/integration.model.js | 82 ++ .../models/integrationCredential.model.js | 53 ++ .../integrations/models/outboxEvent.model.js | 79 ++ .../models/publishingRule.model.js | 48 ++ .../integrations/models/syncAttempt.model.js | 81 ++ .../integrations/models/syncError.model.js | 75 ++ .../integrations/models/syncItem.model.js | 97 +++ .../integrations/models/syncJob.model.js | 72 ++ .../projection/canonical-product.builder.js | 142 ++++ .../integrations/routes/integration.routes.js | 36 + .../integrations/workers/outbox.worker.js | 84 ++ .../integrations/workers/sync.worker.js | 127 +++ ...1-create-syndication-and-outbox-tables.cjs | 751 ++++++++++++++++++ src/shared/database/models.js | 23 + .../infrastructure/cache/cache.service.js | 57 ++ .../infrastructure/cache/redis.client.js | 54 ++ .../infrastructure/http/circuit-breaker.js | 32 + src/shared/infrastructure/http/http.client.js | 64 ++ .../infrastructure/http/rate-limiter.js | 19 + .../infrastructure/http/retry.policy.js | 14 + .../infrastructure/http/sanitization.js | 31 + .../infrastructure/outbox/outbox.service.js | 27 + .../infrastructure/queue/queue.constants.js | 6 + .../infrastructure/queue/queue.service.js | 106 +++ .../infrastructure/secrets/secret.service.js | 47 ++ src/test_integrations_pipeline.js | 48 ++ 42 files changed, 4094 insertions(+), 29 deletions(-) create mode 100644 src/features/integrations/adapters/adapter.registry.js create mode 100644 src/features/integrations/adapters/contracts/channel-adapter.contract.js create mode 100644 src/features/integrations/adapters/shopify/shopify.adapter.js create mode 100644 src/features/integrations/adapters/shopify/shopify.client.js create mode 100644 src/features/integrations/adapters/shopify/shopify.mapper.js create mode 100644 src/features/integrations/adapters/shopify/shopify.oauth.service.js create mode 100644 src/features/integrations/adapters/shopify/shopify.publisher.js create mode 100644 src/features/integrations/controllers/integration.controller.js create mode 100644 src/features/integrations/domain/integration.service.js create mode 100644 src/features/integrations/domain/rule.service.js create mode 100644 src/features/integrations/domain/sync.service.js create mode 100644 src/features/integrations/models/channelMapping.model.js create mode 100644 src/features/integrations/models/externalResource.model.js create mode 100644 src/features/integrations/models/integration.model.js create mode 100644 src/features/integrations/models/integrationCredential.model.js create mode 100644 src/features/integrations/models/outboxEvent.model.js create mode 100644 src/features/integrations/models/publishingRule.model.js create mode 100644 src/features/integrations/models/syncAttempt.model.js create mode 100644 src/features/integrations/models/syncError.model.js create mode 100644 src/features/integrations/models/syncItem.model.js create mode 100644 src/features/integrations/models/syncJob.model.js create mode 100644 src/features/integrations/projection/canonical-product.builder.js create mode 100644 src/features/integrations/routes/integration.routes.js create mode 100644 src/features/integrations/workers/outbox.worker.js create mode 100644 src/features/integrations/workers/sync.worker.js create mode 100644 src/migrations/20260830000001-create-syndication-and-outbox-tables.cjs create mode 100644 src/shared/infrastructure/cache/cache.service.js create mode 100644 src/shared/infrastructure/cache/redis.client.js create mode 100644 src/shared/infrastructure/http/circuit-breaker.js create mode 100644 src/shared/infrastructure/http/http.client.js create mode 100644 src/shared/infrastructure/http/rate-limiter.js create mode 100644 src/shared/infrastructure/http/retry.policy.js create mode 100644 src/shared/infrastructure/http/sanitization.js create mode 100644 src/shared/infrastructure/outbox/outbox.service.js create mode 100644 src/shared/infrastructure/queue/queue.constants.js create mode 100644 src/shared/infrastructure/queue/queue.service.js create mode 100644 src/shared/infrastructure/secrets/secret.service.js create mode 100644 src/test_integrations_pipeline.js diff --git a/package-lock.json b/package-lock.json index ceaeb61..6eb7208 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,7 +9,9 @@ "version": "1.0.0", "dependencies": { "@aws-sdk/client-s3": "^3.1113.0", + "axios": "^1.20.0", "bcrypt": "^6.0.0", + "bullmq": "^6.3.2", "cloudinary": "^2.10.0", "compression": "^1.7.5", "cookie-parser": "^1.4.7", @@ -18,6 +20,7 @@ "express": "^4.21.2", "express-validator": "^7.1.0", "helmet": "^8.0.0", + "ioredis": "^6.0.0", "jsonwebtoken": "^9.0.2", "morgan": "^1.10.0", "multer": "^1.4.5-lts.1", @@ -423,6 +426,12 @@ "dev": true, "license": "MIT" }, + "node_modules/@ioredis/commands": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/@ioredis/commands/-/commands-2.0.0.tgz", + "integrity": "sha512-vrx0AE/T0h7cRZwfo1M39Cr+ZhZrkf0V8mQN75wucKCxCLD9l/VX6no3gFvrLqD1IlG/1LtzWovqEw3t0Vr9zg==", + "license": "MIT" + }, "node_modules/@isaacs/cliui": { "version": "8.0.2", "resolved": "https://registry.npmjs.org/@isaacs/cliui/-/cliui-8.0.2.tgz", @@ -441,6 +450,84 @@ "node": ">=12" } }, + "node_modules/@msgpackr-extract/msgpackr-extract-darwin-arm64": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-darwin-arm64/-/msgpackr-extract-darwin-arm64-3.0.4.tgz", + "integrity": "sha512-LCkGo6JDfaBhgST7UpPWgNgLINpcpabaHfyz5OBx75nUYxBsaEPxjnyNjWpeb/xBup/682QnBfRBy2/LvPutZQ==", + "cpu": [ + "arm64" + ], + "license": "MIT", + "optional": true, + "os": [ + "darwin" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-darwin-x64": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-darwin-x64/-/msgpackr-extract-darwin-x64-3.0.4.tgz", + "integrity": "sha512-zExlW9zUJKZH/tOtVMttwjKa4Xm/3KcNjnE3dPN92uCktwavMxpgCA3MoJK/DOnTWsQgo224OaST27/mPNAf+w==", + "cpu": [ + "x64" + ], + "license": "MIT", + "optional": true, + "os": [ + "darwin" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-linux-arm": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-arm/-/msgpackr-extract-linux-arm-3.0.4.tgz", + "integrity": "sha512-Tg3yX65f5GbtXLkrYEHE5oibZG9epyYWas7FogTTEJeDEF9JlXJzKgXaNhT3UXlTOeA+AfZpYZYZ0uPj7Cfquw==", + "cpu": [ + "arm" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-linux-arm64": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-arm64/-/msgpackr-extract-linux-arm64-3.0.4.tgz", + "integrity": "sha512-dgX0P/9wGPJeHFBG+ZmhgE6bmtMt7NP5CRBGyyktpopdk/mW4POnrpQsSLtKI1dwpc+pPLuXHDh6vvskyQE/sw==", + "cpu": [ + "arm64" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-linux-x64": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-x64/-/msgpackr-extract-linux-x64-3.0.4.tgz", + "integrity": "sha512-8TNXMEjJc3QEy7R/x1INhgiU+XakDAFUzBhaz7+Rbrs8NH5UQeHQxxmzsSBJGyV6I1jW79undiQm8tOI+D+8FQ==", + "cpu": [ + "x64" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-win32-x64": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-win32-x64/-/msgpackr-extract-win32-x64-3.0.4.tgz", + "integrity": "sha512-CmCXPQrkbwExx3j946/PtHWHbYJiCRBRDl4BlkRQcJB/YOwQxJRTpoo7aTsortjgoJ1x7opzTSxn7C+ASSLVjQ==", + "cpu": [ + "x64" + ], + "license": "MIT", + "optional": true, + "os": [ + "win32" + ] + }, "node_modules/@one-ini/wasm": { "version": "0.1.1", "resolved": "https://registry.npmjs.org/@one-ini/wasm/-/wasm-0.1.1.tgz", @@ -655,6 +742,41 @@ "node": ">= 0.6" } }, + "node_modules/agent-base": { + "version": "6.0.2", + "resolved": "https://registry.npmjs.org/agent-base/-/agent-base-6.0.2.tgz", + "integrity": "sha512-RZNwNclF7+MS/8bDg70amg32dyeZGZxiDuQmZxKLAlQjr3jGyLx+4Kkk58UO7D2QdgFIQCovuSuZESne6RG6XQ==", + "license": "MIT", + "dependencies": { + "debug": "4" + }, + "engines": { + "node": ">= 6.0.0" + } + }, + "node_modules/agent-base/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/agent-base/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, "node_modules/ajv": { "version": "8.20.0", "resolved": "https://registry.npmjs.org/ajv/-/ajv-8.20.0.tgz", @@ -749,6 +871,12 @@ "integrity": "sha512-htCUDlxyyCLMgaM3xXg0C0LW2xqfuQ6p05pCEIsXuyQ+a1koYKTuBMzRNwmybfLgvJDMd0r1LTn4+E0Ti6C2AA==", "license": "MIT" }, + "node_modules/asynckit": { + "version": "0.4.0", + "resolved": "https://registry.npmjs.org/asynckit/-/asynckit-0.4.0.tgz", + "integrity": "sha512-Oei9OH4tRh0YqU3GxhX79dM/mwVgvbZJaSNaRk+bshkj0S5cfHcgYakreBjrHwatXKbz+IoIdYLxrKim2MjW0Q==", + "license": "MIT" + }, "node_modules/at-least-node": { "version": "1.0.0", "resolved": "https://registry.npmjs.org/at-least-node/-/at-least-node-1.0.0.tgz", @@ -759,6 +887,18 @@ "node": ">= 4.0.0" } }, + "node_modules/axios": { + "version": "1.20.0", + "resolved": "https://registry.npmjs.org/axios/-/axios-1.20.0.tgz", + "integrity": "sha512-r8aOh8j9cGKpgQAqpzrUHnSIc6a59Y3Xf/cv8sy1DrHCkZHzQGEuoq1tARk6qSyDdtQGSDgpb9kFlruzPvrgwg==", + "license": "MIT", + "dependencies": { + "follow-redirects": "^1.16.0", + "form-data": "^4.0.6", + "https-proxy-agent": "^5.0.1", + "proxy-from-env": "^2.1.0" + } + }, "node_modules/balanced-match": { "version": "4.0.4", "resolved": "https://registry.npmjs.org/balanced-match/-/balanced-match-4.0.4.tgz", @@ -896,6 +1036,42 @@ "integrity": "sha512-E+XQCRwSbaaiChtv6k6Dwgc+bx+Bs6vuKJHHl5kox/BaKbhiXzqQOwK4cO22yElGp2OCmjwVhT3HmxgyPGnJfQ==", "license": "MIT" }, + "node_modules/bullmq": { + "version": "6.3.2", + "resolved": "https://registry.npmjs.org/bullmq/-/bullmq-6.3.2.tgz", + "integrity": "sha512-jW4mEG1JOKewI2IhtuMery+kNhCs+EZp4qeNXeYRfueW7P4NFOStFGHOdsiJE9EO9Abz+JMrYlymb+r7fCbT7A==", + "license": "MIT", + "dependencies": { + "cron-parser": "5.10.0", + "msgpackr": "2.1.0", + "node-abort-controller": "3.1.1", + "semver": "7.8.5", + "tslib": "2.8.1" + }, + "engines": { + "node": ">=14.17.0" + }, + "peerDependencies": { + "bullmq-otel": ">=2.0.0", + "ioredis": ">=5.0.0", + "pg": ">=8.0.0", + "redis": ">=5.0.0" + }, + "peerDependenciesMeta": { + "bullmq-otel": { + "optional": true + }, + "ioredis": { + "optional": true + }, + "pg": { + "optional": true + }, + "redis": { + "optional": true + } + } + }, "node_modules/busboy": { "version": "1.6.0", "resolved": "https://registry.npmjs.org/busboy/-/busboy-1.6.0.tgz", @@ -1079,6 +1255,15 @@ "node": ">=9" } }, + "node_modules/cluster-key-slot": { + "version": "1.1.1", + "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.1.tgz", + "integrity": "sha512-rwHwUfXL40Chm1r08yrhU3qpUvdVlgkKNeyeGPOxnW8/SyVDvgRaed/Uz54AqWNaTCAThlj6QAs3TZcKI0xDEw==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/color": { "version": "5.0.3", "resolved": "https://registry.npmjs.org/color/-/color-5.0.3.tgz", @@ -1154,6 +1339,18 @@ "node": ">=12.20" } }, + "node_modules/combined-stream": { + "version": "1.0.8", + "resolved": "https://registry.npmjs.org/combined-stream/-/combined-stream-1.0.8.tgz", + "integrity": "sha512-FQN4MRfuJeHf7cBbBMJFXhKSDq+2kAArBlmRBvcvFE5BB1HZKXtSFASDhdlz9zOYwxh8lDdnvmMOe/+5cdoEdg==", + "license": "MIT", + "dependencies": { + "delayed-stream": "~1.0.0" + }, + "engines": { + "node": ">= 0.8" + } + }, "node_modules/commander": { "version": "10.0.1", "resolved": "https://registry.npmjs.org/commander/-/commander-10.0.1.tgz", @@ -1292,6 +1489,18 @@ "url": "https://opencollective.com/express" } }, + "node_modules/cron-parser": { + "version": "5.10.0", + "resolved": "https://registry.npmjs.org/cron-parser/-/cron-parser-5.10.0.tgz", + "integrity": "sha512-izNAxJyRWUP8ljBoDSub5WyrVOUlT4SLGShswE7eoRBpp6QUsSycYxLBMJlbshgPBMcPT/nrfgjNY2918ayv2A==", + "license": "MIT", + "dependencies": { + "luxon": "^3.7.2" + }, + "engines": { + "node": ">=18" + } + }, "node_modules/cross-env": { "version": "10.1.0", "resolved": "https://registry.npmjs.org/cross-env/-/cross-env-10.1.0.tgz", @@ -1333,6 +1542,24 @@ "ms": "2.0.0" } }, + "node_modules/delayed-stream": { + "version": "1.0.0", + "resolved": "https://registry.npmjs.org/delayed-stream/-/delayed-stream-1.0.0.tgz", + "integrity": "sha512-ZySD7Nf91aLB0RxL4KGrKHBXl7Eds1DAmEdcoVawXnLD7SDhpNgtuII2aAkg7a7QS41jxPSZ17p4VdGnMHk3MQ==", + "license": "MIT", + "engines": { + "node": ">=0.4.0" + } + }, + "node_modules/denque": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/denque/-/denque-2.1.0.tgz", + "integrity": "sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10" + } + }, "node_modules/depd": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/depd/-/depd-2.0.0.tgz", @@ -1352,6 +1579,16 @@ "npm": "1.2.8000 || >= 1.4.16" } }, + "node_modules/detect-libc": { + "version": "2.1.2", + "resolved": "https://registry.npmjs.org/detect-libc/-/detect-libc-2.1.2.tgz", + "integrity": "sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ==", + "license": "Apache-2.0", + "optional": true, + "engines": { + "node": ">=8" + } + }, "node_modules/doctrine": { "version": "3.0.0", "resolved": "https://registry.npmjs.org/doctrine/-/doctrine-3.0.0.tgz", @@ -1576,6 +1813,21 @@ "node": ">= 0.4" } }, + "node_modules/es-set-tostringtag": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/es-set-tostringtag/-/es-set-tostringtag-2.1.0.tgz", + "integrity": "sha512-j6vWzfrGVfyXxge+O0x5sh6cvxAog0a/4Rdd2K36zCMV5eJ+/+tOAngRO8cODMNWbVRdVlmGZQL2YS3yR8bIUA==", + "license": "MIT", + "dependencies": { + "es-errors": "^1.3.0", + "get-intrinsic": "^1.2.6", + "has-tostringtag": "^1.0.2", + "hasown": "^2.0.2" + }, + "engines": { + "node": ">= 0.4" + } + }, "node_modules/escalade": { "version": "3.2.0", "resolved": "https://registry.npmjs.org/escalade/-/escalade-3.2.0.tgz", @@ -1743,6 +1995,26 @@ "integrity": "sha512-GRnmB5gPyJpAhTQdSZTSp9uaPSvl09KoYcMQtsB9rQoOmzs9dH6ffeccH+Z+cv6P68Hu5bC6JjRh4Ah/mHSNRw==", "license": "MIT" }, + "node_modules/follow-redirects": { + "version": "1.16.0", + "resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.16.0.tgz", + "integrity": "sha512-y5rN/uOsadFT/JfYwhxRS5R7Qce+g3zG97+JrtFZlC9klX/W5hD7iiLzScI4nZqUS7DNUdhPgw4xI8W2LuXlUw==", + "funding": [ + { + "type": "individual", + "url": "https://github.com/sponsors/RubenVerborgh" + } + ], + "license": "MIT", + "engines": { + "node": ">=4.0" + }, + "peerDependenciesMeta": { + "debug": { + "optional": true + } + } + }, "node_modules/foreground-child": { "version": "3.3.1", "resolved": "https://registry.npmjs.org/foreground-child/-/foreground-child-3.3.1.tgz", @@ -1759,6 +2031,22 @@ "url": "https://github.com/sponsors/isaacs" } }, + "node_modules/form-data": { + "version": "4.0.6", + "resolved": "https://registry.npmjs.org/form-data/-/form-data-4.0.6.tgz", + "integrity": "sha512-vKatAh4SlVfgbv+YtmhiRjhEMJsYpsG1Y2rMQtR+SVSbytsSD1YGzDIcrAJmdFec88u/+VoGmxnl+80gL1tRCQ==", + "license": "MIT", + "dependencies": { + "asynckit": "^0.4.0", + "combined-stream": "^1.0.8", + "es-set-tostringtag": "^2.1.0", + "hasown": "^2.0.4", + "mime-types": "^2.1.35" + }, + "engines": { + "node": ">= 6" + } + }, "node_modules/forwarded": { "version": "0.2.0", "resolved": "https://registry.npmjs.org/forwarded/-/forwarded-0.2.0.tgz", @@ -1973,6 +2261,21 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/has-tostringtag": { + "version": "1.0.2", + "resolved": "https://registry.npmjs.org/has-tostringtag/-/has-tostringtag-1.0.2.tgz", + "integrity": "sha512-NqADB8VjPFLM2V0VvHUewwwsw0ZWBaIdgo+ieHtK3hasLz4qeCRjYcqfB6AQrBggRKppKF8L52/VqdVsO47Dlw==", + "license": "MIT", + "dependencies": { + "has-symbols": "^1.0.3" + }, + "engines": { + "node": ">= 0.4" + }, + "funding": { + "url": "https://github.com/sponsors/ljharb" + } + }, "node_modules/hasown": { "version": "2.0.4", "resolved": "https://registry.npmjs.org/hasown/-/hasown-2.0.4.tgz", @@ -2017,6 +2320,42 @@ "url": "https://opencollective.com/express" } }, + "node_modules/https-proxy-agent": { + "version": "5.0.1", + "resolved": "https://registry.npmjs.org/https-proxy-agent/-/https-proxy-agent-5.0.1.tgz", + "integrity": "sha512-dFcAjpTQFgoLMzC2VwU+C/CbS7uRL0lWmxDITmqm7C+7F0Odmj6s9l6alZc6AELXhrnggM2CeWSXHGOdX2YtwA==", + "license": "MIT", + "dependencies": { + "agent-base": "6", + "debug": "4" + }, + "engines": { + "node": ">= 6" + } + }, + "node_modules/https-proxy-agent/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/https-proxy-agent/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, "node_modules/iconv-lite": { "version": "0.4.24", "resolved": "https://registry.npmjs.org/iconv-lite/-/iconv-lite-0.4.24.tgz", @@ -2058,6 +2397,50 @@ "dev": true, "license": "ISC" }, + "node_modules/ioredis": { + "version": "6.0.0", + "resolved": "https://registry.npmjs.org/ioredis/-/ioredis-6.0.0.tgz", + "integrity": "sha512-f+Dtubxfpf6KYFq7WVXJoOLn0bk4TJrMrN9SzeE+jrWrCWj7XX3fA6vkryafhADX+GMymRxgDJDOI33COkJc0w==", + "license": "MIT", + "dependencies": { + "@ioredis/commands": "2.0.0", + "cluster-key-slot": "1.1.1", + "debug": "4.4.3", + "denque": "2.1.0", + "redis-errors": "1.2.0", + "standard-as-callback": "2.1.0" + }, + "engines": { + "node": ">=20.0.0" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/ioredis" + } + }, + "node_modules/ioredis/node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, + "node_modules/ioredis/node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, "node_modules/ipaddr.js": { "version": "1.9.1", "resolved": "https://registry.npmjs.org/ipaddr.js/-/ipaddr.js-1.9.1.tgz", @@ -2388,6 +2771,15 @@ "dev": true, "license": "ISC" }, + "node_modules/luxon": { + "version": "3.7.2", + "resolved": "https://registry.npmjs.org/luxon/-/luxon-3.7.2.tgz", + "integrity": "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==", + "license": "MIT", + "engines": { + "node": ">=12" + } + }, "node_modules/math-intrinsics": { "version": "1.1.0", "resolved": "https://registry.npmjs.org/math-intrinsics/-/math-intrinsics-1.1.0.tgz", @@ -2558,6 +2950,37 @@ "integrity": "sha512-Tpp60P6IUJDTuOq/5Z8cdskzJujfwqfOTkrwIwj7IRISpnkJnT6SyJ4PCPnGMoFjC9ddhal5KVIYtAt97ix05A==", "license": "MIT" }, + "node_modules/msgpackr": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/msgpackr/-/msgpackr-2.1.0.tgz", + "integrity": "sha512-p/pBCVO63CsvvpkomUnNNag6+n38rULuDA6HHe70o2gtC8ODI52foF/4ko2qQcp6OiErJXTmrZeXmsGGHsIQNQ==", + "license": "MIT", + "optionalDependencies": { + "msgpackr-extract": "^3.0.4" + } + }, + "node_modules/msgpackr-extract": { + "version": "3.0.4", + "resolved": "https://registry.npmjs.org/msgpackr-extract/-/msgpackr-extract-3.0.4.tgz", + "integrity": "sha512-4kmO/MdyUIkLIvTPr8VHLil4AtoKIoniWPIEk5+CDy0xnWC84azhSFmuJ7PxZdsYtiP5kEeQsORAVIeMgxT+Hw==", + "hasInstallScript": true, + "license": "MIT", + "optional": true, + "dependencies": { + "node-gyp-build-optional-packages": "5.2.2" + }, + "bin": { + "download-msgpackr-prebuilds": "bin/download-prebuilds.js" + }, + "optionalDependencies": { + "@msgpackr-extract/msgpackr-extract-darwin-arm64": "3.0.4", + "@msgpackr-extract/msgpackr-extract-darwin-x64": "3.0.4", + "@msgpackr-extract/msgpackr-extract-linux-arm": "3.0.4", + "@msgpackr-extract/msgpackr-extract-linux-arm64": "3.0.4", + "@msgpackr-extract/msgpackr-extract-linux-x64": "3.0.4", + "@msgpackr-extract/msgpackr-extract-win32-x64": "3.0.4" + } + }, "node_modules/multer": { "version": "1.4.5-lts.2", "resolved": "https://registry.npmjs.org/multer/-/multer-1.4.5-lts.2.tgz", @@ -2586,6 +3009,12 @@ "node": ">= 0.6" } }, + "node_modules/node-abort-controller": { + "version": "3.1.1", + "resolved": "https://registry.npmjs.org/node-abort-controller/-/node-abort-controller-3.1.1.tgz", + "integrity": "sha512-AGK2yQKIjRuqnc6VkX2Xj5d+QW8xZ87pa1UK6yA6ouUyuxfHuMP6umE5QK7UmTeOAymo+Zx1Fxiuw9rVx8taHQ==", + "license": "MIT" + }, "node_modules/node-addon-api": { "version": "8.9.0", "resolved": "https://registry.npmjs.org/node-addon-api/-/node-addon-api-8.9.0.tgz", @@ -2606,6 +3035,21 @@ "node-gyp-build-test": "build-test.js" } }, + "node_modules/node-gyp-build-optional-packages": { + "version": "5.2.2", + "resolved": "https://registry.npmjs.org/node-gyp-build-optional-packages/-/node-gyp-build-optional-packages-5.2.2.tgz", + "integrity": "sha512-s+w+rBWnpTMwSFbaE0UXsRlg7hU4FjekKU4eyAih5T8nJuNZT1nNsskXpxmeqSK9UzkBl6UgRlnKc8hz8IEqOw==", + "license": "MIT", + "optional": true, + "dependencies": { + "detect-libc": "^2.0.1" + }, + "bin": { + "node-gyp-build-optional-packages": "bin.js", + "node-gyp-build-optional-packages-optional": "optional.js", + "node-gyp-build-optional-packages-test": "build-test.js" + } + }, "node_modules/nodemailer": { "version": "9.0.3", "resolved": "https://registry.npmjs.org/nodemailer/-/nodemailer-9.0.3.tgz", @@ -3002,6 +3446,15 @@ "node": ">= 0.10" } }, + "node_modules/proxy-from-env": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/proxy-from-env/-/proxy-from-env-2.1.0.tgz", + "integrity": "sha512-cJ+oHTW1VAEa8cJslgmUZrc+sjRKgAKl3Zyse6+PV38hZe/V6Z14TbCuXcan9F9ghlz4QrFr2c92TNF82UkYHA==", + "license": "MIT", + "engines": { + "node": ">=10" + } + }, "node_modules/pstree.remy": { "version": "1.1.8", "resolved": "https://registry.npmjs.org/pstree.remy/-/pstree.remy-1.1.8.tgz", @@ -3083,6 +3536,15 @@ "node": ">=8.10.0" } }, + "node_modules/redis-errors": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/redis-errors/-/redis-errors-1.2.0.tgz", + "integrity": "sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==", + "license": "MIT", + "engines": { + "node": ">=4" + } + }, "node_modules/require-directory": { "version": "2.1.1", "resolved": "https://registry.npmjs.org/require-directory/-/require-directory-2.1.1.tgz", @@ -3591,6 +4053,12 @@ "node": "*" } }, + "node_modules/standard-as-callback": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/standard-as-callback/-/standard-as-callback-2.1.0.tgz", + "integrity": "sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==", + "license": "MIT" + }, "node_modules/statuses": { "version": "2.0.2", "resolved": "https://registry.npmjs.org/statuses/-/statuses-2.0.2.tgz", diff --git a/package.json b/package.json index 469efd0..efe8c8f 100644 --- a/package.json +++ b/package.json @@ -22,7 +22,9 @@ }, "dependencies": { "@aws-sdk/client-s3": "^3.1113.0", + "axios": "^1.20.0", "bcrypt": "^6.0.0", + "bullmq": "^6.3.2", "cloudinary": "^2.10.0", "compression": "^1.7.5", "cookie-parser": "^1.4.7", @@ -31,6 +33,7 @@ "express": "^4.21.2", "express-validator": "^7.1.0", "helmet": "^8.0.0", + "ioredis": "^6.0.0", "jsonwebtoken": "^9.0.2", "morgan": "^1.10.0", "multer": "^1.4.5-lts.1", diff --git a/src/features/integrations/adapters/adapter.registry.js b/src/features/integrations/adapters/adapter.registry.js new file mode 100644 index 0000000..14275b4 --- /dev/null +++ b/src/features/integrations/adapters/adapter.registry.js @@ -0,0 +1,12 @@ +import { ShopifyAdapter } from './shopify/shopify.adapter.js'; + +export const adapterRegistry = { + getAdapter(channel, credentials) { + switch (channel.toLowerCase()) { + case 'shopify': + return new ShopifyAdapter(credentials); + default: + throw new Error(`Unsupported channel adapter type: ${channel}`); + } + } +}; diff --git a/src/features/integrations/adapters/contracts/channel-adapter.contract.js b/src/features/integrations/adapters/contracts/channel-adapter.contract.js new file mode 100644 index 0000000..650de47 --- /dev/null +++ b/src/features/integrations/adapters/contracts/channel-adapter.contract.js @@ -0,0 +1,21 @@ +export class BaseChannelAdapter { + constructor(config = {}) { + this.config = config; + } + + async testConnection() { + throw new Error('Method testConnection() must be implemented'); + } +} + +export class ProductPublisher { + async publishProduct(context) { + throw new Error('Method publishProduct() must be implemented'); + } +} + +export class ProductDeleter { + async deleteProduct(context) { + throw new Error('Method deleteProduct() must be implemented'); + } +} diff --git a/src/features/integrations/adapters/shopify/shopify.adapter.js b/src/features/integrations/adapters/shopify/shopify.adapter.js new file mode 100644 index 0000000..b21dd85 --- /dev/null +++ b/src/features/integrations/adapters/shopify/shopify.adapter.js @@ -0,0 +1,70 @@ +import { BaseChannelAdapter } from '../contracts/channel-adapter.contract.js'; +import { ShopifyClient } from './shopify.client.js'; +import { ShopifyPublisher } from './shopify.publisher.js'; + +const SHOP_QUERY = ` + query { + shop { + name + email + myshopifyDomain + plan { + displayName + } + } + } +`; + +export class ShopifyAdapter extends BaseChannelAdapter { + constructor(credentials = {}) { + super(credentials); + // Support both credential naming conventions + const { + shop_domain, access_token, + api_key, api_secret_key, + shopDomain, accessToken, apiKey, apiSecretKey + } = credentials; + + this.shopDomain = shop_domain || shopDomain; + this.accessToken = access_token || accessToken; + this.apiKey = api_key || apiKey; + this.apiSecretKey = api_secret_key || apiSecretKey; + + this.client = new ShopifyClient( + this.shopDomain, + this.accessToken, + this.apiKey, + this.apiSecretKey + ); + this.publisher = new ShopifyPublisher(this.client); + } + + async testConnection() { + if (!this.shopDomain) { + throw new Error('Shopify connection requires shop_domain credential'); + } + if (!this.accessToken && !this.apiSecretKey) { + throw new Error('Shopify connection requires either access_token or api_secret_key credential'); + } + const res = await this.client.graphql(SHOP_QUERY); + const shop = res.data?.shop; + if (!shop) { + throw new Error('Invalid Shopify GraphQL shop query response — check your credentials'); + } + return { + connected: true, + shopName: shop.name, + shopDomain: shop.myshopifyDomain, + email: shop.email, + plan: shop.plan?.displayName + }; + } + + async publishProduct(context) { + return await this.publisher.publishProduct(context); + } + + async deleteProduct(context) { + return await this.publisher.deleteProduct(context); + } +} diff --git a/src/features/integrations/adapters/shopify/shopify.client.js b/src/features/integrations/adapters/shopify/shopify.client.js new file mode 100644 index 0000000..71a95f9 --- /dev/null +++ b/src/features/integrations/adapters/shopify/shopify.client.js @@ -0,0 +1,65 @@ +import { createHttpClient } from '../../../../shared/infrastructure/http/http.client.js'; + +/** + * ShopifyClient for Admin API (GraphQL v2025-01) + * Uses X-Shopify-Access-Token header for all authenticated requests (OAuth & Custom App tokens) + */ +export class ShopifyClient { + constructor(shopDomain, accessToken, apiKey = null, apiSecretKey = null) { + this.shopDomain = shopDomain + ? shopDomain.replace(/^https?:\/\//, '').replace(/\/$/, '').trim() + : ''; + + // Effective Admin API token is accessToken (or fallback to apiSecretKey if provided) + this.accessToken = accessToken || apiSecretKey; + this.apiKey = apiKey; + this.apiSecretKey = apiSecretKey; + + this.graphqlEndpoint = `/admin/api/2025-01/graphql.json`; + + const headers = { + 'Content-Type': 'application/json' + }; + + if (this.accessToken) { + headers['X-Shopify-Access-Token'] = this.accessToken; + } + + this.httpClient = createHttpClient({ + baseURL: `https://${this.shopDomain}`, + headers + }); + } + + async graphql(query, variables = {}) { + const res = await this.httpClient.post(this.graphqlEndpoint, { + query, + variables + }); + + const responseData = res.data || res; + + // Inspect Shopify cost extensions for rate limiting + if (responseData?.extensions?.cost) { + const { currentlyAvailable, requestedQueryCost } = responseData.extensions.cost; + if (currentlyAvailable && currentlyAvailable < requestedQueryCost) { + const waitMs = Math.ceil((requestedQueryCost - currentlyAvailable) / 50) * 1000; + await new Promise(r => setTimeout(r, Math.min(waitMs, 10000))); + } + } + + if (responseData?.errors && responseData.errors.length > 0) { + throw new Error(`Shopify GraphQL Error: ${responseData.errors.map(e => e.message).join(', ')}`); + } + + return responseData; + } + + /** + * REST API helper for endpoints not available in GraphQL + */ + async restGet(path) { + const res = await this.httpClient.get(path); + return res.data || res; + } +} diff --git a/src/features/integrations/adapters/shopify/shopify.mapper.js b/src/features/integrations/adapters/shopify/shopify.mapper.js new file mode 100644 index 0000000..ba63d91 --- /dev/null +++ b/src/features/integrations/adapters/shopify/shopify.mapper.js @@ -0,0 +1,59 @@ +export const shopifyMapper = { + toProductInput(canonicalProduct, existingGid = null) { + const { content, taxonomy, identity, variants = [] } = canonicalProduct; + + const input = { + title: content.name, + descriptionHtml: content.description || '', + status: content.status === 'active' ? 'ACTIVE' : 'DRAFT', + vendor: taxonomy.brand?.name || 'PIM Catalog', + productType: taxonomy.category?.name || 'General' + }; + + // Generate 100% unique handle using product title + unique UUID snippet to prevent handle collisions + if (!existingGid && identity.id) { + const baseSlug = (content.name || 'product') + .toLowerCase() + .replace(/[^a-z0-9]+/g, '-') + .replace(/(^-|-$)/g, ''); + const uniqueSuffix = String(identity.id).replace(/[^a-z0-9]/gi, '').slice(0, 10); + input.handle = `${baseSlug}-${uniqueSuffix}`; + } + + // productOptions can only be passed on CREATE (when existingGid is null) and capped at 3 options (Shopify limit) + if (!existingGid && Array.isArray(variants) && variants.length > 0) { + const optionKeys = new Set(); + variants.forEach(v => { + if (v.attributes) { + Object.keys(v.attributes).forEach(k => optionKeys.add(k)); + } + }); + + if (optionKeys.size > 0) { + input.productOptions = Array.from(optionKeys).slice(0, 3).map(optName => ({ + name: optName, + values: Array.from(new Set( + variants + .map(v => v.attributes?.[optName]) + .filter(Boolean) + )).map(val => ({ name: String(val) })) + })); + } + } + + if (existingGid) { + input.id = existingGid; + } + + return input; + }, + + toVariantsBulkInput(shopifyProductId, variants = []) { + return variants.map(v => ({ + price: String(v.price || 0), + compareAtPrice: v.costPrice ? String(v.costPrice) : null, + sku: v.sku || '', + barcode: v.sku || '' + })); + } +}; diff --git a/src/features/integrations/adapters/shopify/shopify.oauth.service.js b/src/features/integrations/adapters/shopify/shopify.oauth.service.js new file mode 100644 index 0000000..66a10b4 --- /dev/null +++ b/src/features/integrations/adapters/shopify/shopify.oauth.service.js @@ -0,0 +1,275 @@ +import crypto from 'crypto'; +import axios from 'axios'; +import { models } from '../../../../shared/database/models.js'; +import { secretService } from '../../../../shared/infrastructure/secrets/secret.service.js'; + +/** + * Shopify OAuth 2.0 service — Production-grade implementation + * + * Security measures: + * - State nonce stored in DB (survives server restarts) + * - HMAC-SHA256 signature validated before code exchange + * - Shop domain validated against stored expected value + * - Code exchanged server-side only — never exposed to frontend + * - Access token stored AES-256-GCM encrypted + * - All debug logs redact secrets + */ + +const REQUIRED_SCOPES = [ + 'read_product_feeds', + 'write_product_feeds', + 'read_product_listings', + 'write_product_listings', + 'read_products', + 'write_products' +].join(','); + +const NONCE_TTL_MS = 10 * 60 * 1000; // 10 minutes + +// ─── HMAC Validation ────────────────────────────────────────────────────────── + +/** + * Validate Shopify's HMAC signature on the callback. + * Per Shopify docs: remove hmac from params, sort remaining, + * build query string, HMAC-SHA256 with client_secret, compare. + */ +function validateHmac(queryParams, clientSecret) { + const { hmac, ...rest } = queryParams; + if (!hmac) return false; + + // Build sorted query string (keys sorted lexicographically) + const message = Object.keys(rest) + .sort() + .map(k => `${k}=${rest[k]}`) + .join('&'); + + const digest = crypto + .createHmac('sha256', clientSecret) + .update(message) + .digest('hex'); + + // Constant-time comparison + try { + return crypto.timingSafeEqual( + Buffer.from(digest, 'hex'), + Buffer.from(hmac, 'hex') + ); + } catch { + return false; + } +} + +// ─── Public API ─────────────────────────────────────────────────────────────── + +export const shopifyOAuthService = { + /** + * Build the Shopify authorization URL. + * Persists state nonce as an encrypted credential in DB (credential_type = 'oauth_nonce'). + */ + async buildAuthorizationUrl(integrationId, rawShopDomain, rawApiKey, tenantId) { + let shopDomain = (rawShopDomain || '').trim().replace(/^https?:\/\//, '').replace(/\/$/, ''); + let apiKey = (rawApiKey || '').trim(); + + // Auto-fix if domain and API Key were swapped + if (apiKey.includes('.myshopify.com') && !shopDomain.includes('.myshopify.com')) { + const temp = shopDomain; + shopDomain = apiKey; + apiKey = temp; + } + + if (shopDomain && !shopDomain.includes('.')) { + shopDomain = `${shopDomain}.myshopify.com`; + } + + const nonce = crypto.randomBytes(24).toString('hex'); + const state = `${integrationId}::${nonce}`; + + // Persist the state in DB so it survives restarts + // Remove any old nonce for this integration first + await models.IntegrationCredential.destroy({ + where: { integration_id: integrationId, credential_type: 'oauth_nonce' } + }); + + await models.IntegrationCredential.create({ + tenant_id: tenantId, + integration_id: integrationId, + credential_type: 'oauth_nonce', + encrypted_secret: secretService.encrypt(JSON.stringify({ + nonce, + shopDomain, + createdAt: Date.now() + })), + key_version: 1, + expires_at: new Date(Date.now() + NONCE_TTL_MS) + }); + + const backendPort = process.env.PORT || 5002; + const backendUrl = process.env.BACKEND_URL || `http://localhost:${backendPort}`; + const callbackUrl = `${backendUrl}/api/v1/integrations/shopify/oauth/callback`; + + const url = [ + `https://${shopDomain}/admin/oauth/authorize`, + `?client_id=${apiKey}`, + `&scope=${encodeURIComponent(REQUIRED_SCOPES)}`, + `&redirect_uri=${encodeURIComponent(callbackUrl)}`, + `&state=${encodeURIComponent(state)}` + ].join(''); + + console.log(`[ShopifyOAuth] Authorization URL built for integration ${integrationId}, shop: ${shopDomain}`); + return { authorizationUrl: url, state }; + }, + + /** + * Handle the Shopify OAuth callback. + * 1. Parse & validate state → look up nonce in DB + * 2. Validate HMAC + * 3. Validate shop domain + * 4. Exchange code for access token + * 5. Store token encrypted, mark integration active + * 6. Return integrationId for redirect + */ + async handleCallback(query) { + const { code, shop, state, hmac, timestamp } = query; + + // ── Basic presence check ────────────────────────────────────────────────── + if (!code || !shop || !state) { + throw new Error('Missing required OAuth callback parameters (code, shop, state)'); + } + + // ── Parse state — format is integrationId::nonce ────────────────────────── + const stateParts = decodeURIComponent(state).split('::'); + if (stateParts.length !== 2) { + throw new Error('Malformed OAuth state parameter'); + } + const [integrationId, nonce] = stateParts; + + // ── Load stored nonce from DB ───────────────────────────────────────────── + const nonceRecord = await models.IntegrationCredential.findOne({ + where: { integration_id: integrationId, credential_type: 'oauth_nonce' } + }); + + if (!nonceRecord) { + throw new Error( + 'OAuth state not found or expired. The server may have restarted. Please restart the OAuth flow from the PIM Integration Hub.' + ); + } + + let storedData; + try { + storedData = JSON.parse(secretService.decrypt(nonceRecord.encrypted_secret)); + } catch { + throw new Error('Failed to decrypt stored OAuth state'); + } + + // ── Validate nonce ──────────────────────────────────────────────────────── + if (storedData.nonce !== nonce) { + throw new Error('OAuth state nonce mismatch — possible CSRF attempt'); + } + + // ── Check nonce TTL ─────────────────────────────────────────────────────── + if (Date.now() - storedData.createdAt > NONCE_TTL_MS) { + await nonceRecord.destroy(); + throw new Error('OAuth state expired. Please restart the authorization flow.'); + } + + // ── Validate shop domain ────────────────────────────────────────────────── + const expectedShop = storedData.shopDomain; + const normalizedShop = shop.replace(/^https?:\/\//, '').replace(/\/$/, ''); + if (normalizedShop !== expectedShop) { + throw new Error(`Shop domain mismatch: expected "${expectedShop}", got "${normalizedShop}"`); + } + + // ── Load integration + credentials ──────────────────────────────────────── + const integration = await models.Integration.findByPk(integrationId); + if (!integration) throw new Error(`Integration ${integrationId} not found`); + + const credRecords = await models.IntegrationCredential.findAll({ + where: { integration_id: integrationId } + }); + const credMap = {}; + credRecords.forEach(c => { + credMap[c.credential_type] = secretService.decrypt(c.encrypted_secret); + }); + + const apiKey = credMap.api_key; + const apiSecretKey = credMap.api_secret_key; + + if (!apiKey || !apiSecretKey) { + throw new Error('api_key and api_secret_key credentials must be saved before OAuth can complete'); + } + + // ── Validate Shopify HMAC ───────────────────────────────────────────────── + if (hmac) { + const hmacValid = validateHmac(query, apiSecretKey); + if (!hmacValid) { + throw new Error('Shopify HMAC validation failed — callback may be forged'); + } + console.log('[ShopifyOAuth] HMAC validated successfully'); + } else { + console.warn('[ShopifyOAuth] No HMAC in callback — skipping HMAC validation (dev mode)'); + } + + // ── Exchange code for permanent access token ─────────────────────────────── + console.log(`[ShopifyOAuth] Exchanging code for access token with shop: ${normalizedShop}`); + let tokenData; + try { + const tokenResponse = await axios.post( + `https://${normalizedShop}/admin/oauth/access_token`, + { + client_id: apiKey, + client_secret: apiSecretKey, + code + }, + { + headers: { 'Content-Type': 'application/json', 'Accept': 'application/json' }, + timeout: 15000 + } + ); + tokenData = tokenResponse.data; + } catch (err) { + const detail = err.response?.data ? JSON.stringify(err.response.data) : err.message; + throw new Error(`Shopify token exchange failed: ${detail}`); + } + + const { access_token, scope } = tokenData; + if (!access_token) { + throw new Error('Shopify did not return an access_token in the token exchange response'); + } + + console.log(`[ShopifyOAuth] Access token received. Scopes: ${scope}`); + + // ── Store the permanent access token (encrypted) ────────────────────────── + const tenantId = integration.tenant_id; + + await models.IntegrationCredential.destroy({ + where: { integration_id: integrationId, credential_type: 'access_token' } + }); + await models.IntegrationCredential.create({ + tenant_id: tenantId, + integration_id: integrationId, + credential_type: 'access_token', + encrypted_secret: secretService.encrypt(access_token), + key_version: 1, + expires_at: null + }); + + // ── Remove the one-time nonce ───────────────────────────────────────────── + await nonceRecord.destroy(); + + // ── Mark integration as active ──────────────────────────────────────────── + await integration.update({ + status: 'active', + health_status: 'healthy', + last_synced_at: null + }); + + console.log(`[ShopifyOAuth] Integration ${integrationId} connected to shop: ${normalizedShop}`); + + return { + integrationId, + shop: normalizedShop, + scope, + connected: true + }; + } +}; diff --git a/src/features/integrations/adapters/shopify/shopify.publisher.js b/src/features/integrations/adapters/shopify/shopify.publisher.js new file mode 100644 index 0000000..c12a8b7 --- /dev/null +++ b/src/features/integrations/adapters/shopify/shopify.publisher.js @@ -0,0 +1,188 @@ +import { models } from '../../../../shared/database/models.js'; +import { shopifyMapper } from './shopify.mapper.js'; + +const PRODUCT_CREATE_MUTATION = ` + mutation productCreate($input: ProductInput!) { + productCreate(input: $input) { + product { + id + title + handle + onlineStoreUrl + } + userErrors { + field + message + } + } + } +`; + +const PRODUCT_UPDATE_MUTATION = ` + mutation productUpdate($input: ProductInput!) { + productUpdate(input: $input) { + product { + id + title + handle + onlineStoreUrl + } + userErrors { + field + message + } + } + } +`; + +const PRODUCT_DELETE_MUTATION = ` + mutation productDelete($input: ProductDeleteInput!) { + productDelete(input: $input) { + deletedProductId + userErrors { + field + message + } + } + } +`; + +const PRODUCT_CREATE_MEDIA_MUTATION = ` + mutation productCreateMedia($media: [CreateMediaInput!]!, $productId: ID!) { + productCreateMedia(media: $media, productId: $productId) { + media { + id + status + } + userErrors { + field + message + } + } + } +`; + +export class ShopifyPublisher { + constructor(shopifyClient) { + this.client = shopifyClient; + } + + async publishProduct(context) { + const { tenantId, integrationId, canonicalProduct } = context; + const productId = canonicalProduct.identity.id; + + // Check existing mapping + const existingResource = await models.ExternalResource.findOne({ + where: { + tenant_id: tenantId, + integration_id: integrationId, + source_id: productId, + resource_type: 'PRODUCT' + } + }); + + const isUpdate = !!existingResource; + const input = shopifyMapper.toProductInput(canonicalProduct, existingResource?.external_id); + + let response; + let shopifyProduct; + + if (isUpdate) { + response = await this.client.graphql(PRODUCT_UPDATE_MUTATION, { input }); + const result = response.data?.productUpdate; + if (result?.userErrors?.length > 0) { + throw new Error(`Shopify productUpdate user errors: ${result.userErrors.map(e => e.message).join(', ')}`); + } + shopifyProduct = result.product; + } else { + response = await this.client.graphql(PRODUCT_CREATE_MUTATION, { input }); + const result = response.data?.productCreate; + if (result?.userErrors?.length > 0) { + throw new Error(`Shopify productCreate user errors: ${result.userErrors.map(e => e.message).join(', ')}`); + } + shopifyProduct = result.product; + } + + if (!shopifyProduct || !shopifyProduct.id) { + throw new Error('Shopify GraphQL returned empty product response'); + } + + // Attach media/image URLs to Shopify product + const validMedia = (canonicalProduct.media || []) + .filter(m => m.url && (m.url.startsWith('http://') || m.url.startsWith('https://'))) + .map(m => ({ + mediaContentType: 'IMAGE', + originalSource: m.url, + alt: canonicalProduct.content.name + })); + + if (validMedia.length > 0 && shopifyProduct?.id) { + try { + await this.client.graphql(PRODUCT_CREATE_MEDIA_MUTATION, { + productId: shopifyProduct.id, + media: validMedia + }); + } catch (mediaErr) { + console.warn(`[ShopifyPublisher] Media attach warning for product ${shopifyProduct.id}:`, mediaErr.message); + } + } + + // Save external_resources mapping + if (existingResource) { + await existingResource.update({ + external_url: shopifyProduct.onlineStoreUrl || existingResource.external_url, + last_synced_at: new Date(), + last_source_version: canonicalProduct.version || 1 + }); + } else { + await models.ExternalResource.create({ + tenant_id: tenantId, + integration_id: integrationId, + resource_type: 'PRODUCT', + source_type: 'PIM', + source_id: productId, + external_id: shopifyProduct.id, + external_url: shopifyProduct.onlineStoreUrl || `https://${this.client.shopDomain}/admin/products/${shopifyProduct.id.split('/').pop()}`, + status: 'active', + last_synced_at: new Date(), + last_source_version: canonicalProduct.version || 1 + }); + } + + return { + success: true, + externalId: shopifyProduct.id, + externalUrl: shopifyProduct.onlineStoreUrl, + operation: isUpdate ? 'UPDATE' : 'CREATE' + }; + } + + async deleteProduct(context) { + const { tenantId, integrationId, productId } = context; + + const existingResource = await models.ExternalResource.findOne({ + where: { + tenant_id: tenantId, + integration_id: integrationId, + source_id: productId, + resource_type: 'PRODUCT' + } + }); + + if (!existingResource) { + return { success: true, message: 'Resource was not published to Shopify' }; + } + + const response = await this.client.graphql(PRODUCT_DELETE_MUTATION, { + input: { id: existingResource.external_id } + }); + + const result = response.data?.productDelete; + if (result?.userErrors?.length > 0) { + throw new Error(`Shopify productDelete user errors: ${result.userErrors.map(e => e.message).join(', ')}`); + } + + await existingResource.destroy(); + return { success: true, deletedProductId: result.deletedProductId }; + } +} diff --git a/src/features/integrations/controllers/integration.controller.js b/src/features/integrations/controllers/integration.controller.js new file mode 100644 index 0000000..765d49f --- /dev/null +++ b/src/features/integrations/controllers/integration.controller.js @@ -0,0 +1,281 @@ +import { integrationService } from '../domain/integration.service.js'; +import { syncService } from '../domain/sync.service.js'; +import { queueService } from '../../../shared/infrastructure/queue/queue.service.js'; +import { QUEUE_NAMES } from '../../../shared/infrastructure/queue/queue.constants.js'; +import { models } from '../../../shared/database/models.js'; +import { shopifyOAuthService } from '../adapters/shopify/shopify.oauth.service.js'; + +const getTenantId = (req) => { + return req.context?.tenant_id || req.context?.tenantId || req.user?.tenant_id || req.user?.tenantId || 19; +}; + +export const integrationController = { + async listIntegrations(req, res, next) { + try { + const tenantId = getTenantId(req); + const data = await integrationService.listIntegrations(tenantId); + res.json({ success: true, data }); + } catch (err) { + next(err); + } + }, + + async getIntegration(req, res, next) { + try { + const tenantId = getTenantId(req); + const data = await integrationService.getById(req.params.id, tenantId); + res.json({ success: true, data }); + } catch (err) { + next(err); + } + }, + + async createIntegration(req, res, next) { + try { + const tenantId = getTenantId(req); + const data = await integrationService.createIntegration(tenantId, req.body); + res.status(201).json({ success: true, data }); + } catch (err) { + next(err); + } + }, + + async updateIntegration(req, res, next) { + try { + const tenantId = getTenantId(req); + const data = await integrationService.updateIntegration(req.params.id, tenantId, req.body); + res.json({ success: true, data }); + } catch (err) { + next(err); + } + }, + + async deleteIntegration(req, res, next) { + try { + const tenantId = getTenantId(req); + await integrationService.deleteIntegration(req.params.id, tenantId); + res.json({ success: true, message: 'Integration deleted successfully' }); + } catch (err) { + next(err); + } + }, + + async setCredentials(req, res, next) { + try { + const tenantId = getTenantId(req); + const { credential_type, secret_value, expires_at } = req.body; + if (!credential_type || !secret_value) { + return res.status(400).json({ success: false, message: 'credential_type and secret_value are required' }); + } + const data = await integrationService.setCredentials( + req.params.id, + tenantId, + credential_type, + secret_value, + expires_at + ); + res.json({ success: true, message: 'Credentials updated securely', data: { id: data.id, type: data.credential_type } }); + } catch (err) { + next(err); + } + }, + + async getCredentials(req, res, next) { + try { + const tenantId = getTenantId(req); + const data = await integrationService.getDecryptedCredentials(req.params.id, tenantId); + res.json({ success: true, data }); + } catch (err) { + next(err); + } + }, + + async testConnection(req, res, next) { + try { + const tenantId = getTenantId(req); + const result = await integrationService.testConnection(req.params.id, tenantId); + res.json({ success: true, data: result }); + } catch (err) { + res.status(400).json({ success: false, message: err.message }); + } + }, + + async triggerSync(req, res, next) { + try { + const tenantId = getTenantId(req); + const integrationId = req.params.id; + const { productId, productIds } = req.body; + + const integration = await integrationService.getById(integrationId, tenantId); + const syncJob = await syncService.createSyncJob(tenantId, integrationId, 'manual'); + + let targets = []; + if (productId) { + targets.push(productId); + } else if (Array.isArray(productIds)) { + targets = productIds; + } else { + // Fetch all products for tenant if no specific ID provided + const allProds = await models.Product.findAll({ + where: { tenant_id: tenantId }, + attributes: ['id'] + }); + targets = allProds.map(p => p.id); + } + + await syncJob.update({ total_items: targets.length }); + + for (const pId of targets) { + await queueService.addJob(QUEUE_NAMES.INTEGRATION_SYNC, `sync_product_${pId}`, { + tenantId, + integrationId, + syncJobId: syncJob.id, + productId: pId, + operation: 'UPDATE' + }); + } + + res.status(202).json({ + success: true, + message: `Sync job initialized for ${targets.length} products`, + data: { syncJobId: syncJob.id, totalItems: targets.length } + }); + } catch (err) { + next(err); + } + }, + + async listAllSyncJobs(req, res, next) { + try { + const tenantId = getTenantId(req); + const jobs = await models.IntegrationSyncJob.findAll({ + where: { tenant_id: tenantId }, + include: [ + { + model: models.Integration, + as: 'integration', + attributes: ['id', 'name', 'channel'] + } + ], + order: [['created_at', 'DESC']], + limit: 100 + }); + res.json({ success: true, data: jobs }); + } catch (err) { + next(err); + } + }, + + async listSyncJobs(req, res, next) { + try { + const tenantId = getTenantId(req); + const integrationId = req.params.id; + const jobs = await models.IntegrationSyncJob.findAll({ + where: { tenant_id: tenantId, integration_id: integrationId }, + order: [['created_at', 'DESC']], + limit: 50 + }); + res.json({ success: true, data: jobs }); + } catch (err) { + next(err); + } + }, + + async listSyncItems(req, res, next) { + try { + const tenantId = getTenantId(req); + const { jobId } = req.params; + + // Validate UUID format before querying Postgres + const isUuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(jobId); + if (!isUuid) { + return res.json({ success: true, data: [] }); + } + + const items = await models.SyncItem.findAll({ + where: { tenant_id: tenantId, sync_job_id: jobId }, + order: [['created_at', 'DESC']], + raw: true + }); + + // Safely enrich with Product details in-memory to prevent SQL type mismatch JOIN errors + const productIds = Array.from(new Set(items.map(i => i.product_id).filter(Boolean))); + const productMap = {}; + if (productIds.length > 0) { + try { + const prods = await models.Product.findAll({ + where: { id: productIds }, + attributes: ['id', 'name', 'sku'], + raw: true + }); + prods.forEach(p => { productMap[p.id] = p; }); + } catch (pErr) { + console.warn('[listSyncItems] Failed to fetch product metadata:', pErr.message); + } + } + + const enriched = items.map(item => ({ + ...item, + product: productMap[item.product_id] || { id: item.product_id, name: `Product #${item.product_id.slice(0, 8)}`, sku: item.sku || '' } + })); + + res.json({ success: true, data: enriched }); + } catch (err) { + console.error('[listSyncItems] Error:', err); + next(err); + } + }, + + /** + * Start Shopify OAuth flow: + * Requires credentials: shop_domain, api_key already saved on this integration. + * Returns { authorizationUrl } which the frontend opens in a new tab. + */ + async startShopifyOAuth(req, res, next) { + try { + const tenantId = getTenantId(req); + const integrationId = req.params.id; + + // Load decrypted credentials to get shop_domain + api_key + const creds = await integrationService.getDecryptedCredentials(integrationId, tenantId); + + if (!creds.shop_domain) { + return res.status(400).json({ success: false, message: 'Save shop_domain credential first before starting OAuth' }); + } + if (!creds.api_key) { + return res.status(400).json({ success: false, message: 'Save api_key credential first before starting OAuth' }); + } + + const result = await shopifyOAuthService.buildAuthorizationUrl( + integrationId, + creds.shop_domain, + creds.api_key, + tenantId + ); + + console.log(`[ShopifyOAuth] Generated Authorization URL:\n Domain: ${creds.shop_domain}\n API Key: ${creds.api_key}\n URL: ${result.authorizationUrl}`); + + res.json({ success: true, data: result }); + } catch (err) { + next(err); + } + }, + + /** + * Shopify OAuth callback — public route (no auth middleware needed). + * Shopify redirects here with ?code=...&shop=...&state=... + * After exchange, redirects user to frontend /integrations?oauth=success&integrationId=... + */ + async handleShopifyOAuthCallback(req, res) { + const frontendUrl = process.env.FRONTEND_URL || 'http://localhost:5173'; + + try { + const result = await shopifyOAuthService.handleCallback(req.query); + res.redirect(`${frontendUrl}/integrations?oauth=success&integrationId=${result.integrationId}&shop=${result.shop}`); + } catch (err) { + console.error('[ShopifyOAuth] Callback error:', err.message); + res.redirect(`${frontendUrl}/integrations?oauth=error&message=${encodeURIComponent(err.message)}`); + } + } +}; + diff --git a/src/features/integrations/domain/integration.service.js b/src/features/integrations/domain/integration.service.js new file mode 100644 index 0000000..f1d9221 --- /dev/null +++ b/src/features/integrations/domain/integration.service.js @@ -0,0 +1,116 @@ +import { models } from '../../../shared/database/models.js'; +import { secretService } from '../../../shared/infrastructure/secrets/secret.service.js'; +import { adapterRegistry } from '../adapters/adapter.registry.js'; + +export const integrationService = { + async listIntegrations(tenantId) { + return await models.Integration.findAll({ + where: { tenant_id: tenantId }, + include: [ + { + model: models.IntegrationCredential, + as: 'credentials', + attributes: ['id', 'credential_type', 'expires_at', 'created_at'] + } + ], + order: [['created_at', 'DESC']] + }); + }, + + async getById(id, tenantId) { + const integration = await models.Integration.findOne({ + where: { id, tenant_id: tenantId }, + include: [ + { + model: models.IntegrationCredential, + as: 'credentials', + attributes: ['id', 'credential_type', 'expires_at', 'created_at'] + } + ] + }); + + if (!integration) { + throw new Error(`Integration ${id} not found`); + } + return integration; + }, + + async createIntegration(tenantId, body) { + return await models.Integration.create({ + tenant_id: tenantId, + name: body.name, + channel: body.channel || body.channelName || 'shopify', + integration_type: body.integration_type || body.integrationType || 'ecommerce', + status: body.status || 'active', + sync_mode: body.sync_mode || body.syncMode || 'auto', + sync_frequency: body.sync_frequency || body.syncFrequency || 'realtime', + health_status: 'healthy' + }); + }, + + async updateIntegration(id, tenantId, updates) { + const integration = await this.getById(id, tenantId); + return await integration.update(updates); + }, + + async setCredentials(id, tenantId, credentialType, secretValue, expiresAt = null) { + const integration = await this.getById(id, tenantId); + const encryptedSecret = secretService.encrypt(secretValue); + + // Remove old credential of same type + await models.IntegrationCredential.destroy({ + where: { integration_id: id, credential_type: credentialType, tenant_id: tenantId } + }); + + return await models.IntegrationCredential.create({ + tenant_id: tenantId, + integration_id: id, + credential_type: credentialType, + encrypted_secret: encryptedSecret, + key_version: 1, + expires_at: expiresAt + }); + }, + + async getDecryptedCredentials(id, tenantId) { + const credentials = await models.IntegrationCredential.findAll({ + where: { integration_id: id, tenant_id: tenantId } + }); + + const map = {}; + credentials.forEach(c => { + map[c.credential_type] = secretService.decrypt(c.encrypted_secret); + }); + return map; + }, + + async testConnection(id, tenantId) { + const integration = await this.getById(id, tenantId); + const decryptedConfig = await this.getDecryptedCredentials(id, tenantId); + + const adapter = adapterRegistry.getAdapter(integration.channel, decryptedConfig); + return await adapter.testConnection(); + }, + + async deleteIntegration(id, tenantId) { + const integration = await this.getById(id, tenantId); + + // Clean up credentials and associated records + await models.IntegrationCredential.destroy({ + where: { integration_id: id, tenant_id: tenantId } + }); + if (models.ExternalResource) { + await models.ExternalResource.destroy({ + where: { integration_id: id, tenant_id: tenantId } + }); + } + if (models.PublishingRule) { + await models.PublishingRule.destroy({ + where: { integration_id: id, tenant_id: tenantId } + }); + } + + await integration.destroy(); + return true; + } +}; diff --git a/src/features/integrations/domain/rule.service.js b/src/features/integrations/domain/rule.service.js new file mode 100644 index 0000000..de033ae --- /dev/null +++ b/src/features/integrations/domain/rule.service.js @@ -0,0 +1,37 @@ +import { models } from '../../../shared/database/models.js'; + +export const ruleService = { + async evaluateRules(integrationId, tenantId, canonicalProduct) { + const rules = await models.PublishingRule.findAll({ + where: { integration_id: integrationId, tenant_id: tenantId } + }); + + if (rules.length === 0) { + return { eligible: true, reasons: [] }; + } + + const reasons = []; + + for (const rule of rules) { + const cond = rule.conditions_json || {}; + + // Category check + if (cond.allowedCategories && Array.isArray(cond.allowedCategories) && cond.allowedCategories.length > 0) { + const prodCatId = canonicalProduct.taxonomy?.category?.id; + if (!prodCatId || !cond.allowedCategories.includes(prodCatId)) { + reasons.push(`Product category not in allowed rule list for rule ${rule.name}`); + } + } + + // Status check + if (cond.requiredStatus && canonicalProduct.content?.status !== cond.requiredStatus) { + reasons.push(`Product status '${canonicalProduct.content?.status}' does not match required status '${cond.requiredStatus}'`); + } + } + + return { + eligible: reasons.length === 0, + reasons + }; + } +}; diff --git a/src/features/integrations/domain/sync.service.js b/src/features/integrations/domain/sync.service.js new file mode 100644 index 0000000..3f1573b --- /dev/null +++ b/src/features/integrations/domain/sync.service.js @@ -0,0 +1,89 @@ +import crypto from 'crypto'; +import { models } from '../../../shared/database/models.js'; + +export const syncService = { + generateIdempotencyKey(tenantId, integrationId, productId, sourceVersion, operation) { + return crypto + .createHash('sha256') + .update(`${tenantId}:${integrationId}:${productId}:${sourceVersion || 1}:${operation || 'UPDATE'}`) + .digest('hex'); + }, + + async createSyncJob(tenantId, integrationId, triggerSource = 'manual') { + return await models.IntegrationSyncJob.create({ + tenant_id: tenantId, + integration_id: integrationId, + trigger_source: triggerSource, + status: 'pending', + total_items: 0, + success_items: 0, + failed_items: 0, + started_at: new Date() + }); + }, + + async createSyncItem(tenantId, syncJobId, integrationId, productId, operation = 'UPDATE', sourceVersion = 1) { + const idempotencyKey = this.generateIdempotencyKey(tenantId, integrationId, productId, sourceVersion, operation); + + // Find existing item or create + const [item] = await models.SyncItem.findOrCreate({ + where: { tenant_id: tenantId, idempotency_key: idempotencyKey }, + defaults: { + tenant_id: tenantId, + sync_job_id: syncJobId, + integration_id: integrationId, + product_id: productId, + operation, + status: 'pending', + source_version: sourceVersion, + idempotency_key: idempotencyKey, + attempt_count: 0 + } + }); + + return item; + }, + + async recordAttempt(tenantId, syncItemId, attemptNumber, requestMethod, requestUrl, idempotencyKey) { + return await models.SyncAttempt.create({ + tenant_id: tenantId, + sync_item_id: syncItemId, + attempt_number: attemptNumber, + started_at: new Date(), + status: 'processing', + request_method: requestMethod, + request_url: requestUrl, + idempotency_key: idempotencyKey + }); + }, + + async completeAttempt(attemptId, status, responseStatus, durationMs, errorCode = null, errorMessage = null) { + const attempt = await models.SyncAttempt.findByPk(attemptId); + if (attempt) { + await attempt.update({ + status, + response_status: responseStatus, + duration_ms: durationMs, + error_code: errorCode, + error_message: errorMessage, + completed_at: new Date() + }); + } + }, + + async logError(tenantId, syncJobId, syncItemId, errorCode, errorType, message, providerMessage = null, httpStatus = null, retryable = true, attemptNumber = 1, metadata = {}) { + return await models.SyncError.create({ + tenant_id: tenantId, + sync_job_id: syncJobId, + sync_item_id: syncItemId, + error_code: errorCode, + error_type: errorType, + message, + provider_message: providerMessage, + http_status: httpStatus, + retryable, + attempt_number: attemptNumber, + metadata + }); + } +}; diff --git a/src/features/integrations/index.js b/src/features/integrations/index.js index e0ceb85..f5b3f3c 100644 --- a/src/features/integrations/index.js +++ b/src/features/integrations/index.js @@ -1,32 +1,13 @@ -import { Router } from 'express'; +import router from './routes/integration.routes.js'; +import { startSyncWorker } from './workers/sync.worker.js'; +import { startOutboxWorker } from './workers/outbox.worker.js'; -const router = Router(); - -const defaultIntegrations = [ - { id: '1', name: 'Shopify Storefront Sync', type: 'ecommerce', status: 'active', target_channel: 'shopify', createdAt: new Date() }, - { id: '2', name: 'Amazon Seller Central', type: 'marketplace', status: 'active', target_channel: 'amazon', createdAt: new Date() }, - { id: '3', name: 'SAP ERP Connector', type: 'erp', status: 'inactive', target_channel: 'b2b', createdAt: new Date() } -]; - -router.get('/integrations', (req, res) => { - res.json({ success: true, data: defaultIntegrations }); -}); - -router.get('/integrations/:id', (req, res) => { - const item = defaultIntegrations.find(i => i.id === req.params.id) || defaultIntegrations[0]; - res.json({ success: true, data: item }); -}); - -router.post('/integrations', (req, res) => { - res.json({ success: true, data: { id: String(Date.now()), ...req.body, status: 'active' } }); -}); - -router.put('/integrations/:id', (req, res) => { - res.json({ success: true, data: { id: req.params.id, ...req.body } }); -}); - -router.delete('/integrations/:id', (req, res) => { - res.json({ success: true, message: 'Deleted' }); -}); +// Initialize background queue worker and outbox polling worker +try { + startSyncWorker(); + startOutboxWorker(); +} catch (err) { + console.warn('Background worker initialization deferred:', err.message); +} export default router; diff --git a/src/features/integrations/models/channelMapping.model.js b/src/features/integrations/models/channelMapping.model.js new file mode 100644 index 0000000..cb1abf5 --- /dev/null +++ b/src/features/integrations/models/channelMapping.model.js @@ -0,0 +1,72 @@ +import { DataTypes } from 'sequelize'; + +export default function (sequelize) { + const ChannelMapping = sequelize.define( + 'ChannelMapping', + { + id: { + type: DataTypes.UUID, + defaultValue: DataTypes.UUIDV4, + primaryKey: true + }, + tenant_id: { + type: DataTypes.INTEGER, + allowNull: true + }, + integration_id: { + type: DataTypes.UUID, + allowNull: false + }, + entity_type: { + type: DataTypes.STRING(50), + allowNull: false + }, + source_path: { + type: DataTypes.STRING(255), + allowNull: false + }, + target_path: { + type: DataTypes.STRING(255), + allowNull: false + }, + transformation_type: { + type: DataTypes.STRING(50), + defaultValue: 'string' + }, + transformation_config: { + type: DataTypes.JSONB, + defaultValue: {} + }, + default_value: { + type: DataTypes.TEXT, + allowNull: true + }, + required: { + type: DataTypes.BOOLEAN, + defaultValue: false + }, + version: { + type: DataTypes.INTEGER, + defaultValue: 1 + }, + status: { + type: DataTypes.STRING(20), + defaultValue: 'active' + } + }, + { + tableName: 'channel_mappings', + timestamps: true, + createdAt: 'created_at', + updatedAt: 'updated_at' + } + ); + + ChannelMapping.associate = (models) => { + if (models.Integration) { + ChannelMapping.belongsTo(models.Integration, { foreignKey: 'integration_id', as: 'integration' }); + } + }; + + return ChannelMapping; +} diff --git a/src/features/integrations/models/externalResource.model.js b/src/features/integrations/models/externalResource.model.js new file mode 100644 index 0000000..a86a978 --- /dev/null +++ b/src/features/integrations/models/externalResource.model.js @@ -0,0 +1,73 @@ +import { DataTypes } from 'sequelize'; + +export default function (sequelize) { + const ExternalResource = sequelize.define( + 'ExternalResource', + { + id: { + type: DataTypes.UUID, + defaultValue: DataTypes.UUIDV4, + primaryKey: true + }, + tenant_id: { + type: DataTypes.INTEGER, + allowNull: true + }, + integration_id: { + type: DataTypes.UUID, + allowNull: false + }, + resource_type: { + type: DataTypes.STRING(50), + allowNull: false + }, + source_type: { + type: DataTypes.STRING(50), + defaultValue: 'PIM' + }, + source_id: { + type: DataTypes.UUID, + allowNull: false + }, + external_id: { + type: DataTypes.STRING(255), + allowNull: false + }, + external_parent_id: { + type: DataTypes.STRING(255), + allowNull: true + }, + external_url: { + type: DataTypes.TEXT, + allowNull: true + }, + status: { + type: DataTypes.STRING(20), + defaultValue: 'active' + }, + last_synced_at: { + type: DataTypes.DATE, + allowNull: false, + defaultValue: DataTypes.NOW + }, + last_source_version: { + type: DataTypes.INTEGER, + defaultValue: 1 + } + }, + { + tableName: 'external_resources', + timestamps: true, + createdAt: 'created_at', + updatedAt: 'updated_at' + } + ); + + ExternalResource.associate = (models) => { + if (models.Integration) { + ExternalResource.belongsTo(models.Integration, { foreignKey: 'integration_id', as: 'integration' }); + } + }; + + return ExternalResource; +} diff --git a/src/features/integrations/models/integration.model.js b/src/features/integrations/models/integration.model.js new file mode 100644 index 0000000..5be74fd --- /dev/null +++ b/src/features/integrations/models/integration.model.js @@ -0,0 +1,82 @@ +import { DataTypes } from 'sequelize'; + +export default function (sequelize) { + const Integration = sequelize.define( + 'Integration', + { + id: { + type: DataTypes.UUID, + defaultValue: DataTypes.UUIDV4, + primaryKey: true + }, + tenant_id: { + type: DataTypes.INTEGER, + allowNull: true + }, + name: { + type: DataTypes.STRING(255), + allowNull: false + }, + channel: { + type: DataTypes.STRING(50), + allowNull: false + }, + integration_type: { + type: DataTypes.STRING(50), + allowNull: true, + defaultValue: 'ecommerce' + }, + status: { + type: DataTypes.STRING(20), + allowNull: false, + defaultValue: 'active' + }, + sync_mode: { + type: DataTypes.STRING(20), + allowNull: false, + defaultValue: 'auto' + }, + sync_frequency: { + type: DataTypes.STRING(50), + defaultValue: 'realtime' + }, + health_status: { + type: DataTypes.STRING(20), + defaultValue: 'healthy' + }, + last_synced_at: { + type: DataTypes.DATE, + allowNull: true + } + }, + { + tableName: 'integrations', + timestamps: true, + createdAt: 'created_at', + updatedAt: 'updated_at' + } + ); + + Integration.associate = (models) => { + if (models.Tenant) { + Integration.belongsTo(models.Tenant, { foreignKey: 'tenant_id', as: 'tenant' }); + } + if (models.IntegrationCredential) { + Integration.hasMany(models.IntegrationCredential, { foreignKey: 'integration_id', as: 'credentials' }); + } + if (models.PublishingRule) { + Integration.hasMany(models.PublishingRule, { foreignKey: 'integration_id', as: 'publishingRules' }); + } + if (models.ChannelMapping) { + Integration.hasMany(models.ChannelMapping, { foreignKey: 'integration_id', as: 'channelMappings' }); + } + if (models.SyncJob) { + Integration.hasMany(models.SyncJob, { foreignKey: 'integration_id', as: 'syncJobs' }); + } + if (models.ExternalResource) { + Integration.hasMany(models.ExternalResource, { foreignKey: 'integration_id', as: 'externalResources' }); + } + }; + + return Integration; +} diff --git a/src/features/integrations/models/integrationCredential.model.js b/src/features/integrations/models/integrationCredential.model.js new file mode 100644 index 0000000..8c59846 --- /dev/null +++ b/src/features/integrations/models/integrationCredential.model.js @@ -0,0 +1,53 @@ +import { DataTypes } from 'sequelize'; + +export default function (sequelize) { + const IntegrationCredential = sequelize.define( + 'IntegrationCredential', + { + id: { + type: DataTypes.UUID, + defaultValue: DataTypes.UUIDV4, + primaryKey: true + }, + tenant_id: { + type: DataTypes.INTEGER, + allowNull: true + }, + integration_id: { + type: DataTypes.UUID, + allowNull: false + }, + credential_type: { + type: DataTypes.STRING(50), + allowNull: false + }, + encrypted_secret: { + type: DataTypes.TEXT, + allowNull: false + }, + key_version: { + type: DataTypes.INTEGER, + defaultValue: 1, + allowNull: false + }, + expires_at: { + type: DataTypes.DATE, + allowNull: true + } + }, + { + tableName: 'integration_credentials', + timestamps: true, + createdAt: 'created_at', + updatedAt: 'updated_at' + } + ); + + IntegrationCredential.associate = (models) => { + if (models.Integration) { + IntegrationCredential.belongsTo(models.Integration, { foreignKey: 'integration_id', as: 'integration' }); + } + }; + + return IntegrationCredential; +} diff --git a/src/features/integrations/models/outboxEvent.model.js b/src/features/integrations/models/outboxEvent.model.js new file mode 100644 index 0000000..57eb923 --- /dev/null +++ b/src/features/integrations/models/outboxEvent.model.js @@ -0,0 +1,79 @@ +import { DataTypes } from 'sequelize'; + +export default function (sequelize) { + const OutboxEvent = sequelize.define( + 'OutboxEvent', + { + id: { + type: DataTypes.UUID, + defaultValue: DataTypes.UUIDV4, + primaryKey: true + }, + tenant_id: { + type: DataTypes.INTEGER, + allowNull: true + }, + event_type: { + type: DataTypes.STRING(100), + allowNull: false + }, + aggregate_type: { + type: DataTypes.STRING(50), + defaultValue: 'PRODUCT' + }, + aggregate_id: { + type: DataTypes.UUID, + allowNull: false + }, + payload: { + type: DataTypes.JSONB, + allowNull: false + }, + status: { + type: DataTypes.STRING(20), + defaultValue: 'pending', + allowNull: false + }, + event_version: { + type: DataTypes.INTEGER, + defaultValue: 1 + }, + idempotency_key: { + type: DataTypes.STRING(255), + allowNull: false + }, + retry_count: { + type: DataTypes.INTEGER, + defaultValue: 0 + }, + locked_at: { + type: DataTypes.DATE, + allowNull: true + }, + locked_by: { + type: DataTypes.STRING(100), + allowNull: true + }, + last_error: { + type: DataTypes.TEXT, + allowNull: true + }, + scheduled_at: { + type: DataTypes.DATE, + defaultValue: DataTypes.NOW + }, + processed_at: { + type: DataTypes.DATE, + allowNull: true + } + }, + { + tableName: 'outbox_events', + timestamps: true, + createdAt: 'created_at', + updatedAt: 'updated_at' + } + ); + + return OutboxEvent; +} diff --git a/src/features/integrations/models/publishingRule.model.js b/src/features/integrations/models/publishingRule.model.js new file mode 100644 index 0000000..5fe975d --- /dev/null +++ b/src/features/integrations/models/publishingRule.model.js @@ -0,0 +1,48 @@ +import { DataTypes } from 'sequelize'; + +export default function (sequelize) { + const PublishingRule = sequelize.define( + 'PublishingRule', + { + id: { + type: DataTypes.UUID, + defaultValue: DataTypes.UUIDV4, + primaryKey: true + }, + tenant_id: { + type: DataTypes.INTEGER, + allowNull: true + }, + integration_id: { + type: DataTypes.UUID, + allowNull: false + }, + name: { + type: DataTypes.STRING(255), + allowNull: false + }, + rule_type: { + type: DataTypes.STRING(50), + allowNull: false + }, + conditions_json: { + type: DataTypes.JSONB, + defaultValue: {} + } + }, + { + tableName: 'publishing_rules', + timestamps: true, + createdAt: 'created_at', + updatedAt: 'updated_at' + } + ); + + PublishingRule.associate = (models) => { + if (models.Integration) { + PublishingRule.belongsTo(models.Integration, { foreignKey: 'integration_id', as: 'integration' }); + } + }; + + return PublishingRule; +} diff --git a/src/features/integrations/models/syncAttempt.model.js b/src/features/integrations/models/syncAttempt.model.js new file mode 100644 index 0000000..f633138 --- /dev/null +++ b/src/features/integrations/models/syncAttempt.model.js @@ -0,0 +1,81 @@ +import { DataTypes } from 'sequelize'; + +export default function (sequelize) { + const SyncAttempt = sequelize.define( + 'SyncAttempt', + { + id: { + type: DataTypes.UUID, + defaultValue: DataTypes.UUIDV4, + primaryKey: true + }, + tenant_id: { + type: DataTypes.INTEGER, + allowNull: true + }, + sync_item_id: { + type: DataTypes.UUID, + allowNull: false + }, + attempt_number: { + type: DataTypes.INTEGER, + allowNull: false + }, + started_at: { + type: DataTypes.DATE, + allowNull: false, + defaultValue: DataTypes.NOW + }, + completed_at: { + type: DataTypes.DATE, + allowNull: true + }, + status: { + type: DataTypes.STRING(20), + allowNull: false + }, + request_method: { + type: DataTypes.STRING(10), + allowNull: false + }, + request_url: { + type: DataTypes.TEXT, + allowNull: false + }, + idempotency_key: { + type: DataTypes.STRING(255), + allowNull: true + }, + response_status: { + type: DataTypes.INTEGER, + allowNull: true + }, + error_code: { + type: DataTypes.STRING(100), + allowNull: true + }, + error_message: { + type: DataTypes.TEXT, + allowNull: true + }, + duration_ms: { + type: DataTypes.INTEGER, + allowNull: true + } + }, + { + tableName: 'sync_attempts', + timestamps: true, + createdAt: 'created_at', + updatedAt: false + } + ); + + SyncAttempt.associate = (models) => { + if (models.SyncItem) { + SyncAttempt.belongsTo(models.SyncItem, { foreignKey: 'sync_item_id', as: 'syncItem' }); + } + }; + + return SyncAttempt; +} diff --git a/src/features/integrations/models/syncError.model.js b/src/features/integrations/models/syncError.model.js new file mode 100644 index 0000000..e6dfab9 --- /dev/null +++ b/src/features/integrations/models/syncError.model.js @@ -0,0 +1,75 @@ +import { DataTypes } from 'sequelize'; + +export default function (sequelize) { + const SyncError = sequelize.define( + 'SyncError', + { + id: { + type: DataTypes.UUID, + defaultValue: DataTypes.UUIDV4, + primaryKey: true + }, + tenant_id: { + type: DataTypes.INTEGER, + allowNull: true + }, + sync_job_id: { + type: DataTypes.UUID, + allowNull: false + }, + sync_item_id: { + type: DataTypes.UUID, + allowNull: false + }, + error_code: { + type: DataTypes.STRING(100), + allowNull: false + }, + error_type: { + type: DataTypes.STRING(50), + allowNull: false + }, + message: { + type: DataTypes.TEXT, + allowNull: false + }, + provider_message: { + type: DataTypes.TEXT, + allowNull: true + }, + http_status: { + type: DataTypes.INTEGER, + allowNull: true + }, + retryable: { + type: DataTypes.BOOLEAN, + defaultValue: true + }, + attempt_number: { + type: DataTypes.INTEGER, + allowNull: false + }, + metadata: { + type: DataTypes.JSONB, + defaultValue: {} + } + }, + { + tableName: 'sync_errors', + timestamps: true, + createdAt: 'created_at', + updatedAt: false + } + ); + + SyncError.associate = (models) => { + if (models.SyncJob) { + SyncError.belongsTo(models.SyncJob, { foreignKey: 'sync_job_id', as: 'job' }); + } + if (models.SyncItem) { + SyncError.belongsTo(models.SyncItem, { foreignKey: 'sync_item_id', as: 'item' }); + } + }; + + return SyncError; +} diff --git a/src/features/integrations/models/syncItem.model.js b/src/features/integrations/models/syncItem.model.js new file mode 100644 index 0000000..c9c904f --- /dev/null +++ b/src/features/integrations/models/syncItem.model.js @@ -0,0 +1,97 @@ +import { DataTypes } from 'sequelize'; + +export default function (sequelize) { + const SyncItem = sequelize.define( + 'SyncItem', + { + id: { + type: DataTypes.UUID, + defaultValue: DataTypes.UUIDV4, + primaryKey: true + }, + tenant_id: { + type: DataTypes.INTEGER, + allowNull: true + }, + sync_job_id: { + type: DataTypes.UUID, + allowNull: false + }, + integration_id: { + type: DataTypes.UUID, + allowNull: false + }, + product_id: { + type: DataTypes.UUID, + allowNull: false + }, + variant_id: { + type: DataTypes.UUID, + allowNull: true + }, + sku: { + type: DataTypes.STRING(100), + allowNull: true + }, + operation: { + type: DataTypes.STRING(20), + allowNull: false + }, + status: { + type: DataTypes.STRING(20), + allowNull: false, + defaultValue: 'pending' + }, + source_version: { + type: DataTypes.INTEGER, + defaultValue: 1 + }, + idempotency_key: { + type: DataTypes.STRING(255), + allowNull: false + }, + external_resource_id: { + type: DataTypes.UUID, + allowNull: true + }, + attempt_count: { + type: DataTypes.INTEGER, + defaultValue: 0 + }, + error_code: { + type: DataTypes.STRING(100), + allowNull: true + }, + error_message: { + type: DataTypes.TEXT, + allowNull: true + } + }, + { + tableName: 'sync_items', + timestamps: true, + createdAt: 'created_at', + updatedAt: 'updated_at' + } + ); + + SyncItem.associate = (models) => { + if (models.SyncJob) { + SyncItem.belongsTo(models.SyncJob, { foreignKey: 'sync_job_id', as: 'job' }); + } + if (models.Integration) { + SyncItem.belongsTo(models.Integration, { foreignKey: 'integration_id', as: 'integration' }); + } + if (models.Product) { + SyncItem.belongsTo(models.Product, { foreignKey: 'product_id', as: 'product' }); + } + if (models.SyncAttempt) { + SyncItem.hasMany(models.SyncAttempt, { foreignKey: 'sync_item_id', as: 'attempts' }); + } + if (models.SyncError) { + SyncItem.hasMany(models.SyncError, { foreignKey: 'sync_item_id', as: 'errors' }); + } + }; + + return SyncItem; +} diff --git a/src/features/integrations/models/syncJob.model.js b/src/features/integrations/models/syncJob.model.js new file mode 100644 index 0000000..e2b211f --- /dev/null +++ b/src/features/integrations/models/syncJob.model.js @@ -0,0 +1,72 @@ +import { DataTypes } from 'sequelize'; + +export default function (sequelize) { + const SyncJob = sequelize.define( + 'SyncJob', + { + id: { + type: DataTypes.UUID, + defaultValue: DataTypes.UUIDV4, + primaryKey: true + }, + tenant_id: { + type: DataTypes.INTEGER, + allowNull: true + }, + integration_id: { + type: DataTypes.UUID, + allowNull: false + }, + trigger_source: { + type: DataTypes.STRING(50), + allowNull: false, + defaultValue: 'manual' + }, + status: { + type: DataTypes.STRING(20), + allowNull: false, + defaultValue: 'pending' + }, + total_items: { + type: DataTypes.INTEGER, + defaultValue: 0 + }, + success_items: { + type: DataTypes.INTEGER, + defaultValue: 0 + }, + failed_items: { + type: DataTypes.INTEGER, + defaultValue: 0 + }, + started_at: { + type: DataTypes.DATE, + allowNull: true + }, + completed_at: { + type: DataTypes.DATE, + allowNull: true + } + }, + { + tableName: 'sync_jobs', + timestamps: true, + createdAt: 'created_at', + updatedAt: 'updated_at' + } + ); + + SyncJob.associate = (models) => { + if (models.Integration) { + SyncJob.belongsTo(models.Integration, { foreignKey: 'integration_id', as: 'integration' }); + } + if (models.SyncItem) { + SyncJob.hasMany(models.SyncItem, { foreignKey: 'sync_job_id', as: 'items' }); + } + if (models.SyncError) { + SyncJob.hasMany(models.SyncError, { foreignKey: 'sync_job_id', as: 'errors' }); + } + }; + + return SyncJob; +} diff --git a/src/features/integrations/projection/canonical-product.builder.js b/src/features/integrations/projection/canonical-product.builder.js new file mode 100644 index 0000000..68e919f --- /dev/null +++ b/src/features/integrations/projection/canonical-product.builder.js @@ -0,0 +1,142 @@ +import { models } from '../../../shared/database/models.js'; + +export const canonicalProductBuilder = { + async build(productId, tenantId) { + const product = await models.Product.findOne({ + where: { id: productId, tenant_id: tenantId }, + include: [ + { model: models.Brand, as: 'brand', required: false }, + { model: models.Categorie, as: 'category', required: false }, + { model: models.Catalog, as: 'family', required: false }, + { + model: models.ProductAttributeValue, + as: 'attributeValues', + required: false, + include: [{ model: models.Attribute, as: 'attribute', required: false }] + }, + { + model: models.ProductAsset, + as: 'productAssets', + required: false, + include: [{ model: models.Asset, as: 'asset', required: false }] + } + ] + }); + + if (!product) { + throw new Error(`Product ${productId} not found for tenant ${tenantId}`); + } + + // Fetch product variants + const variants = await models.Variant.findAll({ + where: { product_id: productId, tenant_id: tenantId }, + include: [ + { + model: models.VariantValue, + as: 'values', + required: false, + include: [{ model: models.Attribute, as: 'axis', required: false }] + }, + { + model: models.VariantAsset, + as: 'variantAssets', + required: false, + include: [{ model: models.Asset, as: 'asset', required: false }] + } + ] + }); + + // Compile attributes map + const attributes = {}; + if (Array.isArray(product.attributeValues)) { + product.attributeValues.forEach(av => { + const key = av.attribute?.code || av.axis?.code || av.attribute_id; + attributes[key] = av.value || av.value_text || av.value_number || av.value_boolean; + }); + } + + // Compile media assets + const media = []; + const baseUrl = process.env.PUBLIC_APP_URL || process.env.APP_URL || 'http://localhost:5002'; + if (Array.isArray(product.productAssets)) { + product.productAssets.forEach(pa => { + if (pa.asset) { + let url = pa.asset.file_url || ''; + if (url.startsWith('/')) { + url = `${baseUrl}${url}`; + } + media.push({ + id: pa.asset.id, + url, + role: pa.role || 'gallery', + isPrimary: pa.is_primary || false + }); + } + }); + } + + // Compile variants list + const compiledVariants = variants.map(v => { + const vAttrs = {}; + if (Array.isArray(v.values)) { + v.values.forEach(val => { + const key = val.axis?.code || val.attribute_id; + vAttrs[key] = val.value_text || val.value_number; + }); + } + + const vMedia = []; + if (Array.isArray(v.variantAssets)) { + v.variantAssets.forEach(va => { + if (va.asset) { + let url = va.asset.file_url || ''; + if (url.startsWith('/')) { + url = `${baseUrl}${url}`; + } + vMedia.push({ + id: va.asset.id, + url, + role: va.role || 'gallery', + isPrimary: va.is_primary || false + }); + } + }); + } + + return { + id: v.id, + sku: v.sku, + name: v.name, + price: parseFloat(v.price) || 0, + costPrice: parseFloat(v.costPrice) || 0, + stock: parseInt(v.stock, 10) || 0, + status: v.status, + attributes: vAttrs, + media: vMedia + }; + }); + + return { + identity: { + id: product.id, + tenantId: product.tenant_id, + code: product.code || product.id, + sku: product.sku || '' + }, + content: { + name: product.name, + description: product.description || '', + status: product.status || 'draft' + }, + taxonomy: { + brand: product.brand ? { id: product.brand.id, name: product.brand.name } : null, + category: product.category ? { id: product.category.id, name: product.category.name } : null, + family: product.family ? { id: product.family.id, name: product.family.name } : null + }, + attributes, + media, + variants: compiledVariants, + version: product.updated_at ? new Date(product.updated_at).getTime() : 1 + }; + } +}; diff --git a/src/features/integrations/routes/integration.routes.js b/src/features/integrations/routes/integration.routes.js new file mode 100644 index 0000000..1304b24 --- /dev/null +++ b/src/features/integrations/routes/integration.routes.js @@ -0,0 +1,36 @@ +import { Router } from 'express'; +import { integrationController } from '../controllers/integration.controller.js'; +import { authenticate } from '../../../shared/middleware/auth.middleware.js'; +import { authorize } from '../../../shared/middleware/permission.middleware.js'; + +const router = Router(); + +// ─── PUBLIC ROUTE — Shopify OAuth callback (no auth; Shopify redirects here) ── +router.get( + '/integrations/shopify/oauth/callback', + integrationController.handleShopifyOAuthCallback +); + +// ─── Protected routes ───────────────────────────────────────────────────────── +router.use(authenticate); + +router.get('/integrations', authorize(['settings.integrations']), integrationController.listIntegrations); +router.post('/integrations', authorize(['settings.integrations']), integrationController.createIntegration); + +router.get('/integrations/:id', authorize(['settings.integrations']), integrationController.getIntegration); +router.put('/integrations/:id', authorize(['settings.integrations']), integrationController.updateIntegration); +router.delete('/integrations/:id', authorize(['settings.integrations']), integrationController.deleteIntegration); + +router.get('/integrations/:id/credentials', authorize(['settings.integrations']), integrationController.getCredentials); +router.post('/integrations/:id/credentials', authorize(['settings.integrations']), integrationController.setCredentials); +router.post('/integrations/:id/test-connection', authorize(['settings.integrations']), integrationController.testConnection); +router.post('/integrations/:id/sync', authorize(['settings.integrations']), integrationController.triggerSync); + +// ─── Shopify OAuth — Start flow (protected; user-initiated) ────────────────── +router.post('/integrations/:id/shopify/oauth/start', authorize(['settings.integrations']), integrationController.startShopifyOAuth); + +router.get('/integrations/jobs/all', authorize(['settings.integrations']), integrationController.listAllSyncJobs); +router.get('/integrations/:id/jobs', authorize(['settings.integrations']), integrationController.listSyncJobs); +router.get('/integrations/jobs/:jobId/items', authorize(['settings.integrations']), integrationController.listSyncItems); + +export default router; diff --git a/src/features/integrations/workers/outbox.worker.js b/src/features/integrations/workers/outbox.worker.js new file mode 100644 index 0000000..40389fb --- /dev/null +++ b/src/features/integrations/workers/outbox.worker.js @@ -0,0 +1,84 @@ +import { models, sequelize } from '../../../shared/database/models.js'; +import { queueService } from '../../../shared/infrastructure/queue/queue.service.js'; +import { QUEUE_NAMES } from '../../../shared/infrastructure/queue/queue.constants.js'; +import { syncService } from '../domain/sync.service.js'; +import { Op } from 'sequelize'; +import winston from 'winston'; + +const logger = winston.createLogger({ + level: 'info', + format: winston.format.combine(winston.format.timestamp(), winston.format.json()), + transports: [new winston.transports.Console()] +}); + +export const processOutboxEvents = async () => { + const workerId = `worker_${process.pid}_${Date.now()}`; + const now = new Date(); + + try { + // 1. Claim pending outbox events using locking + const [claimedCount] = await models.OutboxEvent.update( + { + status: 'processing', + locked_at: now, + locked_by: workerId + }, + { + where: { + status: 'pending', + scheduled_at: { [Op.lte]: now } + }, + limit: 50 + } + ); + + if (claimedCount === 0) return; + + // 2. Fetch claimed events + const claimedEvents = await models.OutboxEvent.findAll({ + where: { + locked_by: workerId, + status: 'processing' + } + }); + + for (const event of claimedEvents) { + try { + const { tenant_id, aggregate_id, event_type, payload } = event; + const productId = aggregate_id; + + // Find active integrations for tenant + const activeIntegrations = await models.Integration.findAll({ + where: { tenant_id, status: 'active', sync_mode: 'auto' } + }); + + for (const integration of activeIntegrations) { + const syncJob = await syncService.createSyncJob(tenant_id, integration.id, 'outbox'); + + await queueService.addJob(QUEUE_NAMES.INTEGRATION_SYNC, `sync_product_${productId}`, { + tenantId: tenant_id, + integrationId: integration.id, + syncJobId: syncJob.id, + productId, + operation: event_type.includes('delete') ? 'DELETE' : 'UPDATE' + }); + } + + await event.update({ status: 'completed', processed_at: new Date() }); + } catch (err) { + logger.error(`Error processing outbox event ${event.id}: ${err.message}`); + await event.update({ + status: 'failed', + retry_count: (event.retry_count || 0) + 1, + last_error: err.message + }); + } + } + } catch (err) { + logger.error(`Outbox worker failure: ${err.message}`); + } +}; + +export const startOutboxWorker = (intervalMs = 10000) => { + setInterval(processOutboxEvents, intervalMs); +}; diff --git a/src/features/integrations/workers/sync.worker.js b/src/features/integrations/workers/sync.worker.js new file mode 100644 index 0000000..f91bf63 --- /dev/null +++ b/src/features/integrations/workers/sync.worker.js @@ -0,0 +1,127 @@ +import { queueService } from '../../../shared/infrastructure/queue/queue.service.js'; +import { QUEUE_NAMES } from '../../../shared/infrastructure/queue/queue.constants.js'; +import { canonicalProductBuilder } from '../projection/canonical-product.builder.js'; +import { ruleService } from '../domain/rule.service.js'; +import { syncService } from '../domain/sync.service.js'; +import { integrationService } from '../domain/integration.service.js'; +import { adapterRegistry } from '../adapters/adapter.registry.js'; +import { models } from '../../../shared/database/models.js'; +import winston from 'winston'; + +const logger = winston.createLogger({ + level: 'info', + format: winston.format.combine(winston.format.timestamp(), winston.format.json()), + transports: [new winston.transports.Console()] +}); + +export const startSyncWorker = () => { + queueService.registerWorker(QUEUE_NAMES.INTEGRATION_SYNC, async (job) => { + const { tenantId, integrationId, syncJobId, productId, operation = 'UPDATE' } = job.data; + logger.info(`Processing sync job for product ${productId} on integration ${integrationId}`); + + // Build canonical product + const canonicalProduct = await canonicalProductBuilder.build(productId, tenantId); + + // Create sync item + const syncItem = await syncService.createSyncItem( + tenantId, + syncJobId, + integrationId, + productId, + operation, + canonicalProduct.version + ); + + // Rule evaluation + const ruleResult = await ruleService.evaluateRules(integrationId, tenantId, canonicalProduct); + if (!ruleResult.eligible) { + await syncItem.update({ + status: 'skipped', + error_code: 'RULE_INELIGIBLE', + error_message: ruleResult.reasons.join('; ') + }); + logger.info(`Skipped product ${productId}: ${ruleResult.reasons.join('; ')}`); + return { skipped: true, reasons: ruleResult.reasons }; + } + + const attemptNumber = (syncItem.attempt_count || 0) + 1; + const requestUrl = `integration://${integrationId}/product/${productId}`; + const attemptRecord = await syncService.recordAttempt( + tenantId, + syncItem.id, + attemptNumber, + 'POST', + requestUrl, + syncItem.idempotency_key + ); + + const startTime = Date.now(); + try { + await syncItem.update({ status: 'processing', attempt_count: attemptNumber }); + + const integration = await integrationService.getById(integrationId, tenantId); + const decryptedCredentials = await integrationService.getDecryptedCredentials(integrationId, tenantId); + const adapter = adapterRegistry.getAdapter(integration.channel, decryptedCredentials); + + let result; + if (operation === 'DELETE') { + result = await adapter.deleteProduct({ tenantId, integrationId, productId }); + } else { + result = await adapter.publishProduct({ tenantId, integrationId, canonicalProduct }); + } + + const durationMs = Date.now() - startTime; + await syncService.completeAttempt(attemptRecord.id, 'success', 200, durationMs); + await syncItem.update({ status: 'success', error_code: null, error_message: null }); + + // Increment job success count and check completion + if (syncJobId) { + await models.IntegrationSyncJob.increment('success_items', { where: { id: syncJobId } }); + const jobRecord = await models.IntegrationSyncJob.findByPk(syncJobId); + if (jobRecord && (jobRecord.success_items + jobRecord.failed_items) >= jobRecord.total_items) { + await jobRecord.update({ + status: jobRecord.failed_items === 0 ? 'completed' : (jobRecord.success_items > 0 ? 'completed' : 'failed'), + completed_at: new Date() + }); + } + } + + logger.info(`Successfully synced product ${productId} to ${integration.channel}`); + return result; + } catch (err) { + const durationMs = Date.now() - startTime; + const httpStatus = err.status || 500; + const errorMessage = err.message || 'Unknown integration error'; + + await syncService.completeAttempt(attemptRecord.id, 'failed', httpStatus, durationMs, 'ADAPTER_ERROR', errorMessage); + await syncItem.update({ status: 'failed', error_code: 'ADAPTER_ERROR', error_message: errorMessage }); + + await syncService.logError( + tenantId, + syncJobId, + syncItem.id, + 'ADAPTER_ERROR', + httpStatus === 429 ? 'RATE_LIMIT' : 'PROVIDER_ERROR', + errorMessage, + err.providerData ? JSON.stringify(err.providerData) : null, + httpStatus, + err.retryable !== false, + attemptNumber, + { productId, integrationId } + ); + + if (syncJobId) { + await models.IntegrationSyncJob.increment('failed_items', { where: { id: syncJobId } }); + const jobRecord = await models.IntegrationSyncJob.findByPk(syncJobId); + if (jobRecord && (jobRecord.success_items + jobRecord.failed_items) >= jobRecord.total_items) { + await jobRecord.update({ + status: jobRecord.success_items > 0 ? 'completed' : 'failed', + completed_at: new Date() + }); + } + } + + throw err; + } + }); +}; diff --git a/src/migrations/20260830000001-create-syndication-and-outbox-tables.cjs b/src/migrations/20260830000001-create-syndication-and-outbox-tables.cjs new file mode 100644 index 0000000..c07cba2 --- /dev/null +++ b/src/migrations/20260830000001-create-syndication-and-outbox-tables.cjs @@ -0,0 +1,751 @@ +'use strict'; + +module.exports = { + async up(queryInterface, Sequelize) { + // 1. integrations + const integrationsExists = await queryInterface.describeTable('integrations').catch(() => null); + if (!integrationsExists) { + await queryInterface.createTable('integrations', { + id: { + type: Sequelize.UUID, + defaultValue: Sequelize.UUIDV4, + primaryKey: true, + allowNull: false + }, + tenant_id: { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + name: { + type: Sequelize.STRING(255), + allowNull: false + }, + channel: { + type: Sequelize.STRING(50), + allowNull: false + }, + status: { + type: Sequelize.STRING(20), + allowNull: false, + defaultValue: 'active' + }, + sync_mode: { + type: Sequelize.STRING(20), + allowNull: false, + defaultValue: 'auto' + }, + sync_frequency: { + type: Sequelize.STRING(50), + defaultValue: 'realtime' + }, + health_status: { + type: Sequelize.STRING(20), + defaultValue: 'healthy' + }, + last_synced_at: { + type: Sequelize.DATE, + allowNull: true + }, + created_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + updated_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + } + }); + await queryInterface.addIndex('integrations', ['tenant_id', 'channel'], { name: 'idx_integrations_tenant_channel' }).catch(() => null); + } else { + if (!integrationsExists.channel) { + await queryInterface.addColumn('integrations', 'channel', { type: Sequelize.STRING(50), allowNull: false, defaultValue: 'shopify' }); + } + if (!integrationsExists.tenant_id) { + await queryInterface.addColumn('integrations', 'tenant_id', { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }); + } + if (!integrationsExists.sync_mode) { + await queryInterface.addColumn('integrations', 'sync_mode', { type: Sequelize.STRING(20), defaultValue: 'auto' }); + } + if (!integrationsExists.sync_frequency) { + await queryInterface.addColumn('integrations', 'sync_frequency', { type: Sequelize.STRING(50), defaultValue: 'realtime' }); + } + if (!integrationsExists.health_status) { + await queryInterface.addColumn('integrations', 'health_status', { type: Sequelize.STRING(20), defaultValue: 'healthy' }); + } + if (!integrationsExists.last_synced_at) { + await queryInterface.addColumn('integrations', 'last_synced_at', { type: Sequelize.DATE, allowNull: true }); + } + } + + // 2. integration_credentials + const credentialsExists = await queryInterface.describeTable('integration_credentials').catch(() => null); + if (!credentialsExists) { + await queryInterface.createTable('integration_credentials', { + id: { + type: Sequelize.UUID, + defaultValue: Sequelize.UUIDV4, + primaryKey: true, + allowNull: false + }, + tenant_id: { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + integration_id: { + type: Sequelize.UUID, + allowNull: false, + references: { model: 'integrations', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + credential_type: { + type: Sequelize.STRING(50), + allowNull: false + }, + encrypted_secret: { + type: Sequelize.TEXT, + allowNull: false + }, + key_version: { + type: Sequelize.INTEGER, + defaultValue: 1, + allowNull: false + }, + expires_at: { + type: Sequelize.DATE, + allowNull: true + }, + created_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + updated_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + } + }); + await queryInterface.addIndex('integration_credentials', ['integration_id'], { name: 'idx_integration_credentials_integration' }).catch(() => null); + } + + // 3. publishing_rules + const rulesExists = await queryInterface.describeTable('publishing_rules').catch(() => null); + if (!rulesExists) { + await queryInterface.createTable('publishing_rules', { + id: { + type: Sequelize.UUID, + defaultValue: Sequelize.UUIDV4, + primaryKey: true, + allowNull: false + }, + tenant_id: { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + integration_id: { + type: Sequelize.UUID, + allowNull: false, + references: { model: 'integrations', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + name: { + type: Sequelize.STRING(255), + allowNull: false + }, + rule_type: { + type: Sequelize.STRING(50), + allowNull: false + }, + conditions_json: { + type: Sequelize.JSONB, + defaultValue: {} + }, + created_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + updated_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + } + }); + } + + // 4. channel_mappings + const mappingsExists = await queryInterface.describeTable('channel_mappings').catch(() => null); + if (!mappingsExists) { + await queryInterface.createTable('channel_mappings', { + id: { + type: Sequelize.UUID, + defaultValue: Sequelize.UUIDV4, + primaryKey: true, + allowNull: false + }, + tenant_id: { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + integration_id: { + type: Sequelize.UUID, + allowNull: false, + references: { model: 'integrations', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + entity_type: { + type: Sequelize.STRING(50), + allowNull: false + }, + source_path: { + type: Sequelize.STRING(255), + allowNull: false + }, + target_path: { + type: Sequelize.STRING(255), + allowNull: false + }, + transformation_type: { + type: Sequelize.STRING(50), + defaultValue: 'string' + }, + transformation_config: { + type: Sequelize.JSONB, + defaultValue: {} + }, + default_value: { + type: Sequelize.TEXT, + allowNull: true + }, + required: { + type: Sequelize.BOOLEAN, + defaultValue: false + }, + version: { + type: Sequelize.INTEGER, + defaultValue: 1 + }, + status: { + type: Sequelize.STRING(20), + defaultValue: 'active' + }, + created_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + updated_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + } + }); + } + + // 5. sync_jobs + const jobsExists = await queryInterface.describeTable('sync_jobs').catch(() => null); + if (!jobsExists) { + await queryInterface.createTable('sync_jobs', { + id: { + type: Sequelize.UUID, + defaultValue: Sequelize.UUIDV4, + primaryKey: true, + allowNull: false + }, + tenant_id: { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + integration_id: { + type: Sequelize.UUID, + allowNull: false, + references: { model: 'integrations', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + trigger_source: { + type: Sequelize.STRING(50), + allowNull: false, + defaultValue: 'manual' + }, + status: { + type: Sequelize.STRING(20), + allowNull: false, + defaultValue: 'pending' + }, + total_items: { + type: Sequelize.INTEGER, + defaultValue: 0 + }, + success_items: { + type: Sequelize.INTEGER, + defaultValue: 0 + }, + failed_items: { + type: Sequelize.INTEGER, + defaultValue: 0 + }, + started_at: { + type: Sequelize.DATE, + allowNull: true + }, + completed_at: { + type: Sequelize.DATE, + allowNull: true + }, + created_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + updated_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + } + }); + } + + // 6. sync_items + const itemsExists = await queryInterface.describeTable('sync_items').catch(() => null); + if (!itemsExists) { + await queryInterface.createTable('sync_items', { + id: { + type: Sequelize.UUID, + defaultValue: Sequelize.UUIDV4, + primaryKey: true, + allowNull: false + }, + tenant_id: { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + sync_job_id: { + type: Sequelize.UUID, + allowNull: false, + references: { model: 'sync_jobs', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + integration_id: { + type: Sequelize.UUID, + allowNull: false, + references: { model: 'integrations', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + product_id: { + type: Sequelize.UUID, + allowNull: false + }, + variant_id: { + type: Sequelize.UUID, + allowNull: true + }, + sku: { + type: Sequelize.STRING(100), + allowNull: true + }, + operation: { + type: Sequelize.STRING(20), + allowNull: false + }, + status: { + type: Sequelize.STRING(20), + allowNull: false, + defaultValue: 'pending' + }, + source_version: { + type: Sequelize.INTEGER, + defaultValue: 1 + }, + idempotency_key: { + type: Sequelize.STRING(255), + allowNull: false + }, + external_resource_id: { + type: Sequelize.UUID, + allowNull: true + }, + attempt_count: { + type: Sequelize.INTEGER, + defaultValue: 0 + }, + error_code: { + type: Sequelize.STRING(100), + allowNull: true + }, + error_message: { + type: Sequelize.TEXT, + allowNull: true + }, + created_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + updated_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + } + }); + await queryInterface.addIndex('sync_items', ['tenant_id', 'idempotency_key'], { + name: 'uq_sync_item_idempotency', + unique: true + }).catch(() => null); + } + + // 7. sync_attempts + const attemptsExists = await queryInterface.describeTable('sync_attempts').catch(() => null); + if (!attemptsExists) { + await queryInterface.createTable('sync_attempts', { + id: { + type: Sequelize.UUID, + defaultValue: Sequelize.UUIDV4, + primaryKey: true, + allowNull: false + }, + tenant_id: { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + sync_item_id: { + type: Sequelize.UUID, + allowNull: false, + references: { model: 'sync_items', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + attempt_number: { + type: Sequelize.INTEGER, + allowNull: false + }, + started_at: { + type: Sequelize.DATE, + allowNull: false, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + completed_at: { + type: Sequelize.DATE, + allowNull: true + }, + status: { + type: Sequelize.STRING(20), + allowNull: false + }, + request_method: { + type: Sequelize.STRING(10), + allowNull: false + }, + request_url: { + type: Sequelize.TEXT, + allowNull: false + }, + idempotency_key: { + type: Sequelize.STRING(255), + allowNull: true + }, + response_status: { + type: Sequelize.INTEGER, + allowNull: true + }, + error_code: { + type: Sequelize.STRING(100), + allowNull: true + }, + error_message: { + type: Sequelize.TEXT, + allowNull: true + }, + duration_ms: { + type: Sequelize.INTEGER, + allowNull: true + }, + created_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + } + }); + } + + // 8. sync_errors + const errorsExists = await queryInterface.describeTable('sync_errors').catch(() => null); + if (!errorsExists) { + await queryInterface.createTable('sync_errors', { + id: { + type: Sequelize.UUID, + defaultValue: Sequelize.UUIDV4, + primaryKey: true, + allowNull: false + }, + tenant_id: { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + sync_job_id: { + type: Sequelize.UUID, + allowNull: false, + references: { model: 'sync_jobs', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + sync_item_id: { + type: Sequelize.UUID, + allowNull: false, + references: { model: 'sync_items', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + error_code: { + type: Sequelize.STRING(100), + allowNull: false + }, + error_type: { + type: Sequelize.STRING(50), + allowNull: false + }, + message: { + type: Sequelize.TEXT, + allowNull: false + }, + provider_message: { + type: Sequelize.TEXT, + allowNull: true + }, + http_status: { + type: Sequelize.INTEGER, + allowNull: true + }, + retryable: { + type: Sequelize.BOOLEAN, + defaultValue: true + }, + attempt_number: { + type: Sequelize.INTEGER, + allowNull: false + }, + metadata: { + type: Sequelize.JSONB, + defaultValue: {} + }, + created_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + } + }); + } + + // 9. external_resources + const externalExists = await queryInterface.describeTable('external_resources').catch(() => null); + if (!externalExists) { + await queryInterface.createTable('external_resources', { + id: { + type: Sequelize.UUID, + defaultValue: Sequelize.UUIDV4, + primaryKey: true, + allowNull: false + }, + tenant_id: { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + integration_id: { + type: Sequelize.UUID, + allowNull: false, + references: { model: 'integrations', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + resource_type: { + type: Sequelize.STRING(50), + allowNull: false + }, + source_type: { + type: Sequelize.STRING(50), + defaultValue: 'PIM' + }, + source_id: { + type: Sequelize.UUID, + allowNull: false + }, + external_id: { + type: Sequelize.STRING(255), + allowNull: false + }, + external_parent_id: { + type: Sequelize.STRING(255), + allowNull: true + }, + external_url: { + type: Sequelize.TEXT, + allowNull: true + }, + status: { + type: Sequelize.STRING(20), + defaultValue: 'active' + }, + last_synced_at: { + type: Sequelize.DATE, + allowNull: false, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + last_source_version: { + type: Sequelize.INTEGER, + defaultValue: 1 + }, + created_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + updated_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + } + }); + await queryInterface.addIndex('external_resources', ['tenant_id', 'integration_id', 'source_id', 'resource_type'], { + name: 'uq_external_resource_mapping', + unique: true + }).catch(() => null); + } + + // 10. outbox_events + const outboxExists = await queryInterface.describeTable('outbox_events').catch(() => null); + if (!outboxExists) { + await queryInterface.createTable('outbox_events', { + id: { + type: Sequelize.UUID, + defaultValue: Sequelize.UUIDV4, + primaryKey: true, + allowNull: false + }, + tenant_id: { + type: Sequelize.INTEGER, + allowNull: true, + references: { model: 'tenants', key: 'id' }, + onUpdate: 'CASCADE', + onDelete: 'CASCADE' + }, + event_type: { + type: Sequelize.STRING(100), + allowNull: false + }, + aggregate_type: { + type: Sequelize.STRING(50), + defaultValue: 'PRODUCT' + }, + aggregate_id: { + type: Sequelize.UUID, + allowNull: false + }, + payload: { + type: Sequelize.JSONB, + allowNull: false + }, + status: { + type: Sequelize.STRING(20), + defaultValue: 'pending', + allowNull: false + }, + event_version: { + type: Sequelize.INTEGER, + defaultValue: 1 + }, + idempotency_key: { + type: Sequelize.STRING(255), + allowNull: false + }, + retry_count: { + type: Sequelize.INTEGER, + defaultValue: 0 + }, + locked_at: { + type: Sequelize.DATE, + allowNull: true + }, + locked_by: { + type: Sequelize.STRING(100), + allowNull: true + }, + last_error: { + type: Sequelize.TEXT, + allowNull: true + }, + scheduled_at: { + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + processed_at: { + type: Sequelize.DATE, + allowNull: true + }, + created_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + }, + updated_at: { + allowNull: false, + type: Sequelize.DATE, + defaultValue: Sequelize.literal('CURRENT_TIMESTAMP') + } + }); + await queryInterface.addIndex('outbox_events', ['tenant_id', 'status', 'scheduled_at'], { name: 'idx_outbox_events_status_scheduled' }).catch(() => null); + } + }, + + async down(queryInterface, Sequelize) { + await queryInterface.dropTable('outbox_events').catch(() => null); + await queryInterface.dropTable('external_resources').catch(() => null); + await queryInterface.dropTable('sync_errors').catch(() => null); + await queryInterface.dropTable('sync_attempts').catch(() => null); + await queryInterface.dropTable('sync_items').catch(() => null); + await queryInterface.dropTable('sync_jobs').catch(() => null); + await queryInterface.dropTable('channel_mappings').catch(() => null); + await queryInterface.dropTable('publishing_rules').catch(() => null); + await queryInterface.dropTable('integration_credentials').catch(() => null); + await queryInterface.dropTable('integrations').catch(() => null); + } +}; diff --git a/src/shared/database/models.js b/src/shared/database/models.js index 5ab30f2..028cfeb 100644 --- a/src/shared/database/models.js +++ b/src/shared/database/models.js @@ -191,6 +191,17 @@ import productAttributeValueModelInit from '../../features/products/products/pro import productVariantValueModelInit from '../../features/products/products/productVariantValue.model.js'; import productCompletenessModelInit from '../../features/products/products/productCompleteness.model.js'; +import integrationModelInit from '../../features/integrations/models/integration.model.js'; +import integrationCredentialModelInit from '../../features/integrations/models/integrationCredential.model.js'; +import publishingRuleModelInit from '../../features/integrations/models/publishingRule.model.js'; +import integrationChannelMappingModelInit from '../../features/integrations/models/channelMapping.model.js'; +import integrationSyncJobModelInit from '../../features/integrations/models/syncJob.model.js'; +import syncItemModelInit from '../../features/integrations/models/syncItem.model.js'; +import syncAttemptModelInit from '../../features/integrations/models/syncAttempt.model.js'; +import syncErrorModelInit from '../../features/integrations/models/syncError.model.js'; +import externalResourceModelInit from '../../features/integrations/models/externalResource.model.js'; +import outboxEventModelInit from '../../features/integrations/models/outboxEvent.model.js'; + export const initializeDatabaseModels = () => { registerModel('Tenant', tenantModelInit); registerModel('User', userModelInit); @@ -239,6 +250,18 @@ export const initializeDatabaseModels = () => { registerModel('ProductVariantValue', productVariantValueModelInit); registerModel('ProductCompleteness', productCompletenessModelInit); + // Integration Hub & Outbox Platform + registerModel('Integration', integrationModelInit); + registerModel('IntegrationCredential', integrationCredentialModelInit); + registerModel('PublishingRule', publishingRuleModelInit); + registerModel('IntegrationChannelMapping', integrationChannelMappingModelInit); + registerModel('IntegrationSyncJob', integrationSyncJobModelInit); + registerModel('SyncItem', syncItemModelInit); + registerModel('SyncAttempt', syncAttemptModelInit); + registerModel('SyncError', syncErrorModelInit); + registerModel('ExternalResource', externalResourceModelInit); + registerModel('OutboxEvent', outboxEventModelInit); + // Variants registerModel('Variant', variantModelInit); registerModel('VariantValue', variantValueModelInit); diff --git a/src/shared/infrastructure/cache/cache.service.js b/src/shared/infrastructure/cache/cache.service.js new file mode 100644 index 0000000..6cac038 --- /dev/null +++ b/src/shared/infrastructure/cache/cache.service.js @@ -0,0 +1,57 @@ +import { getRedisClient, isRedisReady } from './redis.client.js'; + +class CacheService { + buildKey(tenantId, category, key) { + return `tenant:${tenantId || 'global'}:${category}:${key}`; + } + + async get(tenantId, category, key) { + if (!isRedisReady()) return null; + try { + const fullKey = this.buildKey(tenantId, category, key); + const raw = await getRedisClient().get(fullKey); + return raw ? JSON.parse(raw) : null; + } catch (err) { + return null; + } + } + + async set(tenantId, category, key, value, ttlSeconds = 300) { + if (!isRedisReady()) return false; + try { + const fullKey = this.buildKey(tenantId, category, key); + const payload = JSON.stringify(value); + if (ttlSeconds > 0) { + await getRedisClient().set(fullKey, payload, 'EX', ttlSeconds); + } else { + await getRedisClient().set(fullKey, payload); + } + return true; + } catch (err) { + return false; + } + } + + async delete(tenantId, category, key) { + if (!isRedisReady()) return false; + try { + const fullKey = this.buildKey(tenantId, category, key); + await getRedisClient().del(fullKey); + return true; + } catch (err) { + return false; + } + } + + async deleteMany(keys = []) { + if (!isRedisReady() || keys.length === 0) return false; + try { + await getRedisClient().del(...keys); + return true; + } catch (err) { + return false; + } + } +} + +export const cacheService = new CacheService(); diff --git a/src/shared/infrastructure/cache/redis.client.js b/src/shared/infrastructure/cache/redis.client.js new file mode 100644 index 0000000..ceedec0 --- /dev/null +++ b/src/shared/infrastructure/cache/redis.client.js @@ -0,0 +1,54 @@ +import Redis from 'ioredis'; +import winston from 'winston'; + +const logger = winston.createLogger({ + level: 'info', + format: winston.format.combine(winston.format.timestamp(), winston.format.json()), + transports: [new winston.transports.Console()] +}); + +let redisClient = null; +let isConnected = false; + +export const getRedisClient = () => { + if (redisClient) return redisClient; + + const redisHost = process.env.REDIS_HOST || '127.0.0.1'; + const redisPort = parseInt(process.env.REDIS_PORT || '6379', 10); + const redisPassword = process.env.REDIS_PASSWORD || undefined; + + try { + redisClient = new Redis({ + host: redisHost, + port: redisPort, + password: redisPassword, + lazyConnect: true, + maxRetriesPerRequest: null, + enableReadyCheck: true, + retryStrategy: (times) => { + const delay = Math.min(times * 100, 3000); + return delay; + } + }); + + redisClient.on('connect', () => { + isConnected = true; + logger.info('Redis client connected successfully'); + }); + + redisClient.on('error', (err) => { + isConnected = false; + logger.warn(`Redis connection error: ${err.message}`); + }); + + redisClient.connect().catch((err) => { + logger.warn(`Redis failed to connect initially: ${err.message}`); + }); + } catch (err) { + logger.warn(`Failed to initialize Redis client: ${err.message}`); + } + + return redisClient; +}; + +export const isRedisReady = () => isConnected; diff --git a/src/shared/infrastructure/http/circuit-breaker.js b/src/shared/infrastructure/http/circuit-breaker.js new file mode 100644 index 0000000..69681cc --- /dev/null +++ b/src/shared/infrastructure/http/circuit-breaker.js @@ -0,0 +1,32 @@ +class CircuitBreaker { + constructor(threshold = 5, cooldownMs = 60000) { + this.threshold = threshold; + this.cooldownMs = cooldownMs; + this.states = new Map(); // key -> { failures, state, nextAttempt } + } + + getCircuitState(key) { + const current = this.states.get(key) || { failures: 0, state: 'CLOSED', nextAttempt: 0 }; + if (current.state === 'OPEN' && Date.now() >= current.nextAttempt) { + current.state = 'HALF_OPEN'; + this.states.set(key, current); + } + return current; + } + + recordSuccess(key) { + this.states.delete(key); + } + + recordFailure(key) { + const current = this.getCircuitState(key); + current.failures += 1; + if (current.failures >= this.threshold) { + current.state = 'OPEN'; + current.nextAttempt = Date.now() + this.cooldownMs; + } + this.states.set(key, current); + } +} + +export const circuitBreaker = new CircuitBreaker(); diff --git a/src/shared/infrastructure/http/http.client.js b/src/shared/infrastructure/http/http.client.js new file mode 100644 index 0000000..cd59d6a --- /dev/null +++ b/src/shared/infrastructure/http/http.client.js @@ -0,0 +1,64 @@ +import axios from 'axios'; +import { calculateBackoff, isRetryableError } from './retry.policy.js'; +import { circuitBreaker } from './circuit-breaker.js'; +import { sanitizeHeaders, sanitizePayload } from './sanitization.js'; + +export const createHttpClient = (config = {}) => { + const instance = axios.create({ + baseURL: config.baseURL, + timeout: config.timeout || 30000, + headers: config.headers || {} + }); + + const requestKey = config.baseURL || 'default_http'; + + const executeRequest = async (requestFn, maxAttempts = 3) => { + const breaker = circuitBreaker.getCircuitState(requestKey); + if (breaker.state === 'OPEN') { + throw new Error(`Circuit breaker is OPEN for target ${requestKey}. Cooldown active.`); + } + + let attempt = 0; + while (attempt < maxAttempts) { + attempt += 1; + const startTime = Date.now(); + try { + const response = await requestFn(instance); + circuitBreaker.recordSuccess(requestKey); + return { + status: response.status, + data: response.data, + headers: response.headers, + durationMs: Date.now() - startTime, + attempt + }; + } catch (error) { + const durationMs = Date.now() - startTime; + const retryable = isRetryableError(error); + + if (attempt >= maxAttempts || !retryable) { + circuitBreaker.recordFailure(requestKey); + throw { + status: error.response?.status || 500, + message: error.message, + providerData: error.response?.data, + durationMs, + attempt, + retryable, + sanitizedHeaders: sanitizeHeaders(error.config?.headers) + }; + } + + const waitMs = calculateBackoff(attempt); + await new Promise((res) => setTimeout(res, waitMs)); + } + } + }; + + return { + get: (url, opts) => executeRequest((client) => client.get(url, opts)), + post: (url, data, opts) => executeRequest((client) => client.post(url, data, opts)), + put: (url, data, opts) => executeRequest((client) => client.put(url, data, opts)), + delete: (url, opts) => executeRequest((client) => client.delete(url, opts)) + }; +}; diff --git a/src/shared/infrastructure/http/rate-limiter.js b/src/shared/infrastructure/http/rate-limiter.js new file mode 100644 index 0000000..3d33d72 --- /dev/null +++ b/src/shared/infrastructure/http/rate-limiter.js @@ -0,0 +1,19 @@ +export class RateLimiter { + constructor(requestsPerSecond = 2) { + this.requestsPerSecond = requestsPerSecond; + this.queue = []; + this.lastRequestTime = 0; + } + + async throttle() { + const minInterval = 1000 / this.requestsPerSecond; + const now = Date.now(); + const elapsed = now - this.lastRequestTime; + + if (elapsed < minInterval) { + const waitMs = minInterval - elapsed; + await new Promise((resolve) => setTimeout(resolve, waitMs)); + } + this.lastRequestTime = Date.now(); + } +} diff --git a/src/shared/infrastructure/http/retry.policy.js b/src/shared/infrastructure/http/retry.policy.js new file mode 100644 index 0000000..58c307e --- /dev/null +++ b/src/shared/infrastructure/http/retry.policy.js @@ -0,0 +1,14 @@ +export const calculateBackoff = (attempt, baseMs = 500, maxMs = 30000) => { + const exponential = Math.min(maxMs, baseMs * Math.pow(2, attempt - 1)); + const jitter = Math.random() * (exponential * 0.2); // 20% jitter + return Math.round(exponential + jitter); +}; + +export const isRetryableError = (error) => { + if (!error) return false; + const status = error.response?.status; + if (!status) return true; // Network errors, timeouts + if (status === 429) return true; // Rate limiting + if (status >= 500 && status < 600) return true; // Internal server errors + return false; +}; diff --git a/src/shared/infrastructure/http/sanitization.js b/src/shared/infrastructure/http/sanitization.js new file mode 100644 index 0000000..4c25408 --- /dev/null +++ b/src/shared/infrastructure/http/sanitization.js @@ -0,0 +1,31 @@ +export const sanitizeHeaders = (headers = {}) => { + const sanitized = { ...headers }; + const sensitiveKeys = ['authorization', 'x-shopify-access-token', 'api-key', 'secret']; + + Object.keys(sanitized).forEach(key => { + if (sensitiveKeys.includes(key.toLowerCase())) { + sanitized[key] = '[REDACTED]'; + } + }); + return sanitized; +}; + +export const sanitizePayload = (data) => { + if (!data || typeof data !== 'object') return data; + const clone = JSON.parse(JSON.stringify(data)); + const sensitiveFields = ['password', 'secret', 'token', 'accessToken', 'client_secret']; + + const redactRecursive = (obj) => { + if (!obj || typeof obj !== 'object') return; + Object.keys(obj).forEach(k => { + if (sensitiveFields.includes(k)) { + obj[k] = '[REDACTED]'; + } else if (typeof obj[k] === 'object') { + redactRecursive(obj[k]); + } + }); + }; + + redactRecursive(clone); + return clone; +}; diff --git a/src/shared/infrastructure/outbox/outbox.service.js b/src/shared/infrastructure/outbox/outbox.service.js new file mode 100644 index 0000000..13f3b6a --- /dev/null +++ b/src/shared/infrastructure/outbox/outbox.service.js @@ -0,0 +1,27 @@ +import crypto from 'crypto'; +import { models } from '../../database/models.js'; + +export const outboxService = { + async recordEvent(tenantId, aggregateType, aggregateId, eventType, payload, transaction = null) { + const payloadStr = JSON.stringify(payload); + const idempotencyKey = crypto + .createHash('sha256') + .update(`${tenantId}:${aggregateType}:${aggregateId}:${eventType}:${payloadStr}`) + .digest('hex'); + + return await models.OutboxEvent.create( + { + tenant_id: tenantId, + aggregate_type: aggregateType || 'PRODUCT', + aggregate_id: aggregateId, + event_type: eventType, + payload: payload, + status: 'pending', + event_version: payload.version || 1, + idempotency_key: idempotencyKey, + scheduled_at: new Date() + }, + transaction ? { transaction } : {} + ); + } +}; diff --git a/src/shared/infrastructure/queue/queue.constants.js b/src/shared/infrastructure/queue/queue.constants.js new file mode 100644 index 0000000..05e89f4 --- /dev/null +++ b/src/shared/infrastructure/queue/queue.constants.js @@ -0,0 +1,6 @@ +export const QUEUE_NAMES = { + INTEGRATION_SYNC: 'integration-sync', + OUTBOX_DISPATCH: 'outbox-dispatch', + WEBHOOK_DISPATCH: 'webhook-dispatch', + RETRY_PROCESSING: 'retry-processing' +}; diff --git a/src/shared/infrastructure/queue/queue.service.js b/src/shared/infrastructure/queue/queue.service.js new file mode 100644 index 0000000..9fd8178 --- /dev/null +++ b/src/shared/infrastructure/queue/queue.service.js @@ -0,0 +1,106 @@ +import { Queue, Worker } from 'bullmq'; +import { getRedisClient, isRedisReady } from '../cache/redis.client.js'; +import { QUEUE_NAMES } from './queue.constants.js'; +import winston from 'winston'; + +const logger = winston.createLogger({ + level: 'info', + format: winston.format.combine(winston.format.timestamp(), winston.format.json()), + transports: [new winston.transports.Console()] +}); + +class QueueService { + constructor() { + this.queues = new Map(); + this.workers = new Map(); + this.processors = new Map(); + } + + getQueue(queueName) { + if (this.queues.has(queueName)) { + return this.queues.get(queueName); + } + + if (!isRedisReady()) { + return null; + } + + try { + const redisConnection = getRedisClient(); + const queue = new Queue(queueName, { + connection: redisConnection, + defaultJobOptions: { + attempts: 3, + backoff: { + type: 'exponential', + delay: 2000 + }, + removeOnComplete: 100, + removeOnFail: 500 + } + }); + + this.queues.set(queueName, queue); + return queue; + } catch (err) { + logger.error(`Failed to initialize queue ${queueName}: ${err.message}`); + return null; + } + } + + async addJob(queueName, jobName, payload, options = {}) { + const queue = this.getQueue(queueName); + if (!queue) { + // In-memory fallback execution if Redis is offline + const processor = this.processors.get(queueName); + if (processor) { + logger.info(`[QueueFallback] Executing job "${jobName}" in-memory on ${queueName} (Redis offline)`); + setImmediate(async () => { + try { + await processor({ id: `in_mem_${Date.now()}`, data: payload }); + } catch (err) { + logger.error(`[QueueFallback] Execution error on "${jobName}": ${err.message}`); + } + }); + return { id: `in_mem_${Date.now()}` }; + } + logger.warn(`Skipping job enqueuing for ${queueName} as queue and processor are unavailable`); + return null; + } + return await queue.add(jobName, payload, options); + } + + registerWorker(queueName, processor, concurrency = 5) { + // Always store processor for in-memory fallback + this.processors.set(queueName, processor); + + if (!isRedisReady()) { + logger.info(`Redis offline — registered in-memory fallback worker for ${queueName}`); + return null; + } + + try { + const redisConnection = getRedisClient(); + const worker = new Worker(queueName, processor, { + connection: redisConnection, + concurrency + }); + + worker.on('completed', (job) => { + logger.info(`Job ${job.id} completed on queue ${queueName}`); + }); + + worker.on('failed', (job, err) => { + logger.error(`Job ${job?.id} failed on queue ${queueName}: ${err.message}`); + }); + + this.workers.set(queueName, worker); + return worker; + } catch (err) { + logger.error(`Failed to register worker for ${queueName}: ${err.message}`); + return null; + } + } +} + +export const queueService = new QueueService(); diff --git a/src/shared/infrastructure/secrets/secret.service.js b/src/shared/infrastructure/secrets/secret.service.js new file mode 100644 index 0000000..f3991dc --- /dev/null +++ b/src/shared/infrastructure/secrets/secret.service.js @@ -0,0 +1,47 @@ +import crypto from 'crypto'; + +const ALGORITHM = 'aes-256-gcm'; +const IV_LENGTH = 12; // 96 bits for GCM +const DEFAULT_KEY = process.env.ENCRYPTION_SECRET || 'antigravity_default_secret_key_32_bytes_len!!'; + +const getMasterKey = () => { + return crypto.createHash('sha256').update(DEFAULT_KEY).digest(); +}; + +export const secretService = { + encrypt(plaintext) { + if (!plaintext) return ''; + const iv = crypto.randomBytes(IV_LENGTH); + const cipher = crypto.createCipheriv(ALGORITHM, getMasterKey(), iv); + + let encrypted = cipher.update(plaintext, 'utf8', 'hex'); + encrypted += cipher.final('hex'); + + const authTag = cipher.getAuthTag().toString('hex'); + return `${iv.toString('hex')}:${authTag}:${encrypted}`; + }, + + decrypt(ciphertext) { + if (!ciphertext) return ''; + try { + const parts = ciphertext.split(':'); + if (parts.length !== 3) { + throw new Error('Invalid ciphertext format'); + } + + const [ivHex, authTagHex, encryptedText] = parts; + const iv = Buffer.from(ivHex, 'hex'); + const authTag = Buffer.from(authTagHex, 'hex'); + + const decipher = crypto.createDecipheriv(ALGORITHM, getMasterKey(), iv); + decipher.setAuthTag(authTag); + + let decrypted = decipher.update(encryptedText, 'hex', 'utf8'); + decrypted += decipher.final('utf8'); + + return decrypted; + } catch (err) { + throw new Error(`Failed to decrypt credentials: ${err.message}`); + } + } +}; diff --git a/src/test_integrations_pipeline.js b/src/test_integrations_pipeline.js new file mode 100644 index 0000000..205776c --- /dev/null +++ b/src/test_integrations_pipeline.js @@ -0,0 +1,48 @@ +import { secretService } from './shared/infrastructure/secrets/secret.service.js'; +import { shopifyMapper } from './features/integrations/adapters/shopify/shopify.mapper.js'; +import { ShopifyAdapter } from './features/integrations/adapters/shopify/shopify.adapter.js'; + +console.log('=== 1. Testing AES-256-GCM Secret Service ==='); +const sampleToken = 'shpat_1234567890abcdef1234567890abcdef'; +const encrypted = secretService.encrypt(sampleToken); +console.log('Encrypted Token:', encrypted); +const decrypted = secretService.decrypt(encrypted); +console.log('Decrypted Token:', decrypted); +if (decrypted === sampleToken) { + console.log('✅ Secret Service Encryption/Decryption PASSED!'); +} else { + console.error('❌ Secret Service Encryption Failed'); +} + +console.log('\n=== 2. Testing Shopify Mapper ==='); +const mockCanonicalProduct = { + identity: { id: '3a25f11e-5539-4000-9211-8c9f5fe37d99', code: 'HM-FULL-SLEEVES' }, + content: { name: 'H&M Full Sleeves T-shirt', description: 'High quality cotton shirt', status: 'active' }, + taxonomy: { brand: { name: 'H&M' }, category: { name: 'Apparel' } }, + attributes: { color: 'Blue', size: 'M' }, + variants: [ + { id: 'v1', sku: 'HM-BLUE-M', price: 29.99, costPrice: 15.00 } + ], + version: 1 +}; + +const shopifyInput = shopifyMapper.toProductInput(mockCanonicalProduct); +console.log('Mapped Shopify Product Input:', JSON.stringify(shopifyInput, null, 2)); + +if (shopifyInput.title === 'H&M Full Sleeves T-shirt' && shopifyInput.status === 'ACTIVE') { + console.log('✅ Shopify Mapper PASSED!'); +} else { + console.error('❌ Shopify Mapper Failed'); +} + +console.log('\n=== 3. Testing Shopify Adapter Instantiation ==='); +const adapter = new ShopifyAdapter({ + shop_domain: 'technova1.myshopify.com', + access_token: sampleToken +}); + +if (adapter && adapter.client) { + console.log('✅ Shopify Adapter Instantiation PASSED!'); +} else { + console.error('❌ Shopify Adapter Instantiation Failed'); +}