SIMSBIOHUB-884: Fix Malware Publish Edge Cases (#340)

* refactor(malware): Fixed a few maleware edge cases issues.

* fix(malware): fix integration tests

* fix(maleware): lint

Author: dwhogan

api/.mocharc.system.json

@@ -2,5 +2,6 @@
   "extension": ["ts"],
   "spec": "src/__integration__/system/*.integration.ts",
   "require": "ts-node/register",
+  "node-option": ["no-experimental-strip-types"],
   "timeout": 60000
 }

api/src/__integration__/system/malware-scan-worker.integration.ts

@@ -6,6 +6,7 @@ import { expect } from 'chai';
 import { Knex, knex } from 'knex';
 import sinon from 'sinon';
 import * as tar from 'tar-stream';
+import { defaultPoolConfig, getAPIUserDBConnection, initDBPool } from '../../database/db';
 import { initPgBoss, stopPgBoss } from '../../queue/pg-boss-service';
 import * as publisher from '../../queue/publisher';
 import { BucketType, ObjectStorageService } from '../../services/object-storage/object-storage-service';
@@ -48,6 +49,8 @@ describe('Malware Scan Worker', function () {
   let db: Knex;
   let storageService: ObjectStorageService;
   const createdArtifactIds: number[] = [];
+  const createdUploadIds: string[] = [];
+  const createdSubmissionIds: number[] = [];
   const createdObjectKeys: string[] = [];
 
   before(async () => {
@@ -65,6 +68,7 @@ describe('Malware Scan Worker', function () {
     storageService = new ObjectStorageService();
     // Stub the downstream job publisher so the scan doesn't chain into validation
     sinon.stub(publisher, 'publishProcessSubmissionFeaturesJob').resolves();
+    initDBPool(defaultPoolConfig);
     await initPgBoss();
   });
 
@@ -80,8 +84,16 @@ describe('Malware Scan Worker', function () {
           )
           .del();
         await db('biohub.artifact_security').whereIn('artifact_id', createdArtifactIds).del();
+        await db('biohub.upload_archive').whereIn('artifact_id', createdArtifactIds).del();
         await db('biohub.artifact').whereIn('artifact_id', createdArtifactIds).del();
       }
+      if (createdUploadIds.length > 0) {
+        await db('biohub.submission_upload').whereIn('upload_id', createdUploadIds).del();
+        await db('biohub.upload').whereIn('upload_id', createdUploadIds).del();
+      }
+      if (createdSubmissionIds.length > 0) {
+        await db('biohub.submission').whereIn('submission_id', createdSubmissionIds).del();
+      }
       for (const key of createdObjectKeys) {
         try {
           await storageService.deleteFile(BucketType.QUARANTINE, key);
@@ -125,6 +137,42 @@ describe('Malware Scan Worker', function () {
       .returning('artifact_id');
     createdArtifactIds.push(artifact.artifact_id);
 
+    // Create upload + upload_archive + submission_upload (mirrors completeArchiveUpload)
+    const [upload] = await db('biohub.upload')
+      .insert({
+        upload_status: 'completed',
+        record_end_date: '9999-12-31',
+        create_user: SYSTEM_USER_ID
+      })
+      .returning('upload_id');
+    createdUploadIds.push(upload.upload_id);
+
+    await db('biohub.upload_archive').insert({
+      upload_id: upload.upload_id,
+      artifact_id: artifact.artifact_id,
+      archive_status: 'blocked',
+      create_user: SYSTEM_USER_ID
+    });
+
+    const [submission] = await db('biohub.submission')
+      .insert({
+        uuid: db.raw('gen_random_uuid()'),
+        system_user_id: SYSTEM_USER_ID,
+        source_system: 'SIMS',
+        name: 'integration-test',
+        description: 'integration-test',
+        comment: '',
+        create_user: SYSTEM_USER_ID
+      })
+      .returning('submission_id');
+    createdSubmissionIds.push(submission.submission_id);
+
+    await db('biohub.submission_upload').insert({
+      submission_id: submission.submission_id,
+      upload_id: upload.upload_id,
+      create_user: SYSTEM_USER_ID
+    });
+
     const [security] = await db('biohub.artifact_security')
       .insert({
         artifact_id: artifact.artifact_id,
@@ -133,7 +181,14 @@ describe('Malware Scan Worker', function () {
       })
       .returning('artifact_security_id');
 
-    await publisher.publishMalwareScanJob({ artifactSecurityId: security.artifact_security_id });
+    const connection = getAPIUserDBConnection();
+    try {
+      await connection.open();
+      await publisher.publishMalwareScanJob(connection, { artifactSecurityId: security.artifact_security_id });
+      await connection.commit();
+    } finally {
+      connection.release();
+    }
     const result = await waitForScanResult(db, security.artifact_security_id);
 
     return { result, objectKey };

api/src/queue/publisher.test.ts

@@ -238,8 +238,9 @@ describe('publisher', () => {
 
       sinon.stub(pgBossService, 'getPgBoss').returns(mockBoss as any);
 
+      const mockConnection = getMockDBConnection();
       const data = { artifactSecurityId: 'artifact-security-123' };
-      const result = await publishMalwareScanJob(data);
+      const result = await publishMalwareScanJob(mockConnection, data);
 
       expect(createQueueStub.calledOnce).to.be.true;
       expect(createQueueStub.firstCall.args[0]).to.equal(JobQueues.MALWARE_SCAN);
@@ -257,7 +258,7 @@ describe('publisher', () => {
 
       sinon.stub(pgBossService, 'getPgBoss').returns(mockBoss as any);
 
-      await publishMalwareScanJob({ artifactSecurityId: 'artifact-security-456' });
+      await publishMalwareScanJob(getMockDBConnection(), { artifactSecurityId: 'artifact-security-456' });
 
       const options = sendStub.firstCall.args[2];
       expect(options.retryLimit).to.equal(3);
@@ -273,20 +274,36 @@ describe('publisher', () => {
 
       sinon.stub(pgBossService, 'getPgBoss').returns(mockBoss as any);
 
-      await publishMalwareScanJob({ artifactSecurityId: '123' });
+      await publishMalwareScanJob(getMockDBConnection(), { artifactSecurityId: '123' });
 
       const options = sendStub.firstCall.args[2];
       expect(options.singletonKey).to.equal('artifact-security-123');
     });
 
+    it('passes db option to boss.send for transactional publishing', async () => {
+      const sendStub = sinon.stub().resolves('scan-job-id');
+      const createQueueStub = sinon.stub().resolves();
+      const mockBoss = { send: sendStub, createQueue: createQueueStub };
+
+      sinon.stub(pgBossService, 'getPgBoss').returns(mockBoss as any);
+
+      await publishMalwareScanJob(getMockDBConnection(), { artifactSecurityId: 'tx-test-123' });
+
+      const options = sendStub.firstCall.args[2];
+      expect(options.db).to.exist;
+      expect(options.db.executeSql).to.be.a('function');
+    });
+
     it('returns duplicate status when send returns null', async () => {
       const sendStub = sinon.stub().resolves(null);
       const createQueueStub = sinon.stub().resolves();
       const mockBoss = { send: sendStub, createQueue: createQueueStub };
 
       sinon.stub(pgBossService, 'getPgBoss').returns(mockBoss as any);
 
-      const result = await publishMalwareScanJob({ artifactSecurityId: 'artifact-security-999' });
+      const result = await publishMalwareScanJob(getMockDBConnection(), {
+        artifactSecurityId: 'artifact-security-999'
+      });
 
       expect(result.status).to.equal('duplicate');
       expect((result as { status: 'duplicate'; message: string }).message).to.equal(
@@ -297,7 +314,9 @@ describe('publisher', () => {
     it('returns error status when pg-boss throws', async () => {
       sinon.stub(pgBossService, 'getPgBoss').throws(new Error('pg-boss not initialized'));
 
-      const result = await publishMalwareScanJob({ artifactSecurityId: 'artifact-security-000' });
+      const result = await publishMalwareScanJob(getMockDBConnection(), {
+        artifactSecurityId: 'artifact-security-000'
+      });
 
       expect(result.status).to.equal('error');
       expect((result as { status: 'error'; message: string }).message).to.equal('pg-boss not initialized');

api/src/queue/publisher.ts

@@ -182,11 +182,13 @@ export const publishProcessSubmissionFeaturesJob = async (
  *
  * Queues ClamAV scanning for an uploaded artifact.
  *
+ * @param {IDBConnection} connection Database connection for submission validation tracking
  * @param {IMalwareScanJobData} data Job data containing artifactSecurityId
  * @param {IPublishOptions} [options={}] Job options
  * @return {*}  {Promise<PublishJobResult>} Result indicating success, duplicate, or error
  */
 export const publishMalwareScanJob = async (
+  connection: IDBConnection,
   data: IMalwareScanJobData,
   options: IPublishOptions = {}
 ): Promise<PublishJobResult> => {
@@ -196,10 +198,18 @@ export const publishMalwareScanJob = async (
 
     await boss.createQueue(JobQueues.MALWARE_SCAN);
 
+    const db = {
+      executeSql: async (text: string, values: any[]) => {
+        const result = await connection.query(text, values);
+        return { rows: result.rows, rowCount: result.rowCount };
+      }
+    };
+
     // Use singletonKey to prevent duplicate concurrent jobs for the same artifact security record
     const jobId = await boss.send(JobQueues.MALWARE_SCAN, data, {
       ...mergedOptions,
-      singletonKey: `artifact-security-${data.artifactSecurityId}`
+      singletonKey: `artifact-security-${data.artifactSecurityId}`,
+      db
     });
 
     if (jobId) {

api/src/repositories/upload/submission-upload-repository.ts

@@ -189,7 +189,7 @@ export class SubmissionUploadRepository extends BaseRepository {
    * @param {string} uploadId - The upload_id to look up.
    * @returns {Promise<SubmissionUpload | null>} - The submission_upload record, or null if not found.
    */
-  async findSubmissionUploadByUploadId(uploadId: string): Promise<SubmissionUpload | null> {
+  async getSubmissionUploadByUploadId(uploadId: string): Promise<SubmissionUpload> {
     const sqlStatement = SQL`
       SELECT
         submission_upload_id,
@@ -202,7 +202,15 @@ export class SubmissionUploadRepository extends BaseRepository {
     `;
 
     const response = await this.connection.sql(sqlStatement, SubmissionUpload);
-    return response.rows[0] ?? null;
+
+    if (response.rowCount !== 1) {
+      throw new ApiExecuteSQLError('Failed to get submission upload record', [
+        'SubmissionUploadRepository->getSubmissionUploadByUploadId',
+        `rowCount was ${response.rowCount}, expected 1`
+      ]);
+    }
+
+    return response.rows[0];
   }
 
   /**

api/src/repositories/upload/upload-archive-repository.ts

@@ -85,7 +85,7 @@ export class UploadArchiveRepository extends BaseRepository {
    * @param {string} artifactId - The ID of the artifact.
    * @returns {Promise<UploadArchive | null>} - The upload archive or null if not found.
    */
-  async getUploadArchiveByArtifactId(artifactId: string): Promise<UploadArchive | null> {
+  async getUploadArchiveByArtifactId(artifactId: string): Promise<UploadArchive> {
     const sqlStatement = SQL`
       SELECT
         upload_archive_id,
@@ -99,7 +99,15 @@ export class UploadArchiveRepository extends BaseRepository {
     `;
 
     const response = await this.connection.sql(sqlStatement, UploadArchive);
-    return response.rows[0] ?? null;
+
+    if (response.rowCount !== 1) {
+      throw new ApiExecuteSQLError('Failed to get upload archive record', [
+        'UploadArchiveRepository->getUploadArchiveByArtifactId',
+        `rowCount was ${response.rowCount}, expected 1`
+      ]);
+    }
+
+    return response.rows[0];
   }
 
   /**