SIMSBIOHUB-883: Parse Feature Content into Search Tables (#341)

* feat(queue): add index-submission-features job with idempotent indexing

* feat(queue): chain indexing job from validation pipeline

* fix(test): clean up search_ tables

Author: dwhogan

api/src/__integration__/db/search-feature-repository.integration.ts

@@ -0,0 +1,180 @@
+// Integration test for SearchFeatureRepository.deleteSearchRecordsBySubmissionId —
+// verifies the subquery-based delete correctly removes records from all 4 search tables
+// for a given submission, without affecting records belonging to other submissions.
+//
+// Uses a transaction that is ROLLED BACK after each test, so no data is persisted.
+//
+// Run: make test-db
+// Requires: make web (database must be running with seed data)
+
+import { expect } from 'chai';
+import SQL from 'sql-template-strings';
+import { defaultPoolConfig, getAPIUserDBConnection, IDBConnection, initDBPool } from '../../database/db';
+import { SearchFeatureRepository } from '../../repositories/search-feature-repository';
+
+describe('SearchFeatureRepository (integration)', function () {
+  this.timeout(15000);
+
+  let connection: IDBConnection;
+  let repo: SearchFeatureRepository;
+
+  before(() => {
+    initDBPool(defaultPoolConfig);
+  });
+
+  beforeEach(async () => {
+    connection = getAPIUserDBConnection();
+    await connection.open();
+    repo = new SearchFeatureRepository(connection);
+  });
+
+  afterEach(async () => {
+    await connection.rollback();
+    connection.release();
+  });
+
+  /**
+   * Helper: insert a minimal submission and return its ID.
+   */
+  async function createTestSubmission(): Promise<number> {
+    const systemUserId = connection.systemUserId();
+
+    const result = await connection.sql(SQL`
+      INSERT INTO submission (uuid, system_user_id, source_system, name, description, comment, create_user)
+      VALUES (gen_random_uuid(), ${systemUserId}, 'SIMS', 'Search Index Test', 'Test', 'Test', ${systemUserId})
+      RETURNING submission_id;
+    `);
+
+    return result.rows[0].submission_id;
+  }
+
+  /**
+   * Helper: insert a submission_feature and return its ID.
+   */
+  async function createTestFeature(submissionId: number): Promise<number> {
+    const systemUserId = connection.systemUserId();
+
+    const result = await connection.sql(SQL`
+      INSERT INTO submission_feature (submission_id, feature_type_id, data, data_byte_size, create_user)
+      VALUES (
+        ${submissionId},
+        (SELECT feature_type_id FROM feature_type WHERE name = 'dataset' LIMIT 1),
+        '{"name": "test"}'::jsonb,
+        100,
+        ${systemUserId}
+      )
+      RETURNING submission_feature_id;
+    `);
+
+    return result.rows[0].submission_feature_id;
+  }
+
+  /**
+   * Helper: insert a search_string record.
+   */
+  async function insertSearchString(submissionFeatureId: number, featurePropertyId: number, value: string) {
+    const systemUserId = connection.systemUserId();
+
+    await connection.sql(SQL`
+      INSERT INTO search_string (submission_feature_id, feature_property_id, value, create_user)
+      VALUES (${submissionFeatureId}, ${featurePropertyId}, ${value}, ${systemUserId});
+    `);
+  }
+
+  /**
+   * Helper: insert a search_number record.
+   */
+  async function insertSearchNumber(submissionFeatureId: number, featurePropertyId: number, value: number) {
+    const systemUserId = connection.systemUserId();
+
+    await connection.sql(SQL`
+      INSERT INTO search_number (submission_feature_id, feature_property_id, value, create_user)
+      VALUES (${submissionFeatureId}, ${featurePropertyId}, ${value}, ${systemUserId});
+    `);
+  }
+
+  /**
+   * Helper: count records in a search table for a given submission_feature_id.
+   */
+  async function countSearchRecords(table: string, submissionFeatureId: number): Promise<number> {
+    const result = await connection.sql(
+      SQL`
+      SELECT count(*)::integer as count
+      FROM `.append(table).append(SQL`
+      WHERE submission_feature_id = ${submissionFeatureId};
+    `)
+    );
+
+    return result.rows[0].count;
+  }
+
+  describe('deleteSearchRecordsBySubmissionId', () => {
+    it('deletes search_string and search_number records for the given submission', async () => {
+      // Arrange: create submission with a feature and search records
+      const submissionId = await createTestSubmission();
+      const featureId = await createTestFeature(submissionId);
+
+      // feature_property_id 1 = site_identifier (string type), 6 = measurement_value (number type)
+      await insertSearchString(featureId, 1, 'test-site');
+      await insertSearchNumber(featureId, 6, 42);
+
+      // Verify records exist
+      expect(await countSearchRecords('search_string', featureId)).to.equal(1);
+      expect(await countSearchRecords('search_number', featureId)).to.equal(1);
+
+      // Act
+      await repo.deleteSearchRecordsBySubmissionId(submissionId);
+
+      // Assert: all records deleted
+      expect(await countSearchRecords('search_string', featureId)).to.equal(0);
+      expect(await countSearchRecords('search_number', featureId)).to.equal(0);
+    });
+
+    it('does not delete records belonging to a different submission', async () => {
+      // Arrange: create two submissions with search records
+      const submissionId1 = await createTestSubmission();
+      const featureId1 = await createTestFeature(submissionId1);
+      await insertSearchString(featureId1, 1, 'submission-1-value');
+
+      const submissionId2 = await createTestSubmission();
+      const featureId2 = await createTestFeature(submissionId2);
+      await insertSearchString(featureId2, 1, 'submission-2-value');
+
+      // Act: delete only submission 1's records
+      await repo.deleteSearchRecordsBySubmissionId(submissionId1);
+
+      // Assert: submission 1 records gone, submission 2 records intact
+      expect(await countSearchRecords('search_string', featureId1)).to.equal(0);
+      expect(await countSearchRecords('search_string', featureId2)).to.equal(1);
+    });
+
+    it('succeeds with no error when submission has no search records', async () => {
+      const submissionId = await createTestSubmission();
+      await createTestFeature(submissionId);
+
+      // Act: should not throw
+      await repo.deleteSearchRecordsBySubmissionId(submissionId);
+    });
+
+    it('deletes records across multiple features in the same submission', async () => {
+      // Arrange: one submission, two features, each with search records
+      const submissionId = await createTestSubmission();
+      const featureId1 = await createTestFeature(submissionId);
+      const featureId2 = await createTestFeature(submissionId);
+
+      await insertSearchString(featureId1, 1, 'feature-1-value');
+      await insertSearchString(featureId2, 1, 'feature-2-value');
+      await insertSearchNumber(featureId1, 6, 10);
+      await insertSearchNumber(featureId2, 6, 20);
+
+      // Act
+      await repo.deleteSearchRecordsBySubmissionId(submissionId);
+
+      // Assert: all records across both features deleted
+      expect(await countSearchRecords('search_string', featureId1)).to.equal(0);
+      expect(await countSearchRecords('search_string', featureId2)).to.equal(0);
+      expect(await countSearchRecords('search_number', featureId1)).to.equal(0);
+      expect(await countSearchRecords('search_number', featureId2)).to.equal(0);
+    });
+  });
+});

api/src/__integration__/system/process-submission-features-worker.integration.ts

@@ -89,6 +89,17 @@ describe('Process Submission Features Worker', function () {
     try {
       // Clean up in reverse FK order
       for (const submissionId of createdSubmissionIds) {
+        // search_ tables FK to submission_feature — delete first
+        const featureIds = await db('biohub.submission_feature')
+          .where('submission_id', submissionId)
+          .select('submission_feature_id');
+        const ids = featureIds.map((r: { submission_feature_id: number }) => r.submission_feature_id);
+        if (ids.length) {
+          await db('biohub.search_string').whereIn('submission_feature_id', ids).del();
+          await db('biohub.search_number').whereIn('submission_feature_id', ids).del();
+          await db('biohub.search_datetime').whereIn('submission_feature_id', ids).del();
+          await db('biohub.search_spatial').whereIn('submission_feature_id', ids).del();
+        }
         await db('biohub.submission_feature').where('submission_id', submissionId).del();
         await db('biohub.submission_validation').where('submission_id', submissionId).del();
         await db('biohub.submission_upload').where('submission_id', submissionId).del();

api/src/queue/jobs/index-submission-features-job.test.ts

@@ -0,0 +1,139 @@
+import chai, { expect } from 'chai';
+import { describe } from 'mocha';
+import PgBoss from 'pg-boss';
+import sinon from 'sinon';
+import sinonChai from 'sinon-chai';
+import * as db from '../../database/db';
+import { SearchFeatureService } from '../../services/search-feature-service';
+import { getMockDBConnection } from '../../__mocks__/db';
+import {
+  IIndexSubmissionFeaturesJobData,
+  indexSubmissionFeaturesFailedHandler,
+  indexSubmissionFeaturesJobHandler
+} from './index-submission-features-job';
+
+chai.use(sinonChai);
+
+describe('indexSubmissionFeaturesJobHandler', () => {
+  afterEach(() => {
+    sinon.restore();
+  });
+
+  const createMockJob = (submissionId: number, id = 'job-1'): PgBoss.Job<IIndexSubmissionFeaturesJobData> =>
+    ({
+      id,
+      name: 'index-submission-features',
+      data: { submissionId }
+    } as PgBoss.Job<IIndexSubmissionFeaturesJobData>);
+
+  it('should index submission successfully', async () => {
+    const mockDBConnection = getMockDBConnection();
+    const openStub = sinon.stub().resolves();
+    const commitStub = sinon.stub().resolves();
+    const releaseStub = sinon.stub();
+    mockDBConnection.open = openStub;
+    mockDBConnection.commit = commitStub;
+    mockDBConnection.release = releaseStub;
+
+    sinon.stub(db, 'getAPIUserDBConnection').returns(mockDBConnection);
+
+    const indexStub = sinon.stub(SearchFeatureService.prototype, 'indexFeaturesBySubmissionId').resolves();
+
+    await indexSubmissionFeaturesJobHandler([createMockJob(777)]);
+
+    expect(indexStub).to.have.been.calledOnceWith(777);
+    expect(commitStub).to.have.been.calledOnce;
+    expect(releaseStub).to.have.been.calledOnce;
+  });
+
+  it('should roll back and throw on indexing failure', async () => {
+    const mockDBConnection = getMockDBConnection();
+    const rollbackStub = sinon.stub().resolves();
+    const releaseStub = sinon.stub();
+    mockDBConnection.open = sinon.stub().resolves();
+    mockDBConnection.rollback = rollbackStub;
+    mockDBConnection.release = releaseStub;
+
+    sinon.stub(db, 'getAPIUserDBConnection').returns(mockDBConnection);
+
+    const testError = new Error('Indexing failed');
+    sinon.stub(SearchFeatureService.prototype, 'indexFeaturesBySubmissionId').rejects(testError);
+
+    try {
+      await indexSubmissionFeaturesJobHandler([createMockJob(777)]);
+      expect.fail('Should have thrown an error');
+    } catch (error) {
+      expect((error as Error).message).to.equal('Indexing failed');
+    }
+
+    expect(rollbackStub).to.have.been.calledOnce;
+    expect(releaseStub).to.have.been.calledOnce;
+  });
+
+  it('should process multiple jobs in sequence', async () => {
+    const openStub = sinon.stub().resolves();
+    const commitStub = sinon.stub().resolves();
+    const releaseStub = sinon.stub();
+
+    const mockDBConnection = getMockDBConnection();
+    mockDBConnection.open = openStub;
+    mockDBConnection.commit = commitStub;
+    mockDBConnection.release = releaseStub;
+
+    sinon.stub(db, 'getAPIUserDBConnection').returns(mockDBConnection);
+
+    const indexStub = sinon.stub(SearchFeatureService.prototype, 'indexFeaturesBySubmissionId').resolves();
+
+    await indexSubmissionFeaturesJobHandler([createMockJob(1, 'job-1'), createMockJob(2, 'job-2')]);
+
+    expect(indexStub.callCount).to.equal(2);
+    expect(openStub.callCount).to.equal(2);
+    expect(commitStub.callCount).to.equal(2);
+    expect(releaseStub.callCount).to.equal(2);
+  });
+
+  it('should handle empty jobs array', async () => {
+    const getConnectionStub = sinon.stub(db, 'getAPIUserDBConnection');
+
+    await indexSubmissionFeaturesJobHandler([]);
+
+    expect(getConnectionStub).not.to.have.been.called;
+  });
+});
+
+describe('indexSubmissionFeaturesFailedHandler', () => {
+  afterEach(() => {
+    sinon.restore();
+  });
+
+  it('should log failure with error output without throwing', async () => {
+    const getConnectionStub = sinon.stub(db, 'getAPIUserDBConnection');
+
+    const job = {
+      id: 'job-1',
+      name: 'index-submission-features-failed',
+      data: { submissionId: 777 },
+      output: { message: 'Indexing failed after retries' }
+    } as unknown as PgBoss.Job<IIndexSubmissionFeaturesJobData>;
+
+    await indexSubmissionFeaturesFailedHandler([job]);
+
+    // DLQ handler is log-only — no DB connection should be opened
+    expect(getConnectionStub).not.to.have.been.called;
+  });
+
+  it('should log default message when output is null', async () => {
+    const getConnectionStub = sinon.stub(db, 'getAPIUserDBConnection');
+
+    const job = {
+      id: 'job-2',
+      name: 'index-submission-features-failed',
+      data: { submissionId: 888 },
+      output: null
+    } as unknown as PgBoss.Job<IIndexSubmissionFeaturesJobData>;
+
+    await indexSubmissionFeaturesFailedHandler([job]);
+
+    expect(getConnectionStub).not.to.have.been.called;
+  });
+});