Merge pull request #196 from ts-amz/evt_processor

fix: keep filter from erroring when no events are found, additional c…

source/common/changes/@awssolutions/cdf-events-processor/main_2024-03-20-02-23.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@awssolutions/cdf-events-processor",
+      "comment": "keep filter from erroring when no events are found, additional cleanup",
+      "type": "none"
+    }
+  ],
+  "packageName": "@awssolutions/cdf-events-processor"
+}

source/common/config/rush/repo-state.json

@@ -1,5 +1,5 @@
 // DO NOT MODIFY THIS FILE MANUALLY BUT DO COMMIT IT. It is generated and used by Rush.
 {
-  "pnpmShrinkwrapHash": "def9633ab3e27f220d855c6f5ccdf1e7bdf6dfb9",
+  "pnpmShrinkwrapHash": "d8b7c902b74f2c02b6c151ecccf06d926715e682",
   "preferredVersionsHash": "14c05a7722342014cec64c4bef7d9bed0d0b7b7f"
 }

source/packages/services/events-processor/src/api/subscriptions/subscription.service.ts

@@ -319,13 +319,13 @@ export class SubscriptionService {
 
         ow(eventId, ow.string.nonEmpty);
 
-        const [results, pagintion] = await this.subscriptionDao.listSubscriptionsForEvent(
+        const [results, pagination] = await this.subscriptionDao.listSubscriptionsForEvent(
             eventId,
             from
         );
 
         logger.debug(`subscription.service listByEvent: exit: model: ${JSON.stringify(results)}`);
-        return [results, pagintion];
+        return [results, pagination];
     }
 
     public async create(item: SubscriptionItem): Promise<string> {

source/packages/services/events-processor/src/filter/filter.service.spec.ts

@@ -311,6 +311,61 @@ describe('FilterService', () => {
         expect(mockedGetEventConfigCall).toBeCalled();
     });
 
+    it('undefined value from listSubscriptionsForEvent should not error filter function', async () => {
+        const eventSourceId = 'ES123';
+        const principal = 'deviceId';
+        const principalValue = 'device001';
+
+        // stubs
+        const commonMessageAttributes = {
+            eventSourceId,
+            principal,
+            principalValue,
+        };
+
+        const events: CommonEvent[] = [
+            {
+                ...commonMessageAttributes,
+                attributes: {
+                    sequence: 1,
+                    batteryLevel: 25,
+                    batteryLevelThreshold: 22,
+                },
+            },
+            {
+                ...commonMessageAttributes,
+                attributes: {
+                    sequence: 2,
+                    batteryLevel: 22,
+                    batteryLevelThreshold: 22,
+                },
+            },
+        ];
+
+        // mocks
+        // mockedSubscriptionDao.listSubscriptionsForEventMessage is already mocked
+        // and will return the undefined result we need for this test
+
+        const mockedGetEventConfigCall = (mockedEventDao.getEventConfig = jest
+            .fn()
+            .mockImplementation(() => {
+                // do nothing, acting as a spy only
+            }));
+
+        const mockedCreateAlertsCall = (mockedAlertDao.create = jest
+            .fn()
+            .mockImplementationOnce((_alerts) => {
+                // do nothing, acting as a spy only
+            }));
+
+        // execute
+        await instance.filter(events);
+
+        // verify
+        expect(mockedCreateAlertsCall).toBeCalledTimes(0);
+        expect(mockedGetEventConfigCall).toBeCalledTimes(0);
+    });
+
     it('disabling alert thresholds should alert each time', async () => {
         const eventSourceId = 'ES123';
         const principal = 'deviceId';

source/packages/services/events-processor/src/filter/filter.service.ts

@@ -50,7 +50,7 @@ export class FilterService {
         for (const ev of events) {
             // perform lookup to see if any subscriptions are configured for the event source/principal/principalValue (cached for the duration of the method call)
             const subscriptions = (
-                await this.listSubscriptionsForEvent(ev, subscriptionMap)
+                (await this.listSubscriptionsForEvent(ev, subscriptionMap)) || []
             ).filter((o) => o.enabled);
 
             // if we have subscriptions, lets evaluate them against the datasource
@@ -190,7 +190,7 @@ export class FilterService {
     private async listSubscriptionsForEvent(
         ev: CommonEvent,
         subscriptionMap: { [key: string]: SubscriptionItem[] }
-    ) {
+    ): Promise<SubscriptionItem[] | undefined> {
         logger.debug(
             `filter.service listSubscriptionsForEvent: in: ev:${JSON.stringify(
                 ev

source/packages/services/events-processor/src/lambda_proxy_invoke.spec.ts

@@ -0,0 +1,68 @@
+/*********************************************************************************************************************
+ *  Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.                                           *
+ *                                                                                                                    *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file except in compliance    *
+ *  with the License. A copy of the License is located at                                                             *
+ *                                                                                                                    *
+ *      http://www.apache.org/licenses/LICENSE-2.0                                                                    *
+ *                                                                                                                    *
+ *  or in the 'license' file accompanying this file. This file is distributed on an 'AS IS' BASIS, WITHOUT WARRANTIES *
+ *  OR CONDITIONS OF ANY KIND, express or implied. See the License for the specific language governing permissions    *
+ *  and limitations under the License.                                                                                *
+ *********************************************************************************************************************/
+import 'reflect-metadata';
+
+const { handler } = require('./lambda_proxy_invoke');
+
+describe('LambdaProxyInvoke', () => {
+    beforeEach(() => {});
+
+    it('error on missing eventSourceId ', async () => {
+        let err: Error;
+        try {
+            await handler(
+                {
+                    principal: 'mockPrincipal',
+                    principalValue: 'mockPrincipalValue',
+                },
+                {}
+            );
+        } catch (e: any) {
+            err = e;
+        }
+
+        expect(err.message).toEqual('Missing eventSourceId');
+    });
+    it('error on missing eventSourceId ', async () => {
+        let err: Error;
+        try {
+            await handler(
+                {
+                    eventSourceId: 'mockEventSourceId',
+                    principalValue: 'mockPrincipalValue',
+                },
+                {}
+            );
+        } catch (e: any) {
+            err = e;
+        }
+
+        expect(err.message).toEqual('Missing principal');
+    });
+    it('error on missing eventSourceId ', async () => {
+        let err: Error;
+        try {
+            await handler(
+                {
+                    eventSourceId: 'mockEventSourceId',
+                    principal: 'mockPrincipal',
+                },
+                {}
+            );
+        } catch (e: any) {
+            err = e;
+        }
+
+        expect(err.message).toEqual('Missing principalValue');
+    });
+});