Add the fastq sync manager (#900)

Author: alexiswl

config/config.ts

@@ -72,6 +72,10 @@ import {
   getFastqUnarchivingManagerStackProps,
   getFastqUnarchivingManagerTableStackProps,
 } from './stacks/fastqUnarchivingManager';
+import {
+  getFastqSyncManagerStackProps,
+  getFastqSyncManagerTableStackProps,
+} from './stacks/fastqSyncManager';
 
 interface EnvironmentConfig {
   name: string;
@@ -115,6 +119,7 @@ export const getEnvironmentConfig = (stage: AppStage): EnvironmentConfig | null
       accessKeySecretStackProps: getAccessKeySecretStackProps(stage),
       fastqManagerTableStackProps: getFastqManagerTableStackProps(stage),
       fastqUnarchivingManagerTableStackProps: getFastqUnarchivingManagerTableStackProps(),
+      fastqSyncManagerTableStackProps: getFastqSyncManagerTableStackProps(),
     },
     statelessConfig: {
       metadataManagerStackProps: getMetadataManagerStackProps(stage),
@@ -147,6 +152,7 @@ export const getEnvironmentConfig = (stage: AppStage): EnvironmentConfig | null
       pgDDProps: getPgDDProps(stage),
       fastqManagerStackProps: getFastqManagerStackProps(stage),
       fastqUnarchivingManagerStackProps: getFastqUnarchivingManagerStackProps(stage),
+      fastqSyncManagerStackProps: getFastqSyncManagerStackProps(stage),
     },
   };
 

config/constants.ts

@@ -961,3 +961,8 @@ export const fastqUnarchivingEventDetailType = {
   updateJob: 'FastqUnarchivingJobUpdated',
 };
 export const fastqUnarchivingManagerEventSource = 'orcabus.fastqunarchivingmanager';
+
+/*
+Fastq sync service
+*/
+export const fastqSyncEventDetailType = 'fastqSync';

config/stacks/fastqSyncManager.ts

@@ -0,0 +1,91 @@
+import {
+  AppStage,
+  /* Events */
+  eventBusName,
+  fastqManagerEventDetails,
+  fastqManagerEventSource,
+  fastqSyncEventDetailType,
+  fastqUnarchivingEventDetailType,
+  fastqUnarchivingManagerEventSource,
+  hostedZoneNameParameterPath,
+  icav2PipelineCacheBucket,
+  icav2PipelineCachePrefix,
+  jwtSecretName,
+} from '../constants';
+
+import { FastqSyncManagerStackConfig } from '../../lib/workload/stateless/stacks/fastq-sync/deploy/interfaces';
+import { FastqSyncTableConfig } from '../../lib/workload/stateful/stacks/fastq-sync-dynamodb/deploy/stack';
+
+/*
+export interface FastqSyncManagerStackConfig {
+  /*
+  Orcabus token and zone name for external lambda functions
+  */
+//orcabusTokenSecretsManagerPath: string;
+//hostedZoneNameSsmParameterPath: string;
+
+/*
+  Data tables
+  */
+//fastqSyncDynamodbTableName: string;
+/*
+  Event bus stuff
+  */
+//eventBusName: string;
+// eventTriggers: FastqSyncEventTriggers
+//}
+//*/
+
+const fastqSyncDynamodbTableName = 'fastqSyncTokenTable';
+
+// Stateful
+export const getFastqSyncManagerTableStackProps = (): FastqSyncTableConfig => {
+  return {
+    /* DynamoDB table for fastq list rows */
+    dynamodbTableName: fastqSyncDynamodbTableName,
+  };
+};
+
+// Stateless
+export const getFastqSyncManagerStackProps = (stage: AppStage): FastqSyncManagerStackConfig => {
+  return {
+    /*
+    Table stuff
+    */
+    fastqSyncDynamodbTableName: fastqSyncDynamodbTableName,
+
+    /*
+    Events stuff
+    */
+    eventBusName: eventBusName,
+    eventTriggers: {
+      fastqSetUpdated: {
+        eventSource: fastqManagerEventSource,
+        eventDetailType: fastqManagerEventDetails.updateFastqSet,
+      },
+      fastqListRowUpdated: {
+        eventSource: fastqManagerEventSource,
+        eventDetailType: fastqManagerEventDetails.updateFastqListRow,
+      },
+      fastqUnarchiving: {
+        eventSource: fastqUnarchivingManagerEventSource,
+        eventDetailType: fastqUnarchivingEventDetailType.updateJob,
+      },
+      fastqSync: {
+        eventDetailType: fastqSyncEventDetailType,
+      },
+    },
+
+    /*
+    Orcabus token and zone name for external lambda functions
+    */
+    orcabusTokenSecretsManagerPath: jwtSecretName,
+    hostedZoneNameSsmParameterPath: hostedZoneNameParameterPath,
+
+    /*
+    S3 Stuff
+    */
+    pipelineCacheBucketName: icav2PipelineCacheBucket[stage],
+    pipelineCachePrefix: icav2PipelineCachePrefix[stage],
+  };
+};

lib/workload/stateful/stacks/fastq-sync-dynamodb/Readme.md

@@ -0,0 +1,6 @@
+# Fastq Sync dynamodb
+
+Simple database with dynamodb to store task tokens and link them to fastq set ids (And vice versa)
+
+A task token can only map to one fastq set id, but a fastq set id can map to multiple task tokens.
+

lib/workload/stateful/stacks/fastq-sync-dynamodb/deploy/stack.ts

@@ -0,0 +1,22 @@
+import * as cdk from 'aws-cdk-lib';
+import { Construct } from 'constructs';
+import { DynamodbPartitionedPipelineConstruct } from '../../../../components/dynamodb-partitioned-table';
+
+export interface FastqSyncTableConfig {
+  dynamodbTableName: string;
+}
+
+export type FastqSyncManagerTableStackProps = FastqSyncTableConfig & cdk.StackProps;
+
+export class FastqSyncManagerTable extends cdk.Stack {
+  constructor(scope: Construct, id: string, props: FastqSyncManagerTableStackProps) {
+    super(scope, id, props);
+
+    /*
+    Initialise dynamodb table, where id_type is the primary sort key
+    */
+    const dynamodbTable = new DynamodbPartitionedPipelineConstruct(this, 'fastq_sync_table', {
+      tableName: props.dynamodbTableName,
+    });
+  }
+}

lib/workload/stateful/statefulStackCollectionClass.ts

@@ -74,6 +74,10 @@ import {
   FastqUnarchivingManagerTable,
   FastqUnarchivingManagerTableStackProps,
 } from './stacks/fastq-unarchiving-dynamodb/deploy';
+import {
+  FastqSyncManagerTable,
+  FastqSyncManagerTableStackProps,
+} from './stacks/fastq-sync-dynamodb/deploy/stack';
 
 export interface StatefulStackCollectionProps {
   dataBucketStackProps: DataBucketStackProps;
@@ -98,6 +102,7 @@ export interface StatefulStackCollectionProps {
   accessKeySecretStackProps: AccessKeySecretStackProps;
   fastqManagerTableStackProps: FastqManagerTableStackProps;
   fastqUnarchivingManagerTableStackProps: FastqUnarchivingManagerTableStackProps;
+  fastqSyncManagerTableStackProps: FastqSyncManagerTableStackProps;
 }
 
 export class StatefulStackCollection {
@@ -125,6 +130,7 @@ export class StatefulStackCollection {
   readonly accessKeySecretStack: Stack;
   readonly fastqManagerTableStack: Stack;
   readonly fastqUnarchivingManagerTableStack: Stack;
+  readonly fastqSyncManagerTableStack: Stack;
 
   constructor(
     scope: Construct,
@@ -292,6 +298,15 @@ export class StatefulStackCollection {
         ...statefulConfiguration.fastqUnarchivingManagerTableStackProps,
       }
     );
+
+    this.fastqSyncManagerTableStack = new FastqSyncManagerTable(
+      scope,
+      'FastqSyncManagerTableStack',
+      {
+        ...this.createTemplateProps(env, 'FastqSyncManagerTableStack'),
+        ...statefulConfiguration.fastqSyncManagerTableStackProps,
+      }
+    );
   }
 
   /**

lib/workload/stateless/stacks/fastq-sync/Readme.md

@@ -0,0 +1,62 @@
+# Fastq Sync Service
+
+The fastq sync service is a simple service that allows step functions with fastq set ids as inputs to 'hang' 
+until the requirements of the fastq set have been met. 
+
+This is useful for workflow-glue services that have fastq set ids but need to wait for either
+
+1. The fastq set readsets to be created
+2. The fastq set to have been qc'd AND have a fingerprint file and compression information
+3. This is also useful for data sharing services that require the fastqs to be unarchived before they can be shared
+
+The step function will then hang at that step until the task token has been 'unlocked' by the fastq sync service.
+
+## Registering task tokens
+
+Workflow glue services can use the fastq sync service by generating the following event
+
+```json5
+{
+  "EventBusName": "OrcaBusMain",
+  "Source": "doesnt matter",
+  "DetailType": "FastqSync",
+  "Detail": {
+    "taskToken":  "uuid",
+    "fastqSetId": "fqs.123456",
+    // Then one or more of the following
+    // Requirements can be left out if not needed
+    "requirements": {
+      // Do all fastq list rows in the set contain readsets?
+      "hasActiveReadSet": true,  
+      // Do all fastq list rows in the set contain an ntsm uri?
+      "hasFingerprint": true,  
+      // Do all fastq list rows in the set contain compression information?
+      // Useful if the fastq list rows are in ora format. 
+      // Some pipelines require the gzip file size in bytes in order 
+      // to stream the gzip file from ora back into s3 
+      "hasFileCompressionInformation": true,  
+      // Do all fastq list rows in the set contain qc information?
+      // We don't use this for anything yet but we may use this in the future
+      // to ensure that a fastq set has met the ideal coverage levels
+      "hasQc": true,
+    },
+    "forceUnarchiving": true,  // Force unarchiving of a fastq file if necessary, will fail if not set and fastq is in archive
+  }
+}
+```
+
+The fastq sync service will also trigger the qc, fingerprint or compression information services if they do not exist. 
+
+If any of the fastq list rows are in archive, the fastq sync service will also trigger the fastq unarchiving service to thaw out these fastq list rows.
+And place them into the 'byob' bucket.  
+
+
+## Unlocking task tokens
+
+The fastq sync service will then also listen for the following event types:
+
+1. FastqListRowUpdated (from the fastq management service)
+2. UnarchivingJobUpdated (from the fastq unarchiving service, where the status is 'SUCCEEDED')
+
+Everytime one of the events is triggered, the fastq sync service will check if the fastq list row or fastq set has met the requirements.
+If all requirements are met for the fastq set, the fastq sync service will unlock the task token.