Feat(srm) sequencerun comments module

lib/workload/stateless/stacks/sequence-run-manager/deploy/README.md

@@ -17,10 +17,10 @@ Hot-deploy against dev:
 export AWS_PROFILE=umccr-dev-admin
 
 yarn cdk-stateless list
-yarn cdk-stateless synth -e OrcaBusStatelessPipeline/BetaDeployment/SequenceRunManagerStack
-yarn cdk-stateless diff -e OrcaBusStatelessPipeline/BetaDeployment/SequenceRunManagerStack
-yarn cdk-stateless deploy -e OrcaBusStatelessPipeline/BetaDeployment/SequenceRunManagerStack
-yarn cdk-stateless destroy -e OrcaBusStatelessPipeline/BetaDeployment/SequenceRunManagerStack
+yarn cdk-stateless synth -e OrcaBusStatelessPipeline/OrcaBusBeta/SequenceRunManagerStack
+yarn cdk-stateless diff -e OrcaBusStatelessPipeline/OrcaBusBeta/SequenceRunManagerStack
+yarn cdk-stateless deploy -e OrcaBusStatelessPipeline/OrcaBusBeta/SequenceRunManagerStack
+yarn cdk-stateless destroy -e OrcaBusStatelessPipeline/OrcaBusBeta/SequenceRunManagerStack
 ```
 
 CloudFormation template:

lib/workload/stateless/stacks/sequence-run-manager/deploy/stack.ts

@@ -3,7 +3,8 @@ import * as cdk from 'aws-cdk-lib';
 import { aws_lambda, aws_secretsmanager, Duration, Stack } from 'aws-cdk-lib';
 import { Construct } from 'constructs';
 import { ISecurityGroup, IVpc, SecurityGroup, Vpc, VpcLookupOptions } from 'aws-cdk-lib/aws-ec2';
-import { EventBus, IEventBus } from 'aws-cdk-lib/aws-events';
+import { EventBus, IEventBus, Rule } from 'aws-cdk-lib/aws-events';
+import { LambdaFunction } from 'aws-cdk-lib/aws-events-targets';
 import { PythonFunction, PythonLayerVersion } from '@aws-cdk/aws-lambda-python-alpha';
 import { HttpLambdaIntegration } from 'aws-cdk-lib/aws-apigatewayv2-integrations';
 import { HttpMethod, HttpRoute, HttpRouteKey } from 'aws-cdk-lib/aws-apigatewayv2';
@@ -133,27 +134,47 @@ export class SequenceRunManagerStack extends Stack {
      */
     const procSqsFn = this.createPythonFunction('ProcHandler', {
       index: 'sequence_run_manager_proc/lambdas/bssh_event.py',
-      handler: 'sqs_handler',
+      handler: 'event_handler',
       timeout: Duration.minutes(2),
       memorySize: 512,
       reservedConcurrentExecutions: 1,
     });
 
     this.mainBus.grantPutEventsTo(procSqsFn);
-    // this.setupEventRule(procSqsFn);  // TODO comment this out for now
+    this.setupEventRule(procSqsFn); // TODO comment this out for now
   }
 
-  // private setupEventRule(fn: aws_lambda.Function) {
-  //   const eventRule = new Rule(this, this.id + 'EventRule', {
-  //     ruleName: this.id + 'EventRule',
-  //     description: 'Rule to send {event_type.value} events to the {handler.function_name} Lambda',
-  //     eventBus: this.props.mainBus,
-  //   });
-  //
-  //   eventRule.addTarget(new aws_events_targets.LambdaFunction(fn));
-  //   eventRule.addEventPattern({
-  //     source: ['ORCHESTRATOR'], // FIXME complete source to destination event mapping
-  //     detailType: ['SequenceRunStateChange'],
-  //   });
-  // }
+  private setupEventRule(fn: aws_lambda.Function) {
+    /**
+     * For sequence run manager, we are using orcabus events ( source from BSSH ENS event pipe) to trigger the lambda function.
+     * event rule to filter the events that we are interested in.
+     * event pattern: see below
+     * process lambda will record the event to the database, and emit the 'SequenceRunStateChange' event to the event bus.
+     *
+     */
+    const eventRule = new Rule(this, this.stackName + 'EventRule', {
+      ruleName: this.stackName + 'EventRule',
+      description: 'Rule to send {event_type.value} events to the {handler.function_name} Lambda',
+      eventBus: this.mainBus,
+    });
+    eventRule.addEventPattern({
+      detailType: ['Event from aws:sqs'],
+      detail: {
+        'ica-event': {
+          // mandatory fields (gdsFolderPath, gdsVolumeName(starts with bssh), instrumentRunId, dateModified)
+          gdsFolderPath: [{ exists: true }],
+          gdsVolumeName: [{ prefix: 'bssh' }],
+          instrumentRunId: [{ exists: true }],
+          dateModified: [{ exists: true }],
+
+          // optional fields (flowcell barcode, sample sheet name, reagent barcode, ica project id, api url, name)
+          acl: [{ prefix: 'wid:' }, { prefix: 'tid:' }],
+          id: [{ exists: true }],
+          status: [{ exists: true }],
+        },
+      },
+    });
+
+    eventRule.addTarget(new LambdaFunction(fn));
+  }
 }

lib/workload/stateless/stacks/sequence-run-manager/deps/requirements.txt

@@ -22,3 +22,4 @@ serverless-wsgi==3.0.5
 # for sequencerunstatechange package
 six==1.16.0
 regex==2024.9.11
+ulid-py==1.1.0

lib/workload/stateless/stacks/sequence-run-manager/sequence_run_manager/migrations/0001_initial.py

@@ -1,18 +1,66 @@
-# Generated by Django 4.2.1 on 2023-06-14 07:37
+# Generated by Django 5.1.2 on 2024-12-03 07:25
 
+import django.core.validators
+import django.db.models.deletion
 from django.db import migrations, models
 
 
 class Migration(migrations.Migration):
+
     initial = True
 
     dependencies = []
 
     operations = [
+        migrations.CreateModel(
+            name="Comment",
+            fields=[
+                (
+                    "orcabus_id",
+                    models.CharField(
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                        unique=True,
+                        validators=[
+                            django.core.validators.RegexValidator(
+                                code="invalid_orcabus_id",
+                                message="ULID is expected to be 26 characters long",
+                                regex="^[\\w]{26}$",
+                            )
+                        ],
+                    ),
+                ),
+                ("comment", models.TextField()),
+                ("association_id", models.CharField(max_length=255)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("created_by", models.CharField(max_length=255)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("is_deleted", models.BooleanField(default=False)),
+            ],
+            options={
+                "abstract": False,
+            },
+        ),
         migrations.CreateModel(
             name="Sequence",
             fields=[
-                ("id", models.BigAutoField(primary_key=True, serialize=False)),
+                (
+                    "orcabus_id",
+                    models.CharField(
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                        unique=True,
+                        validators=[
+                            django.core.validators.RegexValidator(
+                                code="invalid_orcabus_id",
+                                message="ULID is expected to be 26 characters long",
+                                regex="^[\\w]{26}$",
+                            )
+                        ],
+                    ),
+                ),
                 ("instrument_run_id", models.CharField(max_length=255, unique=True)),
                 ("run_volume_name", models.TextField()),
                 ("run_folder_path", models.TextField()),
@@ -21,10 +69,10 @@ class Migration(migrations.Migration):
                     "status",
                     models.CharField(
                         choices=[
-                            ("started", "Started"),
-                            ("failed", "Failed"),
-                            ("succeeded", "Succeeded"),
-                            ("aborted", "Aborted"),
+                            ("STARTED", "Started"),
+                            ("FAILED", "Failed"),
+                            ("SUCCEEDED", "Succeeded"),
+                            ("ABORTED", "Aborted"),
                         ],
                         max_length=255,
                     ),
@@ -56,4 +104,39 @@ class Migration(migrations.Migration):
                 "abstract": False,
             },
         ),
+        migrations.CreateModel(
+            name="State",
+            fields=[
+                (
+                    "orcabus_id",
+                    models.CharField(
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                        unique=True,
+                        validators=[
+                            django.core.validators.RegexValidator(
+                                code="invalid_orcabus_id",
+                                message="ULID is expected to be 26 characters long",
+                                regex="^[\\w]{26}$",
+                            )
+                        ],
+                    ),
+                ),
+                ("status", models.CharField(max_length=255)),
+                ("timestamp", models.DateTimeField()),
+                ("comment", models.CharField(blank=True, max_length=255, null=True)),
+                (
+                    "sequence",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="states",
+                        to="sequence_run_manager.sequence",
+                    ),
+                ),
+            ],
+            options={
+                "abstract": False,
+            },
+        ),
     ]

lib/workload/stateless/stacks/sequence-run-manager/sequence_run_manager/models/__init__.py

@@ -1,3 +1,5 @@
 # https://docs.djangoproject.com/en/4.1/topics/db/models/#organizing-models-in-a-package
 
 from .sequence import Sequence
+from .comment import Comment
+from .state import State

lib/workload/stateless/stacks/sequence-run-manager/sequence_run_manager/models/base.py

@@ -1,9 +1,11 @@
 import logging
 import operator
+import ulid
 from functools import reduce
 from typing import List
 
 from django.core.exceptions import FieldError
+from django.core.validators import RegexValidator
 from django.db import models
 from django.db.models import (
     Q,
@@ -23,6 +25,12 @@
 
 logger = logging.getLogger(__name__)
 
+orcabus_id_validator = RegexValidator(
+                regex=r'^[\w]{26}$',
+                message='ULID is expected to be 26 characters long',
+                code='invalid_orcabus_id'
+            )
+
 
 class OrcaBusBaseManager(models.Manager):
     @staticmethod
@@ -78,6 +86,32 @@ def exclude_params(params):
 class OrcaBusBaseModel(models.Model):
     class Meta:
         abstract = True
+
+    orcabus_id_prefix = None
+
+    orcabus_id = models.CharField(
+        primary_key=True,
+        unique=True,
+        editable=False,
+        blank=False,
+        null=False,
+        validators=[orcabus_id_validator]
+    )
+
+    def save(self, *args, **kwargs):
+        # handle the OrcaBus ID
+        if not self.orcabus_id:
+            # if no OrcaBus ID was provided, then generate one
+            self.orcabus_id = ulid.new().str
+        else:
+            # check provided OrcaBus ID
+            if len(self.orcabus_id) > 26:
+                # assume the OrcaBus ID carries the prefix
+                # we strip it off and continue to the validation
+                l = len(self.orcabus_id_prefix)
+                self.orcabus_id = str(self.orcabus_id)[l:]
+        self.full_clean()  # make sure we are validating the inputs (especially the OrcaBus ID)
+        return super(OrcaBusBaseModel, self).save(*args, **kwargs)
 
     @classmethod
     def get_fields(cls):

lib/workload/stateless/stacks/sequence-run-manager/sequence_run_manager/models/comment.py

@@ -0,0 +1,27 @@
+import logging
+
+from django.db import models
+
+from sequence_run_manager.models.base import OrcaBusBaseModel, OrcaBusBaseManager
+
+logger = logging.getLogger(__name__)
+
+class CommentManager(OrcaBusBaseManager):
+    pass
+
+
+class Comment(OrcaBusBaseModel):
+    # primary key
+    orcabus_id_prefix = 'cmt.'
+
+    comment = models.TextField(null=False, blank=False)
+    association_id = models.CharField(max_length=255, null=False, blank=False) # comment association object id
+    created_at = models.DateTimeField(auto_now_add=True)
+    created_by = models.CharField(max_length=255, null=False, blank=False)
+    updated_at = models.DateTimeField(auto_now=True)
+    is_deleted = models.BooleanField(default=False)
+
+    objects = CommentManager()
+
+    def __str__(self):
+        return f"ID: {self.orcabus_id}, comment: {self.comment}, from {self.created_by}, for {self.association_id}"

lib/workload/stateless/stacks/sequence-run-manager/sequence_run_manager/models/sequence.py

@@ -30,7 +30,8 @@ def from_value(cls, value):
     def from_seq_run_status(cls, value):
         """
         See Run Status
-        https://support.illumina.com/help/BaseSpace_Sequence_Hub/Source/Informatics/BS/Statuses_swBS.htm
+        https://help.basespace.illumina.com/automate/statuses
+        https://support.illumina.com/help/BaseSpace_Sequence_Hub/Source/Informatics/BS/Statuses_swBS.htm (deprecated)
 
         Note that we don't necessary support all these statuses. In the following check, those values come
         from observed values from our BSSH run events.
@@ -61,7 +62,7 @@ def get_by_keyword(self, **kwargs) -> QuerySet:
 
 class Sequence(OrcaBusBaseModel):
     # primary key
-    id = models.BigAutoField(primary_key=True)
+    orcabus_id_prefix = 'seq.'
 
     # mandatory non-nullable base fields
     instrument_run_id = models.CharField(

lib/workload/stateless/stacks/sequence-run-manager/sequence_run_manager/models/state.py

@@ -0,0 +1,26 @@
+import logging
+
+from django.db import models
+
+from sequence_run_manager.models.base import OrcaBusBaseModel, OrcaBusBaseManager
+from sequence_run_manager.models.sequence import Sequence
+
+logger = logging.getLogger(__name__)
+
+class StateManager(OrcaBusBaseManager):
+    pass
+
+
+class State(OrcaBusBaseModel):
+    orcabus_id_prefix = 'sqs.'
+
+    status = models.CharField(max_length=255, null=False, blank=False)
+    timestamp = models.DateTimeField()
+    comment = models.CharField(max_length=255, null=True, blank=True)
+
+    sequence = models.ForeignKey(Sequence, on_delete=models.CASCADE, related_name='states')
+
+    objects = StateManager()
+
+    def __str__(self):
+        return f"ID: {self.orcabus_id}, status: {self.status}, for {self.sequence}"