wip: infrastructure for multi-export

.docker/aws-resources/account-data-deleter.sh

@@ -3,7 +3,9 @@ set -x
 
 SQS=(
 pocket-account-data-delete-queue
+pocket-export-request-queue
 pocket-list-export-queue
+pocket-annotations-export-queue
 pocket-list-import-batch-queue
 pocket-list-import-file-queue
 )

.docker/aws-resources/shareable-lists-api.sh

@@ -1,3 +1,11 @@
+SQS=(
+pocket-shareablelist-export-queue
+)
+
+for sqs_queue in "${SQS[@]}"; do
+  awslocal sqs create-queue --queue-name "${sqs_queue}"
+done
+
 #!/bin/bash
 set -x
 bash "$(dirname "${BASH_SOURCE[0]}")/eventbus.sh"

infrastructure/account-data-deleter/src/config/index.ts

@@ -37,6 +37,8 @@ export const config = {
     databasePort: '3306',
     sqsBatchDeleteQueueName: `${prefix}-Sqs-Batch-Delete-Consumer-Queue`,
     listExportQueueName: `${prefix}-List-Export`,
+    exportRequestQueueName: `${prefix}-Export-Request`,
+    annotationsExportQueueName: `${prefix}-Annotations-Export`,
     listImportFileQueue: `${prefix}-List-Import-Files`,
     listImportBatchQueue: `${prefix}-List-Import-Batches`,
     databaseTz: 'US/Central',
@@ -48,6 +50,7 @@ export const config = {
     snsTopicName: {
       userEvents: `PocketEventBridge-${environment}-UserEvents`,
       listEvents: `PocketEventBridge-${environment}-ListEvents`,
+      exportUpdateEvents: `PocketEventBridge-${environment}-ListExportReadyEvents`,
     },
     batchDeleteLambda: {
       name: 'BatchDeleteLambda',

infrastructure/account-data-deleter/src/dataDeleterApp.ts

@@ -7,6 +7,7 @@ import {
   dataAwsRegion,
   sqsQueue,
   dataAwsSnsTopic,
+  dynamodbTable,
 } from '@cdktf/provider-aws';
 import { S3Bucket } from '@cdktf/provider-aws/lib/s3-bucket';
 
@@ -20,10 +21,13 @@ export type DataDeleterAppConfig = {
   secretsManagerKmsAlias: dataAwsKmsAlias.DataAwsKmsAlias;
   snsTopic: dataAwsSnsTopic.DataAwsSnsTopic;
   batchDeleteQueue: sqsQueue.SqsQueue;
+  exportRequestQueue: sqsQueue.SqsQueue;
+  annotationsExportQueue: sqsQueue.SqsQueue;
   listExportQueue: sqsQueue.SqsQueue;
   listExportBucket: S3Bucket;
   listExportPartsPrefix: string;
   listExportArchivesPrefix: string;
+  exportStateDb: dynamodbTable.DynamodbTable;
   importFileQueue: sqsQueue.SqsQueue;
   importBatchQueue: sqsQueue.SqsQueue;
   listImportBucket: S3Bucket;
@@ -114,10 +118,22 @@ export class DataDeleterApp extends Construct {
               name: 'SQS_BATCH_DELETE_QUEUE_URL',
               value: `https://sqs.${region.name}.amazonaws.com/${caller.accountId}/${config.envVars.sqsBatchDeleteQueueName}`,
             },
+            {
+              name: 'EXPORT_REQUEST_QUEUE_URL',
+              value: `https://sqs.${region.name}.amazonaws.com/${caller.accountId}/${config.envVars.exportRequestQueueName}`,
+            },
+            {
+              name: 'EXPORT_REQUEST_STATE_TABLE',
+              value: this.config.exportStateDb.name,
+            },
             {
               name: 'SQS_LIST_EXPORT_QUEUE_URL',
               value: `https://sqs.${region.name}.amazonaws.com/${caller.accountId}/${config.envVars.listExportQueueName}`,
             },
+            {
+              name: 'SQS_ANNOTATIONS_EXPORT_QUEUE_URL',
+              value: `https://sqs.${region.name}.amazonaws.com/${caller.accountId}/${config.envVars.annotationsExportQueueName}`,
+            },
             {
               name: 'SQS_IMPORT_BATCH_QUEUE_URL',
               value: this.config.importBatchQueue.url,
@@ -276,8 +292,10 @@ export class DataDeleterApp extends Construct {
             ],
             resources: [
               this.config.batchDeleteQueue.arn,
-              this.config.listExportQueue.arn,
               this.config.importFileQueue.arn,
+              this.config.listExportQueue.arn,
+              this.config.exportRequestQueue.arn,
+              this.config.annotationsExportQueue.arn,
             ],
             effect: 'Allow',
           },

infrastructure/account-data-deleter/src/main.ts

@@ -23,6 +23,8 @@ import {
   ApplicationSQSQueue,
   ApplicationSqsSnsTopicSubscription,
   PocketVPC,
+  ApplicationDynamoDBTable,
+  ApplicationDynamoDBTableCapacityMode,
 } from '@pocket-tools/terraform-modules';
 
 import { App, S3Backend, TerraformStack, Token } from 'cdktf';
@@ -67,6 +69,46 @@ class AccountDataDeleter extends TerraformStack {
       },
     );
 
+    const exportStateDb = new ApplicationDynamoDBTable(
+      this,
+      'export-request-state',
+      {
+        tags: config.tags,
+        prefix: `${config.shortName}-${config.environment}-Export-Request-State`,
+        capacityMode: ApplicationDynamoDBTableCapacityMode.ON_DEMAND,
+        preventDestroyTable: true,
+        tableConfig: {
+          pointInTimeRecovery: {
+            enabled: true,
+          },
+          ttl: {
+            enabled: true,
+            attributeName: 'expiresAt',
+          },
+          hashKey: 'requestId',
+          attribute: [
+            {
+              name: 'requestId',
+              type: 'S',
+            },
+          ],
+        },
+      },
+    );
+
+    const exportRequestQueue = new ApplicationSQSQueue(
+      this,
+      'export-request-consumer-queue',
+      {
+        name: config.envVars.exportRequestQueueName,
+        tags: config.tags,
+        visibilityTimeoutSeconds: 1800,
+        messageRetentionSeconds: 1209600, //14 days
+        //need to set maxReceiveCount to enable DLQ
+        maxReceiveCount: 3,
+      },
+    );
+
     const listExportQueue = new ApplicationSQSQueue(
       this,
       'list-export-consumer-queue',
@@ -80,13 +122,26 @@ class AccountDataDeleter extends TerraformStack {
       },
     );
 
+    const annotationsExportQueue = new ApplicationSQSQueue(
+      this,
+      'annotations-export-consumer-queue',
+      {
+        name: config.envVars.annotationsExportQueueName,
+        tags: config.tags,
+        visibilityTimeoutSeconds: 1800,
+        messageRetentionSeconds: 1209600, //14 days
+        //need to set maxReceiveCount to enable DLQ
+        maxReceiveCount: 3,
+      },
+    );
+
     new ApplicationSqsSnsTopicSubscription(
       this,
       'list-events-sns-subscription',
       {
-        name: `${config.envVars.listExportQueueName}-SNS`,
+        name: `${config.envVars.exportRequestQueueName}-SNS`,
         snsTopicArn: `arn:aws:sns:${pocketVpc.region}:${pocketVpc.accountId}:${config.lambda.snsTopicName.listEvents}`,
-        sqsQueue: listExportQueue.sqsQueue,
+        sqsQueue: exportRequestQueue.sqsQueue,
         filterPolicyScope: 'MessageBody',
         filterPolicy: JSON.stringify({
           'detail-type': ['list-export-requested'],
@@ -95,6 +150,22 @@ class AccountDataDeleter extends TerraformStack {
       },
     );
 
+    // Forward status updates on export components (shareable list, list, annotations)
+    new ApplicationSqsSnsTopicSubscription(
+      this,
+      'export-status-events-sns-subscription',
+      {
+        name: `${config.envVars.exportRequestQueueName}-Status-SNS`,
+        snsTopicArn: `arn:aws:sns:${pocketVpc.region}:${pocketVpc.accountId}:${config.lambda.snsTopicName.exportUpdateEvents}`,
+        sqsQueue: exportRequestQueue.sqsQueue,
+        filterPolicyScope: 'MessageBody',
+        filterPolicy: JSON.stringify({
+          'detail-type': ['export-part-complete'],
+        }),
+        tags: config.tags,
+      },
+    );
+
     // Bucket for exports plus auto-expiry rules
     const exportBucket = new s3Bucket.S3Bucket(this, 'list-export-bucket', {
       bucket: `com.getpocket-${config.environment.toLowerCase()}.list-exports`,
@@ -193,6 +264,9 @@ class AccountDataDeleter extends TerraformStack {
       snsTopic: this.getCodeDeploySnsTopic(),
       batchDeleteQueue: batchDeleteQueue.sqsQueue,
       listExportQueue: listExportQueue.sqsQueue,
+      exportRequestQueue: exportRequestQueue.sqsQueue,
+      annotationsExportQueue: annotationsExportQueue.sqsQueue,
+      exportStateDb: exportStateDb.dynamodb,
       listExportBucket: exportBucket,
       listExportPartsPrefix: partsPrefix,
       listExportArchivesPrefix: archivesPrefix,

infrastructure/pocket-event-bridge/src/main.ts

@@ -272,8 +272,11 @@ class PocketEventBus extends TerraformStack {
       name: 'ListExportReadyEvents',
       tags: config.tags,
       eventPattern: {
-        'detail-type': [PocketEventType.EXPORT_READY],
-        source: ['account-data-deleter'],
+        'detail-type': [
+          PocketEventType.EXPORT_READY,
+          PocketEventType.EXPORT_PART_COMPLETE,
+        ],
+        source: ['account-data-deleter', 'shareable-list-events'],
       },
     });
   }

infrastructure/shareable-lists-api/src/config/index.ts

@@ -15,12 +15,13 @@ const eventBusName = `PocketEventBridge-${environment}-Shared-Event-Bus`;
 const releaseSha = process.env.CIRCLE_SHA1;
 
 const s3LogsBucket = isDev ? 'pocket-data-items-dev' : 'pocket-data-items';
+const prefix = `${name}-${environment}`;
 
 export const config = {
   name,
   isDev,
   isProd,
-  prefix: `${name}-${environment}`,
+  prefix,
   circleCIPrefix: `/${name}/CircleCI/${environment}`,
   shortName: 'SLAPI',
   releaseSha,
@@ -39,6 +40,10 @@ export const config = {
     component_code: `pocket-${name.toLowerCase()}`,
     env_code: isDev ? 'dev' : 'prod',
   },
+  export: {
+    queue: `${prefix}-SharedList-Export`,
+    requestTopic: `PocketEventBridge-${environment}-ListEvents`,
+  },
   eventBusName,
   lambda: {
     snsTopicName: {

infrastructure/shareable-lists-api/src/main.ts

@@ -14,6 +14,7 @@ import { provider as archiveProvider } from '@cdktf/provider-archive';
 import { provider as nullProvider } from '@cdktf/provider-null';
 import {
   ApplicationRDSCluster,
+  ApplicationSQSQueue,
   ApplicationSqsSnsTopicSubscription,
   PocketALBApplication,
   PocketAwsSyntheticChecks,
@@ -75,12 +76,43 @@ class ShareableListsAPI extends TerraformStack {
       },
     );
 
+    // Export work queue
+    const exportQueue = new ApplicationSQSQueue(
+      this,
+      'sharelists-export-consumer-queue',
+      {
+        name: config.export.queue,
+        tags: config.tags,
+        visibilityTimeoutSeconds: 1800,
+        messageRetentionSeconds: 1209600, //14 days
+        //need to set maxReceiveCount to enable DLQ
+        maxReceiveCount: 3,
+      },
+    );
+
+    // Subscription to list export topic
+    new ApplicationSqsSnsTopicSubscription(
+      this,
+      'list-events-sns-subscription',
+      {
+        name: `${config.export.queue}-SNS`,
+        snsTopicArn: `arn:aws:sns:${pocketVpc.region}:${pocketVpc.accountId}:${config.export.requestTopic}`,
+        sqsQueue: exportQueue.sqsQueue,
+        filterPolicyScope: 'MessageBody',
+        filterPolicy: JSON.stringify({
+          'detail-type': ['list-export-requested'],
+        }),
+        tags: config.tags,
+      },
+    );
+
     const alarmSnsTopic = this.getCodeDeploySnsTopic();
 
     this.createPocketAlbApplication({
       rds: this.createRds(pocketVpc),
       secretsManagerKmsAlias: this.getSecretsManagerKmsAlias(),
       snsTopic: alarmSnsTopic,
+      exportSqs: exportQueue.sqsQueue,
       region,
       caller,
       cache,
@@ -217,10 +249,18 @@ class ShareableListsAPI extends TerraformStack {
     caller: dataAwsCallerIdentity.DataAwsCallerIdentity;
     secretsManagerKmsAlias: dataAwsKmsAlias.DataAwsKmsAlias;
     snsTopic: dataAwsSnsTopic.DataAwsSnsTopic;
+    exportSqs: sqsQueue.SqsQueue;
     cache: { primaryEndpoint: string; readerEndpoint: string };
   }): PocketALBApplication {
-    const { rds, region, caller, secretsManagerKmsAlias, snsTopic, cache } =
-      dependencies;
+    const {
+      rds,
+      region,
+      caller,
+      secretsManagerKmsAlias,
+      snsTopic,
+      cache,
+      exportSqs,
+    } = dependencies;
 
     return new PocketALBApplication(this, 'application', {
       internal: true,
@@ -280,6 +320,10 @@ class ShareableListsAPI extends TerraformStack {
               name: 'REDIS_IS_TLS',
               value: 'true',
             },
+            {
+              name: 'EXPORT_QUEUE_URL',
+              value: `https://sqs.${region.name}.amazonaws.com/${caller.accountId}/${config.export.queue}`,
+            },
           ],
           logGroup: this.createCustomLogGroup('app'),
           logMultilinePattern: '^\\S.+',
@@ -370,6 +414,17 @@ class ShareableListsAPI extends TerraformStack {
             resources: ['*'],
             effect: 'Allow',
           },
+          {
+            // export queue
+            actions: [
+              'sqs:ReceiveMessage',
+              'sqs:DeleteMessage',
+              'sqs:SendMessage',
+              'sqs:SendMessageBatch',
+            ],
+            resources: [exportSqs.arn],
+            effect: 'Allow',
+          },
           {
             actions: ['events:PutEvents'],
             resources: [

infrastructure/transactional-emails/src/config/index.ts

@@ -44,7 +44,13 @@ export const config = {
       { name: 'PremiumPurchaseEvents' },
       { name: 'UserRegistrationEvents' },
       { name: 'ForgotPasswordEvents' },
-      { name: 'ListExportReadyEvents' },
+      {
+        name: 'ListExportReadyEvents',
+        filterPolicyScope: 'MessageBody',
+        filterPolicy: JSON.stringify({
+          'detail-type': ['list-export-ready'],
+        }),
+      },
       {
         name: 'ListEvents',
         filterPolicyScope: 'MessageBody',

packages/event-bridge/src/events/events.ts

@@ -11,6 +11,7 @@ export enum PocketEventType {
   ACCOUNT_PASSWORD_CHANGED = 'account-password-changed', //source: user-event
   FORGOT_PASSWORD = 'Forgot Password Request', //source: web-repo
   EXPORT_READY = 'list-export-ready', // source: account-data-deleter
+  EXPORT_PART_COMPLETE = 'export-part-complete', // source: account-data-deleter or shareable-lists-api
   EXPORT_REQUESTED = 'list-export-requested', // source: list-api
   // List Events
   ADD_ITEM = 'ADD_ITEM',