Feat (MM): State Change Event (#569)

config/stacks/metadataManager.ts

@@ -5,6 +5,7 @@ import {
   corsAllowOrigins,
   logsApiGatewayConfig,
   vpcProps,
+  eventBusName,
 } from '../constants';
 import { MetadataManagerStackProps } from '../../lib/workload/stateless/stacks/metadata-manager/deploy/stack';
 
@@ -15,6 +16,7 @@ export const getMetadataManagerStackProps = (stage: AppStage): MetadataManagerSt
     vpcProps,
     isDailySync: isDailySync,
     lambdaSecurityGroupName: computeSecurityGroupName,
+    eventBusName: eventBusName,
     apiGatewayCognitoProps: {
       ...cognitoApiGatewayConfig,
       corsAllowOrigins: corsAllowOrigins[stage],

config/stacks/schema/events.ts

@@ -40,6 +40,15 @@ export const getEventSchemaStackProps = (): SchemaStackProps => {
           docBase + '/executionservice/WorkflowRunStateChange.schema.json'
         ),
       },
+      {
+        ...defaultProps,
+        schemaName: 'orcabus.metadatamanager@MetadataStateChange',
+        schemaDescription: 'State change event for lab metadata changes',
+        schemaLocation: path.join(
+          __dirname,
+          docBase + '/metadatamanager/MetadataStateChange.schema.json'
+        ),
+      },
     ],
   };
 };

docs/schemas/events/metadatamanager/Makefile

@@ -0,0 +1,3 @@
+test:
+	@check-jsonschema --schemafile MetadataStateChange.schema.json example/MSC__example1.json
+	@check-jsonschema --schemafile MetadataStateChange.schema.json example/MSC__example2.json

docs/schemas/events/metadatamanager/MetadataStateChange.schema.json

@@ -0,0 +1,78 @@
+{
+  "$schema": "http://json-schema.org/draft-04/schema#",
+  "$id": "https://raw.githubusercontent.com/umccr/orcabus/main/docs/schemas/events/metadatamanager/MetadataStateChange.schema.json",
+  "description": "EventBridge custom event schema for orcabus.metadatamanager@MetadataStateChange",
+  "title": "AWSEvent",
+  "type": "object",
+  "required": [
+    "detail-type",
+    "detail",
+    "source"
+  ],
+  "properties": {
+    "id": {
+      "type": "string"
+    },
+    "region": {
+      "type": "string"
+    },
+    "resources": {
+      "type": "array",
+      "items": {
+        "type": "string"
+      }
+    },
+    "source": {
+      "enum": ["orcabus.metadatamanager"]
+    },
+    "time": {
+      "type": "string",
+      "format": "date-time"
+    },
+    "version": {
+      "type": "string"
+    },
+    "account": {
+      "type": "string"
+    },
+    "detail-type": {
+      "enum": ["MetadataStateChange"]
+    },
+    "detail": {
+      "$ref": "#/definitions/MetadataStateChange"
+    }
+  },
+  "definitions": {
+    "MetadataStateChange": {
+      "type": "object",
+      "required": [
+        "model",
+        "action",
+        "data",
+        "refId"
+      ],
+      "properties": {
+        "model": {
+          "type": "string",
+          "enum": [
+            "LIBRARY"
+          ]
+        },
+        "action": {
+          "type": "string",
+          "enum": [
+            "CREATE",
+            "UPDATE",
+            "DELETE"
+          ]
+        },
+        "refId": {
+          "type": "string"
+        },
+        "data": {
+          "type": "object"
+        }
+      }
+    }
+  }
+}

docs/schemas/events/metadatamanager/example/MSC__example1.json

@@ -0,0 +1,27 @@
+{
+  "version": "0",
+  "id": "f71bbbbb-5b36-40c2-f7dc-804ca6270cd6",
+  "detail-type": "MetadataStateChange",
+  "source": "orcabus.metadatamanager",
+  "account": "123456789012",
+  "time": "2024-05-01T09:25:44Z",
+  "region": "ap-southeast-2",
+  "resources": [],
+  "detail": {
+    "action": "CREATE",
+    "model": "LIBRARY",
+    "refId": "lib.01J8GMF3XCHW9CV8ZFS8F1P1RF",
+    "data": {
+      "orcabusId": "lib.01J8GMF3XCHW9CV8ZFS8F1P1RF",
+      "libraryId": "L10001",
+      "phenotype": "normal",
+      "workflow": "research",
+      "quality": "good",
+      "type": "WTS",
+      "assay": "ctTSO",
+      "coverage": 120.0,
+      "sample": "smp.01J8GMF3WD6TD5Y491EEBARYBE",
+      "subject": "sbj.01J8GMF3VZRGYQG1GYDJC6E9MV"
+    }
+  }
+}

docs/schemas/events/metadatamanager/example/MSC__example2.json

@@ -0,0 +1,27 @@
+{
+  "version": "0",
+  "id": "f71bbbbb-5b36-40c2-f7dc-804ca6270cd6",
+  "detail-type": "MetadataStateChange",
+  "source": "orcabus.metadatamanager",
+  "account": "123456789012",
+  "time": "2024-05-01T09:25:44Z",
+  "region": "ap-southeast-2",
+  "resources": [],
+  "detail": {
+    "action": "DELETE",
+    "model": "LIBRARY",
+    "refId": "lib.01J8GNBB7YK8RCVFGXSBV4ZKST",
+    "data": {
+      "orcabusId": "lib.01J8GNBB7YK8RCVFGXSBV4ZKST",
+      "libraryId": "L10001",
+      "phenotype": "normal",
+      "workflow": "research",
+      "quality": "poor",
+      "type": "WTS",
+      "assay": "ctTSO",
+      "coverage": 120.0,
+      "sample": "smp.01J8GNBB67HGG3G17QA5QP98TE",
+      "subject": "sbj.01J8GNBB59GR0KMTNR0BT3F31V"
+    }
+  }
+}

lib/workload/stateless/stacks/metadata-manager/Makefile

@@ -76,7 +76,7 @@ insert-data:
 	@python manage.py insert_mock_data
 
 suite:
-	@python manage.py test
+	@python manage.py test --parallel
 
 # full mock suite test pipeline - install deps, bring up compose stack, run suite, bring down compose stack
 test: install up suite down

lib/workload/stateless/stacks/metadata-manager/README.md

@@ -61,6 +61,43 @@ on the model of the record.
 
 ## How things work
 
+The metadata loader currently supports two Lambda functions: one for syncing from a tracking sheet and another for syncing from a custom CSV presigned URL. Upon a CREATE or UPDATE operation in the metadata library, the Lambda function will publish events to the `MainOrcabusEventBus` using the schema defined in [MetadataStateChange.schema.json](/docs/schemas/events/metadatamanager/MetadataStateChange.schema.json).
+
+The event data will adhere to the same schema as the OpenAPI schema without nested object.
+
+Example of the event emitted.
+
+```json
+{
+  "version": "0",
+  "id": "e7b8a2d4-3b6e-4f9b-9c1e-1a2b3c4d5e6f",
+  "detail-type": "MetadataStateChange",
+  "source": "orcabus.metadatamanager",
+  "account": "12345678",
+  "time": "2000-09-01T00:00:00Z",
+  "region": "ap-southeast-2",
+  "resources": [],
+  "detail": {
+    "action": "CREATE",
+    "model": "LIBRARY",
+    "refId": "lib.01J8GMF3XCHW9CV8ZFS8F1P1RF",
+    "data": {
+      "orcabusId": "lib.01J8GMF3XCHW9CV8ZFS8F1P1RF",
+      "libraryId": "L10001",
+      "phenotype": "normal",
+      "workflow": "research",
+      "quality": "good",
+      "type": "WTS",
+      "assay": "ctTSO",
+      "coverage": 120.0,
+      "sample": "smp.01J8GMF3WD6TD5Y491EEBARYBE",
+      "subject": "sbj.01J8GMF3VZRGYQG1GYDJC6E9MV"
+    }
+  }
+}
+
+```
+
 ### How Syncing The Data Works
 
 In the near future, we might introduce different ways to load data into the application. For the time being, we are

lib/workload/stateless/stacks/metadata-manager/app/serializers/library.py

@@ -1,4 +1,4 @@
-from app.models import Library
+from app.models import Library, Sample, Subject
 from .base import SerializersBase
 from .project import ProjectSerializer
 from .sample import SampleSerializer
@@ -14,6 +14,13 @@ class Meta:
         model = Library
         exclude = ["project_set"]
 
+    def to_representation(self, instance):
+        representation = super().to_representation(instance)
+        representation['sample'] = Sample.orcabus_id_prefix + representation['sample']
+        representation['subject'] = Subject.orcabus_id_prefix + representation['subject']
+        return representation
+
+
 
 class LibraryDetailSerializer(LibraryBaseSerializer):
     project_set = ProjectSerializer(many=True, read_only=True)

lib/workload/stateless/stacks/metadata-manager/app/serializers/utils.py

@@ -0,0 +1,17 @@
+
+
+def to_camel_case_key_dict(data: dict) -> dict:
+    """
+    Convert dictionary keys from snake_case to camelCase.
+    """
+    def snake_to_camel(word):
+        components = word.split('_')
+        # We capitalize the first letter of each component except the first one
+        # with the 'title' method and join them together.
+        return components[0] + ''.join(x.title() for x in components[1:])
+
+    new_data = {}
+    for key, value in data.items():
+        new_key = snake_to_camel(key)
+        new_data[new_key] = value
+    return new_data

lib/workload/stateless/stacks/metadata-manager/app/tests/test_models.py

@@ -1,5 +1,6 @@
 import logging
 
+from unittest.mock import MagicMock, patch
 from django.test import TestCase
 
 from app.models import Subject, Sample, Library, Contact, Project, Individual
@@ -69,3 +70,52 @@ def test_metadata_model_relationship(self):
         # find the linked contact
         cnt_one = prj_one.contact_set.get(contact_id=CONTACT_1['contact_id'])
         self.assertEqual(cnt_one.contact_id, CONTACT_1['contact_id'], "incorrect contact 'id' linked to project")
+
+    def test_upsert_method(self):
+        """
+        python manage.py test app.tests.test_models.MetadataTestCase.test_upsert_method
+        """
+
+        # Test function with updating existing record
+        updated_spc_data = {
+            "sample_id": SAMPLE_1['sample_id'],
+            "source": 'skin',
+        }
+        obj, is_created, is_updated = Sample.objects.update_or_create_if_needed(
+            {"sample_id": updated_spc_data["sample_id"]},
+            updated_spc_data
+        )
+        self.assertIsNotNone(obj, "object should not be None")
+        self.assertFalse(is_created, "object should NOT be created")
+        self.assertTrue(is_updated, "object should be updated")
+
+        smp_one = Sample.objects.get(sample_id=updated_spc_data["sample_id"])
+        self.assertEqual(smp_one.source, updated_spc_data['source'], "incorrect 'source' from updated specimen id")
+
+        # Test function with creating new record
+        new_spc_data = {
+            "sample_id": 'SMP002',
+            "source": 'RNA',
+        }
+        obj, is_created, is_updated = Sample.objects.update_or_create_if_needed(
+            {"sample_id": new_spc_data['sample_id']},
+            new_spc_data
+        )
+        self.assertIsNotNone(obj, "object should not be None")
+        self.assertTrue(is_created, "new object should be created")
+        self.assertFalse(is_updated, "new object should not be updated")
+        spc_two = Sample.objects.get(sample_id=new_spc_data['sample_id'])
+        self.assertEqual(spc_two.sample_id, new_spc_data["sample_id"], "incorrect specimen 'id'")
+        self.assertEqual(spc_two.source, new_spc_data['source'], "incorrect 'source' from new specimen id")
+
+        # Test if no update called if no data has changed
+        with patch.object(Sample.objects, 'update_or_create', return_value=(None, False)) as mock_update_or_create:
+            obj, is_created, is_updated = Sample.objects.update_or_create_if_needed(
+                {"sample_id": new_spc_data['sample_id']},
+                new_spc_data
+            )
+            mock_update_or_create.assert_not_called()
+            self.assertIsNotNone(obj, "object should not be None")
+            self.assertFalse(is_created, "object should not be created")
+            self.assertFalse(is_updated, "object should not be updated")
+

lib/workload/stateless/stacks/metadata-manager/deploy/construct/lambda-load-custom-csv/index.ts

@@ -9,6 +9,7 @@ import {
   DockerImageFunctionProps,
   DockerImageCode,
 } from 'aws-cdk-lib/aws-lambda';
+import { EventBus } from 'aws-cdk-lib/aws-events';
 
 type LambdaProps = {
   /**
@@ -19,6 +20,10 @@ type LambdaProps = {
    * The secret for the db connection where the lambda will need access to
    */
   dbConnectionSecret: ISecret;
+  /**
+   * The eventBusName to notify metadata state change
+   */
+  eventBusName: string;
 };
 
 export class LambdaLoadCustomCSVConstruct extends Construct {
@@ -30,6 +35,8 @@ export class LambdaLoadCustomCSVConstruct extends Construct {
     this.lambda = new DockerImageFunction(this, 'LoadCustomCSVLambda', {
       environment: {
         ...lambdaProps.basicLambdaConfig.environment,
+
+        EVENT_BUS_NAME: lambdaProps.eventBusName,
       },
       securityGroups: lambdaProps.basicLambdaConfig.securityGroups,
       vpc: lambdaProps.basicLambdaConfig.vpc,
@@ -45,10 +52,14 @@ export class LambdaLoadCustomCSVConstruct extends Construct {
     lambdaProps.dbConnectionSecret.grantRead(this.lambda);
 
     // We need to store this lambda ARN somewhere so that we could refer when need to load this manually
-    const ssmParameter = new StringParameter(this, 'LoadCustomCSVLambdaArnParameterStore', {
+    new StringParameter(this, 'LoadCustomCSVLambdaArnParameterStore', {
       parameterName: '/orcabus/metadata-manager/load-custom-csv-lambda-arn',
       description: 'The ARN of the lambda that load metadata from a presigned URL CSV file',
       stringValue: this.lambda.functionArn,
     });
+
+    // The lambda will need permission to put events to the event bus when metadata state change
+    const orcabusEventBus = EventBus.fromEventBusName(this, 'EventBus', lambdaProps.eventBusName);
+    orcabusEventBus.grantPutEventsTo(this.lambda);
   }
 }