feat: onboard plugin integration service #2832
Conversation
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## develop #2832 +/- ##
===========================================
+ Coverage 87.18% 87.20% +0.02%
===========================================
Files 529 531 +2
Lines 28659 28824 +165
Branches 6824 6837 +13
===========================================
+ Hits 24985 25135 +150
- Misses 3340 3355 +15
Partials 334 334 ☔ View full report in Codecov by Sentry. |
|
Important Auto Review SkippedAuto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the To trigger a single review, invoke the WalkthroughThe codebase has been updated with a focus on enhancing plugin integration capabilities. A new Changes
Assessment against linked issues
Poem
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (invoked as PR comments)
Additionally, you can add CodeRabbit Configration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Review Status
Actionable comments generated: 7
Configuration used: CodeRabbit UI
Files ignored due to filter (4)
- package-lock.json
- package.json
- src/cdk/v2/destinations/webhook/rtWorkflow.yaml
- test/tests/data/webhook_router_output.json
Files selected for processing (12)
- src/helpers/pluginAdapter.ts (1 hunks)
- src/helpers/serviceSelector.ts (4 hunks)
- src/routes/utils/constants.js (1 hunks)
- src/services/comparator.ts (2 hunks)
- src/services/destination/pluginIntegration.ts (1 hunks)
- src/services/destination/postTransformation.ts (2 hunks)
- src/services/misc.ts (1 hunks)
- src/services/userTransform.ts (1 hunks)
- src/types/index.ts (7 hunks)
- test/controllerUtility/ctrl-utility.test.ts (22 hunks)
- test/integrations/component.test.ts (5 hunks)
- test/integrations/destinations/webhook/router/data.ts (5 hunks)
Additional comments: 31
src/helpers/pluginAdapter.ts (1)
- 26-251: Overall, the
PluginAdapterclass is well-structured and follows good practices for asynchronous operations, error handling, and data transformation. The methodstransformAtProcessorandtransformAtRouterare designed to handle plugin integrations effectively. Ensure that the TODO regarding dynamic integration config is addressed in the future.src/helpers/serviceSelector.ts (5)
8-11: The import of
PluginIntegrationServiceand the addition ofPLUGIN_DESTin theServiceSelectorclass align with the summary and the pull request's intent to introduce a new plugin integration service.33-35: The new method
isPluginDestinationcorrectly checks for the plugin destination using thedestinationDefinitionConfigobject. Ensure that theisPluginproperty is always present in thedestinationDefinitionConfigobject for plugin destinations.78-80: The logic to handle plugin destinations in
getPrimaryDestinationServiceis correctly implemented, ensuring that thePluginIntegrationServiceis returned when the destination is a plugin.74-85: The
getPrimaryDestinationServicemethod has been updated to handle the new plugin destination type. Ensure that the rest of the service selection logic remains compatible with this change.74-85: The
getSourceServicemethod is not implemented and currently contains a placeholder comment. Verify if this is intentional and if there are plans to implement this method in the future.src/routes/utils/constants.js (2)
9-9: The addition of
PLUGIN_DESTto theINTEGRATION_SERVICEobject aligns with the pull request's intent to introduce a new plugin integration framework.10-13: The correction of the syntax error in the
CHANNELSobject by adding the missing closing brace ensures the integrity of the object definition.src/services/comparator.ts (1)
- 2-2: The import of
Destinationis correctly added and used in thegetTestThresholdfunction.src/services/destination/pluginIntegration.ts (3)
20-20: Verify if the
initmethod is intentionally left empty or if initialization logic is missing.115-138: Ensure that the placeholder methods
doBatchTransformation,deliver, andprocessUserDeletionare implemented before the service goes into production.122-122: Verify the error messages in
TransformationErrorfor consistency and correct references. 'CDKV1' and 'CDV1' might be typos or incorrect version references.Also applies to: 130-130, 137-137
src/services/destination/postTransformation.ts (1)
- 73-80: The logic to convert
userIdto a string for both single and array instances ofbatchedRequestis correctly implemented.src/services/misc.ts (1)
- 5-5: The import of
Metadatafrom@rudderstack/integrations-libshould be verified to ensure that the properties accessed ingetMetaTagsare present and compatible with the newMetadatatype.src/types/index.ts (5)
1-4: The changes to the import statements align with the summary provided, and the renaming of
MetadatatoTransformedOutputis reflected in the code.20-26: The usage of
Metadatatype inProcessorTransformationOutputandMessageIdMetadataMapis consistent with the import changes and renaming.61-74: The optional fields
errorandstatTagsinProcessorTransformationResponseandRouterTransformationResponsetypes are correctly updated as per the summary.82-87: The
SourceTransformationResponsetype has been correctly updated to makeoutputToSourceandstatTagsfields optional.228-229: The export of the
TransformedOutputtype is present and correct as per the summary.test/controllerUtility/ctrl-utility.test.ts (1)
- 71-74: The addition of the
IsProcessorEnabledproperty to thedestinationobject in the test cases is consistent with the changes described in the summary. Ensure that the corresponding code that processes these test cases is updated to handle this new property correctly.Also applies to: 136-139, 205-208, 270-273, 339-342, 404-407, 475-478, 538-541, 601-604, 664-667, 731-734, 794-797, 857-860, 920-923, 992-995, 1055-1058, 1121-1124, 1187-1190, 1255-1258, 1318-1321, 1384-1387, 1450-1453
test/integrations/component.test.ts (5)
24-25: The import
setfrom@rudderstack/integrations-libis not directly used in the provided code. If it's not used elsewhere in the file, consider removing it to keep the code clean.55-55: The
pluginDestinationsarray is used to determine if a destination is a plugin. Ensure that this array is kept up-to-date as new plugin destinations are added.84-90: The
registerAxiosMocksfunction has been correctly updated to accept an additionalmockAxiosAdapterparameter and is used to register mocks for bothaxiosinstances.108-122: The
updateTCDataForPluginDestsfunction is correctly implemented to update the test case data for plugin destinations. Ensure that the path used in thesetfunction matches the structure of the event objects.160-164: The
destinationTestHandlerfunction correctly checks if the destination is a plugin and updates the test case data accordingly. This is a good use of conditional logic to handle different types of destinations.test/integrations/destinations/webhook/router/data.ts (6)
107-107: The
webhookUrlhas been changed to use HTTP. Verify that this change does not introduce security vulnerabilities, especially if sensitive data is being transmitted.117-117: The
headersarray includes additional key-value pairs. Ensure that these new headers are expected and properly documented.254-254: The
batchedRequestfield has been transformed from an object to an array of objects. Confirm that all consuming code has been updated to handle this new structure.369-369: The
batchedfield has been updated totrue. This change should be consistent with the updatedbatchedRequeststructure.370-370: The
statusCodefield has been updated to200. Ensure that this status code is appropriate for all responses expected from the webhook.120-122: A new
DestinationDefinitionobject with acdkV2Enabledproperty set totruehas been added. Verify that this new configuration is supported and properly handled in the destination's implementation.
| @@ -47,7 +47,7 @@ export class UserTransformService { | |||
| } | |||
| const responses = await Promise.all<FixMe>( | |||
| Object.entries(groupedEvents).map(async ([, destEvents]) => { | |||
| const eventsToProcess = destEvents as ProcessorTransformationRequest[]; | |||
| const eventsToProcess = destEvents as any[]; | |||
There was a problem hiding this comment.
The type assertion for eventsToProcess has been changed to any[], which removes the type safety provided by ProcessorTransformationRequest[]. Consider reverting this change or implementing a type guard to ensure the elements of eventsToProcess conform to the expected structure at runtime.
- const eventsToProcess = destEvents as any[];
+ const eventsToProcess = destEvents as ProcessorTransformationRequest[];Committable suggestion
❗ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| const eventsToProcess = destEvents as any[]; | |
| const eventsToProcess = destEvents as ProcessorTransformationRequest[]; |
There was a problem hiding this comment.
| const resultantPayloads: RouterTransformationResponse[] = transformedPayloads; | ||
| resultantPayloads.forEach((resultantPayload) => { | ||
| if (resultantPayload.batchedRequest && resultantPayload.batchedRequest.userId) { | ||
| if (Array.isArray(resultantPayload.batchedRequest)) { | ||
| resultantPayload.batchedRequest.forEach((batchedRequest) => { | ||
| if (batchedRequest.userId) { | ||
| batchedRequest.userId = `${batchedRequest.userId}`; | ||
| } | ||
| }); | ||
| } else if (resultantPayload.batchedRequest && resultantPayload.batchedRequest.userId) { | ||
| resultantPayload.batchedRequest.userId = `${resultantPayload.batchedRequest.userId}`; | ||
| } |
There was a problem hiding this comment.
The change from using cloneDeep to direct assignment in handleRouterTransformSuccessEvents could lead to potential mutability issues if the original transformedPayloads object is expected to remain unchanged. Verify if this change in behavior is intentional and ensure that no side effects will occur due to this modification.
src/services/comparator.ts
Outdated
| const config = destination.DestinationDefinition?.Config as { | ||
| camparisonTestThreshold?: number; | ||
| }; | ||
| return config?.camparisonTestThreshold || 0; |
There was a problem hiding this comment.
There is a typo in the property name camparisonTestThreshold. It should be corrected to comparisonTestThreshold.
- camparisonTestThreshold?: number;
+ comparisonTestThreshold?: number;Committable suggestion
❗ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| const config = destination.DestinationDefinition?.Config as { | |
| camparisonTestThreshold?: number; | |
| }; | |
| return config?.camparisonTestThreshold || 0; | |
| const config = destination.DestinationDefinition?.Config as { | |
| comparisonTestThreshold?: number; | |
| }; | |
| return config?.comparisonTestThreshold || 0; |
| public async doProcessorTransformation( | ||
| events: ProcessorTransformationRequest[], | ||
| destinationType: string, | ||
| _version: string, | ||
| _requestMetadata: unknown, | ||
| ): Promise<ProcessorTransformationResponse[]> { | ||
| const results = await PluginAdapter.transformAtProcessor(events, destinationType); | ||
| const respList: ProcessorTransformationResponse[] = []; | ||
| results.allSuccessList.forEach((successResponse) => { | ||
| respList.push({ | ||
| output: successResponse.payload, | ||
| metadata: successResponse.metadata, | ||
| statusCode: 200, | ||
| }); | ||
| }); | ||
|
|
||
| results.allErrorList.forEach( | ||
| (errorResponse: { | ||
| metadata: Metadata; | ||
| response: { | ||
| // need to build type def here | ||
| message: string; | ||
| status: number; | ||
| destinationResponse?: any; | ||
| statTags?: { [x: string]: string }; | ||
| }; | ||
| }) => { | ||
| respList.push({ | ||
| metadata: errorResponse.metadata, | ||
| statusCode: errorResponse.response.status, | ||
| error: errorResponse.response.message, | ||
| statTags: errorResponse.response.statTags, | ||
| }); | ||
| }, | ||
| ); | ||
| return respList; | ||
| } |
There was a problem hiding this comment.
Consider refactoring the doProcessorTransformation and doRouterTransformation methods to reduce code duplication, if the logic remains similar.
src/helpers/pluginAdapter.ts
Outdated
| // TODO: default integration config need to make it dynamic by making some sort of config call or get it from config file | ||
| // const integrationConfig: IntegrationConfig = { | ||
| // name: integrationName, | ||
| // saveResponse: true, | ||
| // eventOrdering: true, | ||
| // plugins: ['preprocessor', 'multiplexer'], | ||
| // }; |
There was a problem hiding this comment.
There is a TODO comment regarding making the integration config dynamic. This should be prioritized to ensure that the integration configuration can be retrieved from a configuration file or service, rather than being hardcoded or left as a placeholder.
…udder-transformer into feat.onborad-plugin-service
|
can you split this into multiple PRs for easy review |
|
|
This PR is considered to be stale. It has been open for 20 days with no further activity thus it is going to be closed in 7 days. To avoid such a case please consider removing the stale label manually or add a comment to the PR. |
|
This PR is considered to be stale. It has been open for 20 days with no further activity thus it is going to be closed in 7 days. To avoid such a case please consider removing the stale label manually or add a comment to the PR. |
|
This PR is considered to be stale. It has been open for 20 days with no further activity thus it is going to be closed in 7 days. To avoid such a case please consider removing the stale label manually or add a comment to the PR. |
Onboarding Plugin Integration Service
This PR introduces onboarding of plugin integration framework to transformer, we take the approach of onboarding a new service aligning to the predefined interface to dictate the service structure
Major Piece of event order orchestration for transformer <> server contract lie in
pluginAdapter.tswhich after passing the events to integration plugin and handles the consistent structuring for responding back to outer layerThe
pluginAdapteruses integration-store to get integration-plugin to do either processor/router transformerFurther we updated type definitions referencing from
integrations-libwhich acts as a common libraryResolves
INT-837
Type of change
Related issues
Checklists
Development
Code review
Summary by CodeRabbit
New Features
PluginAdapterfor data transformation at processor and router levels.PluginIntegrationServiceto handle plugin destinations.PLUGIN_DESTfor integration services.Bug Fixes
CHANNELSobject definition.Documentation
Refactor
ServiceSelectorclass to support plugin destinations.getTestThresholdfunction for better property access.batchedRequestinDestinationPostTransformationService.Metadatato use the@rudderstack/integrations-libpackage.Tests
IsProcessorEnabledproperty.Chores
MetadatatoTransformedOutputin type definitions.