Optimize database reindex

packages/app/src/admin/SuperAdminPage.test.tsx

@@ -75,20 +75,6 @@ describe('SuperAdminPage', () => {
     expect(screen.getByText('Done')).toBeInTheDocument();
   });
 
-  test('Rebuild compartments for resource type', async () => {
-    setup();
-
-    await act(async () => {
-      fireEvent.change(screen.getByPlaceholderText('Compartments Resource Type'), { target: { value: 'Project' } });
-    });
-
-    await act(async () => {
-      fireEvent.click(screen.getByRole('button', { name: 'Rebuild Compartments' }));
-    });
-
-    expect(screen.getByText('Done')).toBeInTheDocument();
-  });
-
   test('Purge resources', async () => {
     setup();
 

packages/app/src/admin/SuperAdminPage.tsx

@@ -41,10 +41,6 @@ export function SuperAdminPage(): JSX.Element {
     startAsyncJob(medplum, 'Reindexing Resources', 'admin/super/reindex', formData);
   }
 
-  function rebuildCompartments(formData: Record<string, string>): void {
-    startAsyncJob(medplum, 'Rebuilding Compartments', 'admin/super/compartments', formData);
-  }
-
   function removeBotIdJobsFromQueue(formData: Record<string, string>): void {
     medplum
       .post('admin/super/removebotidjobsfromqueue', formData)
@@ -124,20 +120,6 @@ export function SuperAdminPage(): JSX.Element {
         </Stack>
       </Form>
       <Divider my="lg" />
-      <Title order={2}>Rebuild Compartments</Title>
-      <p>
-        When Medplum changes how resource compartments are derived, the system may require a rebuild for old resources
-        to take effect.
-      </p>
-      <Form onSubmit={rebuildCompartments}>
-        <Stack>
-          <FormSection title="Resource Type">
-            <TextInput name="resourceType" placeholder="Compartments Resource Type" />
-          </FormSection>
-          <Button type="submit">Rebuild Compartments</Button>
-        </Stack>
-      </Form>
-      <Divider my="lg" />
       <Title order={2}>Purge Resources</Title>
       <p>As system generated resources accumulate, the system may require a purge to remove old resources.</p>
       <Form onSubmit={purgeResources}>

packages/server/src/admin/super.test.ts

@@ -13,6 +13,7 @@ import { rebuildR4SearchParameters } from '../seeds/searchparameters';
 import { rebuildR4StructureDefinitions } from '../seeds/structuredefinitions';
 import { rebuildR4ValueSets } from '../seeds/valuesets';
 import { createTestProject, waitForAsyncJob, withTestContext } from '../test.setup';
+import { ReindexJobData, getReindexQueue } from '../workers/reindex';
 
 jest.mock('../seeds/valuesets');
 jest.mock('../seeds/structuredefinitions');
@@ -296,6 +297,9 @@ describe('Super Admin routes', () => {
   });
 
   test('Reindex with respond-async', async () => {
+    const queue = getReindexQueue() as any;
+    queue.add.mockClear();
+
     const res = await request(app)
       .post('/admin/super/reindex')
       .set('Authorization', 'Bearer ' + adminAccessToken)
@@ -307,58 +311,35 @@ describe('Super Admin routes', () => {
 
     expect(res.status).toEqual(202);
     expect(res.headers['content-location']).toBeDefined();
-    await waitForAsyncJob(res.headers['content-location'], app, adminAccessToken);
-  });
-
-  test('Rebuild compartments access denied', async () => {
-    const res = await request(app)
-      .post('/admin/super/compartments')
-      .set('Authorization', 'Bearer ' + nonAdminAccessToken)
-      .type('json')
-      .send({
-        resourceType: 'PaymentNotice',
-      });
-
-    expect(res.status).toBe(403);
-  });
-
-  test('Rebuild compartments require async', async () => {
-    const res = await request(app)
-      .post('/admin/super/compartments')
-      .set('Authorization', 'Bearer ' + adminAccessToken)
-      .type('json')
-      .send({
-        resourceType: 'PaymentNotice',
-      });
-
-    expect(res.status).toEqual(400);
-    expect(res.body.issue[0].details.text).toBe('Operation requires "Prefer: respond-async"');
+    expect(queue.add).toHaveBeenCalledWith(
+      'ReindexJobData',
+      expect.objectContaining<Partial<ReindexJobData>>({
+        resourceTypes: ['PaymentNotice'],
+      })
+    );
   });
 
-  test('Rebuild compartments invalid resource type', async () => {
-    const res = await request(app)
-      .post('/admin/super/compartments')
-      .set('Authorization', 'Bearer ' + adminAccessToken)
-      .type('json')
-      .send({
-        resourceType: 'XYZ',
-      });
-
-    expect(res.status).toBe(400);
-  });
+  test('Reindex with multiple resource types', async () => {
+    const queue = getReindexQueue() as any;
+    queue.add.mockClear();
 
-  test('Rebuild compartments with respond-async', async () => {
     const res = await request(app)
-      .post('/admin/super/compartments')
+      .post('/admin/super/reindex')
       .set('Authorization', 'Bearer ' + adminAccessToken)
       .set('Prefer', 'respond-async')
       .type('json')
       .send({
-        resourceType: 'PaymentNotice',
+        resourceType: 'PaymentNotice, MedicinalProductManufactured,BiologicallyDerivedProduct',
       });
 
     expect(res.status).toEqual(202);
     expect(res.headers['content-location']).toBeDefined();
+    expect(queue.add).toHaveBeenCalledWith(
+      'ReindexJobData',
+      expect.objectContaining<Partial<ReindexJobData>>({
+        resourceTypes: ['PaymentNotice', 'MedicinalProductManufactured', 'BiologicallyDerivedProduct'],
+      })
+    );
   });
 
   test('Set password access denied', async () => {

packages/server/src/admin/super.ts

@@ -1,4 +1,5 @@
 import {
+  accepted,
   allOk,
   badRequest,
   forbidden,
@@ -12,7 +13,7 @@ import { asyncWrap } from '../async';
 import { setPassword } from '../auth/setpassword';
 import { AuthenticatedRequestContext, getAuthenticatedContext } from '../context';
 import { getDatabasePool } from '../database';
-import { sendAsyncResponse } from '../fhir/operations/utils/asyncjobexecutor';
+import { AsyncJobExecutor, sendAsyncResponse } from '../fhir/operations/utils/asyncjobexecutor';
 import { invalidRequest, sendOutcome } from '../fhir/outcomes';
 import { getSystemRepo } from '../fhir/repo';
 import * as dataMigrations from '../migrations/data';
@@ -22,6 +23,9 @@ import { rebuildR4SearchParameters } from '../seeds/searchparameters';
 import { rebuildR4StructureDefinitions } from '../seeds/structuredefinitions';
 import { rebuildR4ValueSets } from '../seeds/valuesets';
 import { removeBullMQJobByKey } from '../workers/cron';
+import { ResourceType } from '@medplum/fhirtypes';
+import { addReindexJob } from '../workers/reindex';
+import { getConfig } from '../config';
 
 export const superAdminRouter = Router();
 superAdminRouter.use(authenticateRequest);
@@ -74,32 +78,20 @@ superAdminRouter.post(
     requireSuperAdmin();
     requireAsync(req);
 
-    const resourceType = req.body.resourceType;
-    validateResourceType(resourceType);
+    const resourceTypes = (req.body.resourceType as string).split(',').map((t) => t.trim());
+    for (const resourceType of resourceTypes) {
+      validateResourceType(resourceType);
+    }
+    const systemRepo = getSystemRepo();
 
-    await sendAsyncResponse(req, res, async () => {
-      const systemRepo = getSystemRepo();
-      await systemRepo.reindexResourceType(resourceType);
+    const exec = new AsyncJobExecutor(systemRepo);
+    await exec.init(`${req.protocol}://${req.get('host') + req.originalUrl}`);
+    await exec.run(async (asyncJob) => {
+      await addReindexJob(resourceTypes as ResourceType[], asyncJob);
     });
-  })
-);
-
-// POST to /admin/super/compartments
-// to rebuild compartments for a resource type.
-// Run this after major changes to how compartments are constructed.
-superAdminRouter.post(
-  '/compartments',
-  asyncWrap(async (req: Request, res: Response) => {
-    requireSuperAdmin();
-    requireAsync(req);
-
-    const resourceType = req.body.resourceType;
-    validateResourceType(resourceType);
 
-    await sendAsyncResponse(req, res, async () => {
-      const systemRepo = getSystemRepo();
-      await systemRepo.rebuildCompartmentsForResourceType(resourceType);
-    });
+    const { baseUrl } = getConfig();
+    sendOutcome(res, accepted(exec.getContentLocation(baseUrl)));
   })
 );
 

packages/server/src/fhir/job.test.ts

@@ -1,10 +1,9 @@
-import { AsyncJob } from '@medplum/fhirtypes';
 import { randomUUID } from 'crypto';
 import express from 'express';
 import request from 'supertest';
 import { initApp, shutdownApp } from '../app';
 import { loadTestConfig } from '../config';
-import { createTestProject, withTestContext } from '../test.setup';
+import { createTestProject, waitForAsyncJob, withTestContext } from '../test.setup';
 import { AsyncJobExecutor } from './operations/utils/asyncjobexecutor';
 import { Repository } from './repo';
 
@@ -47,24 +46,16 @@ describe('Job status', () => {
 
   test('completed', () =>
     withTestContext(async () => {
-      const job = await asyncJobManager.init('http://example.com');
+      await asyncJobManager.init('http://example.com');
       const callback = jest.fn();
 
-      await asyncJobManager.run(async () => {
+      await asyncJobManager.start(async () => {
         callback();
       });
 
       expect(callback).toHaveBeenCalled();
 
-      const resCompleted = await request(app)
-        .get(`/fhir/R4/job/${job.id}/status`)
-        .set('Authorization', 'Bearer ' + accessToken);
-
-      expect(resCompleted.status).toBe(200);
-      expect(resCompleted.body).toMatchObject<Partial<AsyncJob>>({
-        resourceType: 'AsyncJob',
-        status: 'completed',
-      });
+      await waitForAsyncJob(asyncJobManager.getContentLocation('http://example.com/'), app, accessToken);
     }));
 
   test('cancel', () =>