feat(export): consolidate on using shared package

Author: kschelonka

packages/aws-utils/package.json

@@ -92,6 +92,7 @@
     "@aws-sdk/client-sqs": "3.716.0",
     "@aws-sdk/lib-storage": "3.716.0",
     "@aws-sdk/s3-request-presigner": "3.716.0",
+    "@opentelemetry/api": "1.9.0",
     "@pocket-tools/event-bridge": "workspace:*",
     "@pocket-tools/ts-logger": "workspace:*",
     "@sentry/node": "8.47.0",
@@ -103,6 +104,7 @@
   "devDependencies": {
     "@pocket-tools/eslint-config": "workspace:*",
     "@semantic-release/exec": "6.0.3",
+    "@sentry/core": "^9.3.0",
     "@smithy/types": "3.5.0",
     "@types/adm-zip": "0.5.7",
     "@types/archiver": "^6.0.3",
@@ -117,7 +119,8 @@
     "ts-node": "10.9.2",
     "tsconfig": "workspace:*",
     "tsup": "8.3.5",
-    "typescript": "5.7.3"
+    "typescript": "5.7.3",
+    "unleash-client": "6.1.2"
   },
   "publishConfig": {
     "access": "public"

packages/aws-utils/src/AsyncDataExportService.spec.ts

@@ -45,7 +45,7 @@ class ExportConcrete extends AsyncDataExportService<
     return Promise.resolve([]);
   }
 
-  formatExport(entries: ExportRecord[]) {
+  async formatExport(entries: ExportRecord[]) {
     return entries;
   }
 

packages/aws-utils/src/AsyncDataExportService.ts

@@ -44,11 +44,11 @@ export abstract class AsyncDataExportService<I extends { cursor: number }, O> {
     return path.join(this.s3.partsPrefix, this.user.encodedId);
   }
 
-  abstract fileKey(part: number): string;
+  abstract fileKey(identifier: number | string): string;
 
   abstract fetchData(from: number, size: number): Promise<Array<I>>;
 
-  abstract formatExport(entries: I[]): O[];
+  abstract formatExport(entries: I[]): Promise<O[]>;
 
   abstract write(records: O[], fileKey: string): Promise<void>;
 
@@ -89,7 +89,7 @@ export abstract class AsyncDataExportService<I extends { cursor: number }, O> {
       }
       // We're finished!
       else if (entries.length <= size) {
-        const records = this.formatExport(entries);
+        const records = await this.formatExport(entries);
         await this.write(records, this.fileKey(part));
         await this.notifyComplete(
           this.user.encodedId,
@@ -100,7 +100,7 @@ export abstract class AsyncDataExportService<I extends { cursor: number }, O> {
         // Will pull in greater than or equal to cursor
         // Updates array in-place to remove the record
         const cursor = entries.splice(size)[0].cursor;
-        const records = this.formatExport(entries);
+        const records = await this.formatExport(entries);
         await this.write(records, this.fileKey(part));
         serverLogger.info({
           message: `DataExportService::${this.serviceName} - Requesting next chunk`,

servers/account-data-deleter/src/queueHandlers/queueHandler.ts packages/aws-utils/src/QueuePoller.ts

@@ -7,20 +7,25 @@ import {
   ReceiveMessageCommandOutput,
 } from '@aws-sdk/client-sqs';
 import * as Sentry from '@sentry/node';
-import { config } from '../config';
-import { SqsMessage } from '../routes/queueDelete';
 import { setTimeout } from 'timers/promises';
-import { SeverityLevel } from '@sentry/types';
-import { unleash } from '../unleash';
 import { type Unleash } from 'unleash-client';
 import { serverLogger } from '@pocket-tools/ts-logger';
-import { QueueConfig } from '../types';
 import * as otel from '@opentelemetry/api';
 
-export abstract class QueueHandler {
-  readonly sqsClient: SQSClient;
+export interface QueueConfig {
+  batchSize: number;
+  url: string;
+  visibilityTimeout: number;
+  maxMessages: number;
+  waitTimeSeconds: number;
+  defaultPollIntervalSeconds: number;
+  afterMessagePollIntervalSeconds: number;
+  messageRetentionSeconds: number;
+  name: string;
+}
+
+export abstract class QueuePoller<TMessageBody> {
   private tracer: otel.Tracer;
-  protected unleashClient: Unleash;
 
   /**
    * Class for deleting records in batches from the database,
@@ -43,23 +48,26 @@ export abstract class QueueHandler {
    * in the future.
    */
   constructor(
-    public readonly emitter: EventEmitter,
-    public readonly eventName: string,
-    public readonly queueConfig: QueueConfig,
-    pollOnInit = true,
-    unleashClient?: Unleash,
+    protected events: {
+      emitter: EventEmitter;
+      eventName: string;
+    },
+    protected sqs: {
+      config: QueueConfig;
+      client: SQSClient;
+    },
+    protected opts: {
+      pollOnInit?: boolean;
+    },
   ) {
-    this.sqsClient = new SQSClient({
-      region: config.aws.region,
-      endpoint: config.aws.endpoint,
-      maxAttempts: 3,
-    });
-    this.unleashClient = unleashClient ?? unleash();
+    // Default to polling when class is instantiated (really only when
+    // testing do you want it otherwise)
+    const pollOnInit = opts.pollOnInit != null ? opts.pollOnInit : true;
     this.tracer = otel.trace.getTracer('queue-tracer');
-    emitter.on(this.eventName, async () => await this.pollQueue());
+    events.emitter.on(events.eventName, async () => await this.pollQueue());
     // Start the polling by emitting an initial event
     if (pollOnInit) {
-      emitter.emit(this.eventName);
+      events.emitter.emit(events.eventName);
     }
   }
 
@@ -70,20 +78,22 @@ export abstract class QueueHandler {
    */
   async deleteMessage(message: Message) {
     const deleteParams = {
-      QueueUrl: this.queueConfig.url,
+      QueueUrl: this.sqs.config.url,
       ReceiptHandle: message.ReceiptHandle,
     };
     try {
-      await this.sqsClient.send(new DeleteMessageCommand(deleteParams));
+      await this.sqs.client.send(new DeleteMessageCommand(deleteParams));
     } catch (error) {
       const errorMessage = 'Error deleting message from queue';
       serverLogger.error({
         message: errorMessage,
         error: error,
         errorData: message,
+        queue: this.sqs.config.url,
+      });
+      Sentry.captureException(error, {
+        data: { ...message, queue: this.sqs.config.url },
       });
-      Sentry.addBreadcrumb({ message: errorMessage, data: message });
-      Sentry.captureException(error);
     }
   }
 
@@ -95,7 +105,7 @@ export abstract class QueueHandler {
    * @returns whether or not the message was successfully handled
    * (underlying call to AccountDeleteDataService completed without error)
    */
-  abstract handleMessage(body: object): Promise<boolean>;
+  abstract handleMessage(body: TMessageBody): Promise<boolean>;
   /**
    * Set a timeout to emit another poll event which will be handled
    * by the listener.
@@ -106,7 +116,7 @@ export abstract class QueueHandler {
       serverLogger.info(`Set next poll timeout at ${timeout}`);
       await setTimeout(timeout);
     }
-    this.emitter.emit(this.eventName);
+    this.events.emitter.emit(this.events.eventName);
   }
 
   /**
@@ -118,7 +128,7 @@ export abstract class QueueHandler {
   async pollQueue() {
     return await Sentry.withIsolationScope(async () => {
       return await this.tracer.startActiveSpan(
-        `poll-queue-${this.queueConfig.name}`,
+        `poll-queue-${this.sqs.config.name}`,
         { root: true },
         async (span: otel.Span) => {
           await this.__pollQueue(span);
@@ -140,20 +150,20 @@ export abstract class QueueHandler {
     const params = {
       // https://github.com/aws/aws-sdk/issues/233
       AttributeNames: ['SentTimestamp'] as any, // see issue above - bug in the SDK
-      MaxNumberOfMessages: this.queueConfig.maxMessages,
+      MaxNumberOfMessages: this.sqs.config.maxMessages,
       MessageAttributeNames: ['All'],
-      QueueUrl: this.queueConfig.url,
-      VisibilityTimeout: this.queueConfig.visibilityTimeout,
-      WaitTimeSeconds: this.queueConfig.waitTimeSeconds,
+      QueueUrl: this.sqs.config.url,
+      VisibilityTimeout: this.sqs.config.visibilityTimeout,
+      WaitTimeSeconds: this.sqs.config.waitTimeSeconds,
     };
 
-    serverLogger.info(`Begining polling of ${this.queueConfig.url}`);
+    serverLogger.info(`Begining polling of ${this.sqs.config.url}`);
 
     let data: ReceiveMessageCommandOutput | null = null;
-    let body: SqsMessage | null = null;
+    let body: TMessageBody | null = null;
 
     try {
-      data = await this.sqsClient.send(new ReceiveMessageCommand(params));
+      data = await this.sqs.client.send(new ReceiveMessageCommand(params));
       body =
         data.Messages &&
         data.Messages.length > 0 &&
@@ -162,9 +172,14 @@ export abstract class QueueHandler {
           : null;
     } catch (error) {
       const receiveError = 'PollQueue: Error receiving messages from queue';
-      serverLogger.error({ message: receiveError, error: error });
-      Sentry.addBreadcrumb({ message: receiveError });
-      Sentry.captureException(error, { level: 'fatal' as SeverityLevel });
+      serverLogger.error({
+        message: receiveError,
+        error: error,
+        queue: this.sqs.config.url,
+      });
+      Sentry.captureException(error, {
+        data: { queue: this.sqs.config.url },
+      });
       span.recordException(error);
       span.setStatus({ code: otel.SpanStatusCode.ERROR });
     }
@@ -176,12 +191,12 @@ export abstract class QueueHandler {
       }
       // Schedule next message poll
       await this.scheduleNextPoll(
-        this.queueConfig.afterMessagePollIntervalSeconds * 1000,
+        this.sqs.config.afterMessagePollIntervalSeconds * 1000,
       );
     } else {
       // If no messages were found, schedule another poll after a short time
       await this.scheduleNextPoll(
-        this.queueConfig.defaultPollIntervalSeconds * 1000,
+        this.sqs.config.defaultPollIntervalSeconds * 1000,
       );
     }
   }

packages/aws-utils/src/S3Bucket.ts

@@ -236,10 +236,10 @@ export class S3Bucket {
       });
       if (credentials) {
         const assumedS3 = new S3Client({
-          endpoint: config.aws.endpoint,
-          region: config.aws.region,
+          endpoint: this.awsConfig?.endpoint,
+          region: this.awsConfig?.region,
           maxAttempts: 3,
-          forcePathStyle: config.aws.endpoint != null ? true : false,
+          forcePathStyle: this.awsConfig?.endpoint != null ? true : false,
           credentials,
         });
         return await getSignedUrl(assumedS3, command, { expiresIn });

packages/aws-utils/src/index.ts

@@ -1,3 +1,4 @@
 export * from './AsyncDataExportService.ts';
 export * from './S3Bucket.ts';
 export * from './s3TestUtils.ts';
+export * from './QueuePoller.ts';