feat(core): Create async job manager for long-running tasks

BREAKING CHANGE: The `reindex` mutation now returns a JobInfo type, which has an id that can then be polled via the new `job` query as to its progress and status.

Relates to #111

packages/common/src/generated-types.ts

@@ -1021,6 +1021,29 @@ export type ImportInfo = {
   imported: Scalars['Int'],
 };
 
+export type JobInfo = {
+  id: Scalars['String'],
+  name: Scalars['String'],
+  state: JobState,
+  progress: Scalars['Float'],
+  result?: Maybe<Scalars['JSON']>,
+  started?: Maybe<Scalars['DateTime']>,
+  ended?: Maybe<Scalars['DateTime']>,
+  duration?: Maybe<Scalars['Int']>,
+};
+
+export type JobListInput = {
+  state?: Maybe<JobState>,
+  ids?: Maybe<Array<Scalars['String']>>,
+};
+
+export enum JobState {
+  PENDING = 'PENDING',
+  RUNNING = 'RUNNING',
+  COMPLETED = 'COMPLETED',
+  FAILED = 'FAILED'
+}
+
 
 /** ISO 639-1 language code */
 export enum LanguageCode {
@@ -1419,12 +1442,6 @@ export type Mutation = {
   createChannel: Channel,
   /** Update an existing Channel */
   updateChannel: Channel,
-  /** Create a new Collection */
-  createCollection: Collection,
-  /** Update an existing Collection */
-  updateCollection: Collection,
-  /** Move a Collection to a different parent or index */
-  moveCollection: Collection,
   /** Create a new Country */
   createCountry: Country,
   /** Update an existing Country */
@@ -1439,18 +1456,12 @@ export type Mutation = {
   addCustomersToGroup: CustomerGroup,
   /** Remove Customers from a CustomerGroup */
   removeCustomersFromGroup: CustomerGroup,
-  /** Create a new Customer. If a password is provided, a new User will also be created an linked to the Customer. */
-  createCustomer: Customer,
-  /** Update an existing Customer */
-  updateCustomer: Customer,
-  /** Delete a Customer */
-  deleteCustomer: DeletionResponse,
-  /** Create a new Address and associate it with the Customer specified by customerId */
-  createCustomerAddress: Address,
-  /** Update an existing Address */
-  updateCustomerAddress: Address,
-  /** Update an existing Address */
-  deleteCustomerAddress: Scalars['Boolean'],
+  /** Create a new Collection */
+  createCollection: Collection,
+  /** Update an existing Collection */
+  updateCollection: Collection,
+  /** Move a Collection to a different parent or index */
+  moveCollection: Collection,
   /** Create a new Facet */
   createFacet: Facet,
   /** Update an existing Facet */
@@ -1464,6 +1475,18 @@ export type Mutation = {
   /** Delete one or more FacetValues */
   deleteFacetValues: Array<DeletionResponse>,
   updateGlobalSettings: GlobalSettings,
+  /** Create a new Customer. If a password is provided, a new User will also be created an linked to the Customer. */
+  createCustomer: Customer,
+  /** Update an existing Customer */
+  updateCustomer: Customer,
+  /** Delete a Customer */
+  deleteCustomer: DeletionResponse,
+  /** Create a new Address and associate it with the Customer specified by customerId */
+  createCustomerAddress: Address,
+  /** Update an existing Address */
+  updateCustomerAddress: Address,
+  /** Update an existing Address */
+  deleteCustomerAddress: Scalars['Boolean'],
   importProducts?: Maybe<ImportInfo>,
   /** Update an existing PaymentMethod */
   updatePaymentMethod: PaymentMethod,
@@ -1471,7 +1494,7 @@ export type Mutation = {
   createProductOptionGroup: ProductOptionGroup,
   /** Update an existing ProductOptionGroup */
   updateProductOptionGroup: ProductOptionGroup,
-  reindex: SearchReindexResponse,
+  reindex: JobInfo,
   /** Create a new Product */
   createProduct: Product,
   /** Update an existing Product */
@@ -1556,21 +1579,6 @@ export type MutationUpdateChannelArgs = {
 };
 
 
-export type MutationCreateCollectionArgs = {
-  input: CreateCollectionInput
-};
-
-
-export type MutationUpdateCollectionArgs = {
-  input: UpdateCollectionInput
-};
-
-
-export type MutationMoveCollectionArgs = {
-  input: MoveCollectionInput
-};
-
-
 export type MutationCreateCountryArgs = {
   input: CreateCountryInput
 };
@@ -1608,35 +1616,18 @@ export type MutationRemoveCustomersFromGroupArgs = {
 };
 
 
-export type MutationCreateCustomerArgs = {
-  input: CreateCustomerInput,
-  password?: Maybe<Scalars['String']>
-};
-
-
-export type MutationUpdateCustomerArgs = {
-  input: UpdateCustomerInput
-};
-
-
-export type MutationDeleteCustomerArgs = {
-  id: Scalars['ID']
-};
-
-
-export type MutationCreateCustomerAddressArgs = {
-  customerId: Scalars['ID'],
-  input: CreateAddressInput
+export type MutationCreateCollectionArgs = {
+  input: CreateCollectionInput
 };
 
 
-export type MutationUpdateCustomerAddressArgs = {
-  input: UpdateAddressInput
+export type MutationUpdateCollectionArgs = {
+  input: UpdateCollectionInput
 };
 
 
-export type MutationDeleteCustomerAddressArgs = {
-  id: Scalars['ID']
+export type MutationMoveCollectionArgs = {
+  input: MoveCollectionInput
 };
 
 
@@ -1677,6 +1668,38 @@ export type MutationUpdateGlobalSettingsArgs = {
 };
 
 
+export type MutationCreateCustomerArgs = {
+  input: CreateCustomerInput,
+  password?: Maybe<Scalars['String']>
+};
+
+
+export type MutationUpdateCustomerArgs = {
+  input: UpdateCustomerInput
+};
+
+
+export type MutationDeleteCustomerArgs = {
+  id: Scalars['ID']
+};
+
+
+export type MutationCreateCustomerAddressArgs = {
+  customerId: Scalars['ID'],
+  input: CreateAddressInput
+};
+
+
+export type MutationUpdateCustomerAddressArgs = {
+  input: UpdateAddressInput
+};
+
+
+export type MutationDeleteCustomerAddressArgs = {
+  id: Scalars['ID']
+};
+
+
 export type MutationImportProductsArgs = {
   csvFile: Scalars['Upload']
 };
@@ -2281,20 +2304,22 @@ export type Query = {
   channels: Array<Channel>,
   channel?: Maybe<Channel>,
   activeChannel: Channel,
-  collections: CollectionList,
-  collection?: Maybe<Collection>,
-  collectionFilters: Array<ConfigurableOperation>,
   countries: CountryList,
   country?: Maybe<Country>,
   customerGroups: Array<CustomerGroup>,
   customerGroup?: Maybe<CustomerGroup>,
-  customers: CustomerList,
-  customer?: Maybe<Customer>,
+  collections: CollectionList,
+  collection?: Maybe<Collection>,
+  collectionFilters: Array<ConfigurableOperation>,
   facets: FacetList,
   facet?: Maybe<Facet>,
   globalSettings: GlobalSettings,
+  customers: CustomerList,
+  customer?: Maybe<Customer>,
   order?: Maybe<Order>,
   orders: OrderList,
+  job?: Maybe<JobInfo>,
+  jobs: Array<JobInfo>,
   paymentMethods: PaymentMethodList,
   paymentMethod?: Maybe<PaymentMethod>,
   productOptionGroups: Array<ProductOptionGroup>,
@@ -2347,18 +2372,6 @@ export type QueryChannelArgs = {
 };
 
 
-export type QueryCollectionsArgs = {
-  languageCode?: Maybe<LanguageCode>,
-  options?: Maybe<CollectionListOptions>
-};
-
-
-export type QueryCollectionArgs = {
-  id: Scalars['ID'],
-  languageCode?: Maybe<LanguageCode>
-};
-
-
 export type QueryCountriesArgs = {
   options?: Maybe<CountryListOptions>
 };
@@ -2374,13 +2387,15 @@ export type QueryCustomerGroupArgs = {
 };
 
 
-export type QueryCustomersArgs = {
-  options?: Maybe<CustomerListOptions>
+export type QueryCollectionsArgs = {
+  languageCode?: Maybe<LanguageCode>,
+  options?: Maybe<CollectionListOptions>
 };
 
 
-export type QueryCustomerArgs = {
-  id: Scalars['ID']
+export type QueryCollectionArgs = {
+  id: Scalars['ID'],
+  languageCode?: Maybe<LanguageCode>
 };
 
 
@@ -2396,6 +2411,16 @@ export type QueryFacetArgs = {
 };
 
 
+export type QueryCustomersArgs = {
+  options?: Maybe<CustomerListOptions>
+};
+
+
+export type QueryCustomerArgs = {
+  id: Scalars['ID']
+};
+
+
 export type QueryOrderArgs = {
   id: Scalars['ID']
 };
@@ -2406,6 +2431,16 @@ export type QueryOrdersArgs = {
 };
 
 
+export type QueryJobArgs = {
+  jobId: Scalars['String']
+};
+
+
+export type QueryJobsArgs = {
+  input?: Maybe<JobListInput>
+};
+
+
 export type QueryPaymentMethodsArgs = {
   options?: Maybe<PaymentMethodListOptions>
 };

packages/core/e2e/graphql/generated-e2e-admin-types.ts

@@ -1019,6 +1019,29 @@ export type ImportInfo = {
     imported: Scalars['Int'];
 };
 
+export type JobInfo = {
+    id: Scalars['String'];
+    name: Scalars['String'];
+    state: JobState;
+    progress: Scalars['Float'];
+    result?: Maybe<Scalars['JSON']>;
+    started?: Maybe<Scalars['DateTime']>;
+    ended?: Maybe<Scalars['DateTime']>;
+    duration?: Maybe<Scalars['Int']>;
+};
+
+export type JobListInput = {
+    state?: Maybe<JobState>;
+    ids?: Maybe<Array<Scalars['String']>>;
+};
+
+export enum JobState {
+    PENDING = 'PENDING',
+    RUNNING = 'RUNNING',
+    COMPLETED = 'COMPLETED',
+    FAILED = 'FAILED',
+}
+
 /** ISO 639-1 language code */
 export enum LanguageCode {
     /** Afar */
@@ -1416,12 +1439,6 @@ export type Mutation = {
     createChannel: Channel;
     /** Update an existing Channel */
     updateChannel: Channel;
-    /** Create a new Collection */
-    createCollection: Collection;
-    /** Update an existing Collection */
-    updateCollection: Collection;
-    /** Move a Collection to a different parent or index */
-    moveCollection: Collection;
     /** Create a new Country */
     createCountry: Country;
     /** Update an existing Country */
@@ -1436,18 +1453,12 @@ export type Mutation = {
     addCustomersToGroup: CustomerGroup;
     /** Remove Customers from a CustomerGroup */
     removeCustomersFromGroup: CustomerGroup;
-    /** Create a new Customer. If a password is provided, a new User will also be created an linked to the Customer. */
-    createCustomer: Customer;
-    /** Update an existing Customer */
-    updateCustomer: Customer;
-    /** Delete a Customer */
-    deleteCustomer: DeletionResponse;
-    /** Create a new Address and associate it with the Customer specified by customerId */
-    createCustomerAddress: Address;
-    /** Update an existing Address */
-    updateCustomerAddress: Address;
-    /** Update an existing Address */
-    deleteCustomerAddress: Scalars['Boolean'];
+    /** Create a new Collection */
+    createCollection: Collection;
+    /** Update an existing Collection */
+    updateCollection: Collection;
+    /** Move a Collection to a different parent or index */
+    moveCollection: Collection;
     /** Create a new Facet */
     createFacet: Facet;
     /** Update an existing Facet */
@@ -1461,6 +1472,18 @@ export type Mutation = {
     /** Delete one or more FacetValues */
     deleteFacetValues: Array<DeletionResponse>;
     updateGlobalSettings: GlobalSettings;
+    /** Create a new Customer. If a password is provided, a new User will also be created an linked to the Customer. */
+    createCustomer: Customer;
+    /** Update an existing Customer */
+    updateCustomer: Customer;
+    /** Delete a Customer */
+    deleteCustomer: DeletionResponse;
+    /** Create a new Address and associate it with the Customer specified by customerId */
+    createCustomerAddress: Address;
+    /** Update an existing Address */
+    updateCustomerAddress: Address;
+    /** Update an existing Address */
+    deleteCustomerAddress: Scalars['Boolean'];
     importProducts?: Maybe<ImportInfo>;
     /** Update an existing PaymentMethod */
     updatePaymentMethod: PaymentMethod;
@@ -1468,7 +1491,7 @@ export type Mutation = {
     createProductOptionGroup: ProductOptionGroup;
     /** Update an existing ProductOptionGroup */
     updateProductOptionGroup: ProductOptionGroup;
-    reindex: SearchReindexResponse;
+    reindex: JobInfo;
     /** Create a new Product */
     createProduct: Product;
     /** Update an existing Product */
@@ -1545,18 +1568,6 @@ export type MutationUpdateChannelArgs = {
     input: UpdateChannelInput;
 };
 
-export type MutationCreateCollectionArgs = {
-    input: CreateCollectionInput;
-};
-
-export type MutationUpdateCollectionArgs = {
-    input: UpdateCollectionInput;
-};
-
-export type MutationMoveCollectionArgs = {
-    input: MoveCollectionInput;
-};
-
 export type MutationCreateCountryArgs = {
     input: CreateCountryInput;
 };
@@ -1587,30 +1598,16 @@ export type MutationRemoveCustomersFromGroupArgs = {
     customerIds: Array<Scalars['ID']>;
 };
 
-export type MutationCreateCustomerArgs = {
-    input: CreateCustomerInput;
-    password?: Maybe<Scalars['String']>;
-};
-
-export type MutationUpdateCustomerArgs = {
-    input: UpdateCustomerInput;
-};
-
-export type MutationDeleteCustomerArgs = {
-    id: Scalars['ID'];
-};
-
-export type MutationCreateCustomerAddressArgs = {
-    customerId: Scalars['ID'];
-    input: CreateAddressInput;
+export type MutationCreateCollectionArgs = {
+    input: CreateCollectionInput;
 };
 
-export type MutationUpdateCustomerAddressArgs = {
-    input: UpdateAddressInput;
+export type MutationUpdateCollectionArgs = {
+    input: UpdateCollectionInput;
 };
 
-export type MutationDeleteCustomerAddressArgs = {
-    id: Scalars['ID'];
+export type MutationMoveCollectionArgs = {
+    input: MoveCollectionInput;
 };
 
 export type MutationCreateFacetArgs = {
@@ -1643,6 +1640,32 @@ export type MutationUpdateGlobalSettingsArgs = {
     input: UpdateGlobalSettingsInput;
 };
 
+export type MutationCreateCustomerArgs = {
+    input: CreateCustomerInput;
+    password?: Maybe<Scalars['String']>;
+};
+
+export type MutationUpdateCustomerArgs = {
+    input: UpdateCustomerInput;
+};
+
+export type MutationDeleteCustomerArgs = {
+    id: Scalars['ID'];
+};
+
+export type MutationCreateCustomerAddressArgs = {
+    customerId: Scalars['ID'];
+    input: CreateAddressInput;
+};
+
+export type MutationUpdateCustomerAddressArgs = {
+    input: UpdateAddressInput;
+};
+
+export type MutationDeleteCustomerAddressArgs = {
+    id: Scalars['ID'];
+};
+
 export type MutationImportProductsArgs = {
     csvFile: Scalars['Upload'];
 };
@@ -2220,20 +2243,22 @@ export type Query = {
     channels: Array<Channel>;
     channel?: Maybe<Channel>;
     activeChannel: Channel;
-    collections: CollectionList;
-    collection?: Maybe<Collection>;
-    collectionFilters: Array<ConfigurableOperation>;
     countries: CountryList;
     country?: Maybe<Country>;
     customerGroups: Array<CustomerGroup>;
     customerGroup?: Maybe<CustomerGroup>;
-    customers: CustomerList;
-    customer?: Maybe<Customer>;
+    collections: CollectionList;
+    collection?: Maybe<Collection>;
+    collectionFilters: Array<ConfigurableOperation>;
     facets: FacetList;
     facet?: Maybe<Facet>;
     globalSettings: GlobalSettings;
+    customers: CustomerList;
+    customer?: Maybe<Customer>;
     order?: Maybe<Order>;
     orders: OrderList;
+    job?: Maybe<JobInfo>;
+    jobs: Array<JobInfo>;
     paymentMethods: PaymentMethodList;
     paymentMethod?: Maybe<PaymentMethod>;
     productOptionGroups: Array<ProductOptionGroup>;
@@ -2280,16 +2305,6 @@ export type QueryChannelArgs = {
     id: Scalars['ID'];
 };
 
-export type QueryCollectionsArgs = {
-    languageCode?: Maybe<LanguageCode>;
-    options?: Maybe<CollectionListOptions>;
-};
-
-export type QueryCollectionArgs = {
-    id: Scalars['ID'];
-    languageCode?: Maybe<LanguageCode>;
-};
-
 export type QueryCountriesArgs = {
     options?: Maybe<CountryListOptions>;
 };
@@ -2302,12 +2317,14 @@ export type QueryCustomerGroupArgs = {
     id: Scalars['ID'];
 };
 
-export type QueryCustomersArgs = {
-    options?: Maybe<CustomerListOptions>;
+export type QueryCollectionsArgs = {
+    languageCode?: Maybe<LanguageCode>;
+    options?: Maybe<CollectionListOptions>;
 };
 
-export type QueryCustomerArgs = {
+export type QueryCollectionArgs = {
     id: Scalars['ID'];
+    languageCode?: Maybe<LanguageCode>;
 };
 
 export type QueryFacetsArgs = {
@@ -2320,6 +2337,14 @@ export type QueryFacetArgs = {
     languageCode?: Maybe<LanguageCode>;
 };
 
+export type QueryCustomersArgs = {
+    options?: Maybe<CustomerListOptions>;
+};
+
+export type QueryCustomerArgs = {
+    id: Scalars['ID'];
+};
+
 export type QueryOrderArgs = {
     id: Scalars['ID'];
 };
@@ -2328,6 +2353,14 @@ export type QueryOrdersArgs = {
     options?: Maybe<OrderListOptions>;
 };
 
+export type QueryJobArgs = {
+    jobId: Scalars['String'];
+};
+
+export type QueryJobsArgs = {
+    input?: Maybe<JobListInput>;
+};
+
 export type QueryPaymentMethodsArgs = {
     options?: Maybe<PaymentMethodListOptions>;
 };

packages/core/src/api/api-internal-modules.ts

@@ -17,6 +17,7 @@ import { CustomerResolver } from './resolvers/admin/customer.resolver';
 import { FacetResolver } from './resolvers/admin/facet.resolver';
 import { GlobalSettingsResolver } from './resolvers/admin/global-settings.resolver';
 import { ImportResolver } from './resolvers/admin/import.resolver';
+import { JobResolver } from './resolvers/admin/job.resolver';
 import { OrderResolver } from './resolvers/admin/order.resolver';
 import { PaymentMethodResolver } from './resolvers/admin/payment-method.resolver';
 import { ProductOptionResolver } from './resolvers/admin/product-option.resolver';
@@ -53,6 +54,7 @@ const adminResolvers = [
     FacetResolver,
     GlobalSettingsResolver,
     ImportResolver,
+    JobResolver,
     OrderResolver,
     PaymentMethodResolver,
     ProductOptionResolver,

packages/core/src/api/resolvers/admin/job.resolver.ts

@@ -0,0 +1,22 @@
+import { Args, Query, Resolver } from '@nestjs/graphql';
+import { Permission, QueryJobArgs, QueryJobsArgs } from '@vendure/common/lib/generated-types';
+
+import { JobService } from '../../../service/services/job.service';
+import { Allow } from '../../decorators/allow.decorator';
+
+@Resolver()
+export class JobResolver {
+    constructor(private jobService: JobService) {}
+
+    @Query()
+    @Allow(Permission.Authenticated)
+    job(@Args() args: QueryJobArgs) {
+        return this.jobService.getOne(args.jobId);
+    }
+
+    @Query()
+    @Allow(Permission.Authenticated)
+    jobs(@Args() args: QueryJobsArgs) {
+        return this.jobService.getAll(args.input || undefined);
+    }
+}

packages/core/src/api/resolvers/admin/search.resolver.ts

@@ -22,7 +22,7 @@ export class SearchResolver {
 
     @Mutation()
     @Allow(Permission.UpdateCatalog)
-    async reindex(...args: any[]): Promise<{ success: boolean } & { [key: string]: any }> {
+    async reindex(...args: any[]): Promise<any> {
         throw new InternalServerError(`error.no-search-plugin-configured`);
     }
 }

packages/core/src/api/schema/admin-api/job.api.graphql

@@ -0,0 +1,27 @@
+type Query {
+    job(jobId: String!): JobInfo
+    jobs(input: JobListInput): [JobInfo!]!
+}
+
+enum JobState {
+    PENDING
+    RUNNING
+    COMPLETED
+    FAILED
+}
+
+input JobListInput {
+    state: JobState
+    ids: [String!]
+}
+
+type JobInfo {
+    id: String!
+    name: String!
+    state: JobState!
+    progress: Float!
+    result: JSON
+    started: DateTime
+    ended: DateTime
+    duration: Int
+}

packages/core/src/api/schema/admin-api/product-search.api.graphql

@@ -3,7 +3,7 @@ type Query {
 }
 
 type Mutation {
-    reindex: SearchReindexResponse!
+    reindex: JobInfo!
 }
 
 type SearchResult {

packages/core/src/plugin/default-search-plugin/fulltext-search.resolver.ts

@@ -1,5 +1,5 @@
 import { Args, Mutation, Parent, Query, ResolveProperty, Resolver } from '@nestjs/graphql';
-import { Permission, QuerySearchArgs, SearchInput, SearchResponse } from '@vendure/common/lib/generated-types';
+import { JobInfo, Permission, QuerySearchArgs, SearchInput, SearchResponse } from '@vendure/common/lib/generated-types';
 import { Omit } from '@vendure/common/lib/omit';
 
 import { Decode } from '../../api';
@@ -9,7 +9,6 @@ import { Ctx } from '../../api/decorators/request-context.decorator';
 import { SearchResolver as BaseSearchResolver } from '../../api/resolvers/admin/search.resolver';
 import { FacetValue } from '../../entity';
 
-import { DefaultSearchReindexResponse } from './default-search-plugin';
 import { FulltextSearchService } from './fulltext-search.service';
 
 @Resolver('SearchResponse')
@@ -66,7 +65,7 @@ export class AdminFulltextSearchResolver implements BaseSearchResolver {
 
     @Mutation()
     @Allow(Permission.UpdateCatalog)
-    async reindex(@Ctx() ctx: RequestContext): Promise<DefaultSearchReindexResponse> {
+    async reindex(@Ctx() ctx: RequestContext): Promise<JobInfo> {
         return this.fulltextSearchService.reindex(ctx);
     }
 }

packages/core/src/plugin/default-search-plugin/fulltext-search.service.ts

@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 import { InjectConnection } from '@nestjs/typeorm';
-import { LanguageCode, SearchInput, SearchResponse } from '@vendure/common/lib/generated-types';
+import { JobInfo, LanguageCode, SearchInput, SearchResponse } from '@vendure/common/lib/generated-types';
 import { Omit } from '@vendure/common/lib/omit';
 import { ID } from '@vendure/common/lib/shared-types';
 import { unique } from '@vendure/common/lib/unique';
@@ -14,11 +14,11 @@ import { FacetValue, Product, ProductVariant } from '../../entity';
 import { EventBus } from '../../event-bus/event-bus';
 import { translateDeep } from '../../service/helpers/utils/translate-entity';
 import { FacetValueService } from '../../service/services/facet-value.service';
+import { JobService } from '../../service/services/job.service';
 import { ProductVariantService } from '../../service/services/product-variant.service';
 import { SearchService } from '../../service/services/search.service';
 
 import { AsyncQueue } from './async-queue';
-import { DefaultSearchReindexResponse } from './default-search-plugin';
 import { SearchIndexItem } from './search-index-item.entity';
 import { MysqlSearchStrategy } from './search-strategy/mysql-search-strategy';
 import { PostgresSearchStrategy } from './search-strategy/postgres-search-strategy';
@@ -48,6 +48,7 @@ export class FulltextSearchService implements SearchService {
 
     constructor(
         @InjectConnection() private connection: Connection,
+        private jobService: JobService,
         private eventBus: EventBus,
         private facetValueService: FacetValueService,
         private productVariantService: ProductVariantService,
@@ -91,42 +92,47 @@ export class FulltextSearchService implements SearchService {
     /**
      * Rebuilds the full search index.
      */
-    async reindex(ctx: RequestContext): Promise<DefaultSearchReindexResponse> {
-        const timeStart = Date.now();
-        const BATCH_SIZE = 100;
-        Logger.verbose('Reindexing search index...');
-        const qb = await this.connection.getRepository(ProductVariant).createQueryBuilder('variants');
-        FindOptionsUtils.applyFindManyOptionsOrConditionsToQueryBuilder(qb, {
-            relations: this.variantRelations,
-        });
-        FindOptionsUtils.joinEagerRelations(qb, qb.alias, this.connection.getMetadata(ProductVariant));
-        const count = await qb.where('variants__product.deletedAt IS NULL').getCount();
-        Logger.verbose(`Getting ${count} variants`);
-        const batches = Math.ceil(count / BATCH_SIZE);
+    async reindex(ctx: RequestContext): Promise<JobInfo> {
+        const job = this.jobService.startJob('reindex', async reporter => {
+            const timeStart = Date.now();
+            const BATCH_SIZE = 100;
+            Logger.verbose('Reindexing search index...');
+            const qb = await this.connection.getRepository(ProductVariant).createQueryBuilder('variants');
+            FindOptionsUtils.applyFindManyOptionsOrConditionsToQueryBuilder(qb, {
+                relations: this.variantRelations,
+            });
+            FindOptionsUtils.joinEagerRelations(qb, qb.alias, this.connection.getMetadata(ProductVariant));
+            const count = await qb.where('variants__product.deletedAt IS NULL').getCount();
+            Logger.verbose(`Getting ${count} variants`);
+            const batches = Math.ceil(count / BATCH_SIZE);
 
-        Logger.verbose('Deleting existing index items...');
-        await this.connection.getRepository(SearchIndexItem).delete({ languageCode: ctx.languageCode });
-        Logger.verbose('Deleted!');
+            Logger.verbose('Deleting existing index items...');
+            await this.connection.getRepository(SearchIndexItem).delete({languageCode: ctx.languageCode});
+            Logger.verbose('Deleted!');
 
-        for (let i = 0; i < batches; i++) {
-            Logger.verbose(`Processing batch ${i + 1} of ${batches}, heap used: `
-                + (process.memoryUsage().heapUsed / 1000 / 1000).toFixed(2) + 'MB');
-            const variants = await qb
-                .where('variants__product.deletedAt IS NULL')
-                .take(BATCH_SIZE)
-                .skip(i * BATCH_SIZE)
-                .getMany();
-            await this.taskQueue.push(async () => {
-                await this.saveSearchIndexItems(ctx, variants);
-            });
-        }
+            for (let i = 0; i < batches; i++) {
+                Logger.verbose(`Processing batch ${i + 1} of ${batches}, heap used: `
+                    + (process.memoryUsage().heapUsed / 1000 / 1000).toFixed(2) + 'MB');
+                const variants = await qb
+                    .where('variants__product.deletedAt IS NULL')
+                    .take(BATCH_SIZE)
+                    .skip(i * BATCH_SIZE)
+                    .getMany();
+                await this.taskQueue.push(async () => {
+                    await this.saveSearchIndexItems(ctx, variants);
+                });
+                reporter.setProgress(Math.round((i / batches) * 100));
+            }
 
-        Logger.verbose(`Reindexing completed in ${Date.now() - timeStart}ms`);
-        return {
-            success: true,
-            indexedItemCount: count,
-            timeTaken: Date.now() - timeStart,
-        };
+            Logger.verbose(`Reindexing completed in ${Date.now() - timeStart}ms`);
+
+            return {
+                success: true,
+                indexedItemCount: count,
+                timeTaken: Date.now() - timeStart,
+            };
+        });
+        return job;
     }
 
     /**

packages/core/src/service/helpers/job-manager/job-manager.spec.ts

@@ -0,0 +1,154 @@
+/* tslint:disable:no-non-null-assertion no-empty */
+import { JobState } from '@vendure/common/lib/generated-types';
+import { pick } from '@vendure/common/lib/pick';
+import { Subject } from 'rxjs';
+
+import { JobManager } from './job-manager';
+
+describe('JobManager', () => {
+    const noop = () => {};
+
+    it('getOne() returns null for invalid id', () => {
+        const jm = new JobManager();
+        expect(jm.getOne('invalid')).toBeNull();
+    });
+
+    it('startJob() returns a job', () => {
+        const jm = new JobManager();
+        const job = jm.startJob('test', noop);
+        expect(job.name).toBe('test');
+    });
+
+    it('getOne() returns job by id', () => {
+        const jm = new JobManager();
+        const job1 = jm.startJob('test', noop);
+        const job2 = jm.getOne(job1.id);
+
+        expect(job1.id).toBe(job2!.id);
+    });
+
+    it('job completes once work fn returns', async () => {
+        const jm = new JobManager();
+        const subject = new Subject();
+        const job = jm.startJob('test', () => subject.toPromise());
+        await tick();
+
+        expect(jm.getOne(job.id)!.state).toBe(JobState.RUNNING);
+
+        subject.next('result');
+        subject.complete();
+        await tick();
+
+        const result = jm.getOne(job.id)!;
+        expect(result.state).toBe(JobState.COMPLETED);
+        expect(result.result).toBe('result');
+    });
+
+    it('job fails if work fn throws', async () => {
+        const jm = new JobManager();
+        const subject = new Subject();
+        const job = jm.startJob('test', () => subject.toPromise());
+        await tick();
+
+        expect(jm.getOne(job.id)!.state).toBe(JobState.RUNNING);
+
+        subject.error('oh no');
+        await tick();
+
+        const result = jm.getOne(job.id)!;
+        expect(result.state).toBe(JobState.FAILED);
+        expect(result.result).toBe('oh no');
+    });
+
+    it('reporter.setProgress updates job progress', async () => {
+        const jm = new JobManager();
+        const subject = new Subject();
+        const progressSubject = new Subject<number>();
+        const job = jm.startJob('test', (reporter => {
+            progressSubject.subscribe(val => reporter.setProgress(val));
+            return subject.toPromise();
+        }));
+        await tick();
+        expect(jm.getOne(job.id)!.progress).toBe(0);
+
+        progressSubject.next(10);
+        expect(jm.getOne(job.id)!.progress).toBe(10);
+
+        progressSubject.next(42);
+        expect(jm.getOne(job.id)!.progress).toBe(42);
+
+        progressSubject.next(500);
+        expect(jm.getOne(job.id)!.progress).toBe(100);
+
+        progressSubject.next(88);
+        expect(jm.getOne(job.id)!.progress).toBe(88);
+
+        subject.complete();
+        await tick();
+
+        const result = jm.getOne(job.id)!;
+        expect(jm.getOne(job.id)!.progress).toBe(100);
+    });
+
+    it('getAll() returns all jobs', () => {
+        const jm = new JobManager();
+        const job1 = jm.startJob('job1', noop);
+        const job2 = jm.startJob('job2', noop);
+        const job3 = jm.startJob('job3', noop);
+
+        expect(jm.getAll().map(j => j.id)).toEqual([job1.id, job2.id, job3.id]);
+    });
+
+    it('getAll() filters by id', () => {
+        const jm = new JobManager();
+        const job1 = jm.startJob('job1', noop);
+        const job2 = jm.startJob('job2', noop);
+        const job3 = jm.startJob('job3', noop);
+
+        expect(jm.getAll({ ids: [job1.id, job3.id]}).map(j => j.id)).toEqual([job1.id, job3.id]);
+    });
+
+    it('getAll() filters by state', async () => {
+        const jm = new JobManager();
+        const subject = new Subject();
+        const job1 = jm.startJob('job1', noop);
+        const job2 = jm.startJob('job2', noop);
+        const job3 = jm.startJob('job3', () => subject.toPromise());
+
+        await tick();
+
+        expect(jm.getAll({ state: JobState.COMPLETED}).map(j => j.id)).toEqual([job1.id, job2.id]);
+        expect(jm.getAll({ state: JobState.RUNNING}).map(j => j.id)).toEqual([job3.id]);
+    });
+
+    it('clean() removes completed jobs older than maxAge', async () => {
+        const jm = new JobManager('10ms');
+        const subject1 = new Subject();
+        const subject2 = new Subject();
+
+        const job1 = jm.startJob('job1', () => subject1.toPromise());
+        const job2 = jm.startJob('job2', () => subject2.toPromise());
+
+        subject1.complete();
+        await tick();
+
+        jm.clean();
+
+        expect(jm.getAll().map(pick(['name', 'state']))).toEqual([
+            { name: 'job1', state: JobState.COMPLETED },
+            { name: 'job2', state: JobState.RUNNING },
+        ]);
+
+        await tick(20);
+
+        jm.clean();
+
+        expect(jm.getAll().map(pick(['name', 'state']))).toEqual([
+            { name: 'job2', state: JobState.RUNNING },
+        ]);
+    });
+});
+
+function tick(duration: number = 0) {
+    return new Promise(resolve => global.setTimeout(resolve, duration));
+}

packages/core/src/service/helpers/job-manager/job-manager.ts

@@ -0,0 +1,91 @@
+import { JobInfo, JobListInput } from '@vendure/common/lib/generated-types';
+import { pick } from '@vendure/common/lib/pick';
+import ms = require('ms');
+
+import { Job } from './job';
+
+/**
+ * The JobReporter allows a long-running job to update its progress during the
+ * duration of the work function. This can then be used in the client application
+ * to display a progress indication to the user.
+ */
+export interface JobReporter {
+    setProgress(percentage: number): void;
+}
+
+/**
+ * The JobManager is responsible for creating and monitoring {@link Job} instances.
+ */
+export class JobManager {
+    private jobs = new Map<string, Job>();
+    /**
+     * Defines the maximum age of a completed/failed Job before it
+     * is removd when `clean()` is called.
+     */
+    private readonly maxAgeInMs = ms('1d');
+
+    constructor(maxAge?: string) {
+        if (maxAge) {
+            this.maxAgeInMs = ms(maxAge);
+        }
+    }
+
+    /**
+     * Creates a new {@link Job} instance with the given work function. When the function
+     * returns, the job will be completed. The return value is then available as the `result`
+     * property of the job. If the function throws, the job will fail and the `result` property
+     * will be the error thrown.
+     */
+    startJob(name: string, work: (reporter: JobReporter) => any | Promise<any>): Job {
+        const job = new Job(name, work);
+        this.jobs.set(job.id, job);
+        job.start();
+        return job;
+    }
+
+    getAll(input?: JobListInput): JobInfo[] {
+        return Array.from(this.jobs.values()).map(this.toJobInfo).filter(job => {
+            if (input) {
+                let match = false;
+                if (input.state) {
+                    match = job.state === input.state;
+                }
+                if (input.ids) {
+                    match = input.ids.includes(job.id);
+                }
+                return match;
+            } else {
+                return true;
+            }
+        });
+    }
+
+    getOne(jobId: string): JobInfo | null {
+        const job = this.jobs.get(jobId);
+        if (!job) {
+            return null;
+        }
+        return this.toJobInfo(job);
+    }
+
+    /**
+     * Removes all completed jobs which are older than the maxAge.
+     */
+    clean() {
+        const nowMs = +(new Date());
+        Array.from(this.jobs.values()).forEach(job => {
+            if (job.ended) {
+                const delta = nowMs - +job.ended;
+                if (this.maxAgeInMs < delta) {
+                    this.jobs.delete(job.id);
+                }
+            }
+        });
+    }
+
+    private toJobInfo(job: Job): JobInfo {
+        const info =  pick(job, ['id', 'name', 'state', 'progress', 'result', 'started', 'ended']);
+        const duration = job.ended ? +job.ended - +info.started : Date.now() - +info.started;
+        return { ...info, duration };
+    }
+}

packages/core/src/service/helpers/job-manager/job.ts

@@ -0,0 +1,43 @@
+import { JobState } from '../../../../../common/lib/generated-types';
+import { generatePublicId } from '../../../common/generate-public-id';
+
+import { JobReporter } from './job-manager';
+
+/**
+ * A Job represents a piece of work to be run in the background, i.e. outside the request-response cycle.
+ * It is intended to be used for long-running work triggered by API requests (e.g. a mutation which
+ * kicks off a re-build of the search index).
+ */
+export class Job {
+    id: string;
+    state: JobState = JobState.PENDING;
+    progress = 0;
+    result = null;
+    started: Date;
+    ended: Date;
+
+    constructor(public name: string, public work: (reporter: JobReporter) => any | Promise<any>) {
+        this.id = generatePublicId();
+        this.started = new Date();
+    }
+
+    async start() {
+        const reporter: JobReporter = {
+            setProgress: (percentage: number) => {
+                this.progress = Math.max(Math.min(percentage, 100), 0);
+            },
+        };
+        let result: any;
+        try {
+            this.state = JobState.RUNNING;
+            result = await this.work(reporter);
+            this.progress = 100;
+            this.result = result;
+            this.state = JobState.COMPLETED;
+        } catch (e) {
+            this.state = JobState.FAILED;
+            this.result = e;
+        }
+        this.ended = new Date();
+    }
+}

packages/core/src/service/service.module.ts

@@ -26,6 +26,7 @@ import { CustomerService } from './services/customer.service';
 import { FacetValueService } from './services/facet-value.service';
 import { FacetService } from './services/facet.service';
 import { GlobalSettingsService } from './services/global-settings.service';
+import { JobService } from './services/job.service';
 import { OrderService } from './services/order.service';
 import { PaymentMethodService } from './services/payment-method.service';
 import { ProductOptionGroupService } from './services/product-option-group.service';
@@ -54,6 +55,7 @@ const exportedProviders = [
     FacetService,
     FacetValueService,
     GlobalSettingsService,
+    JobService,
     OrderService,
     PaymentMethodService,
     CollectionService,

packages/core/src/service/services/job.service.ts

@@ -0,0 +1,33 @@
+import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import { JobInfo, JobListInput } from '@vendure/common/lib/generated-types';
+import ms = require('ms');
+
+import { Job } from '../helpers/job-manager/job';
+import { JobManager, JobReporter } from '../helpers/job-manager/job-manager';
+
+@Injectable()
+export class JobService implements OnModuleInit, OnModuleDestroy {
+    private manager = new JobManager('1d');
+    private readonly cleanJobsInterval = ms('1d');
+    private cleanJobsTimer: NodeJS.Timeout;
+
+    onModuleInit() {
+        this.cleanJobsTimer = global.setInterval(() => this.manager.clean(), this.cleanJobsInterval);
+    }
+
+    onModuleDestroy() {
+        global.clearInterval(this.cleanJobsTimer);
+    }
+
+    startJob(name: string, work: (reporter: JobReporter) => any | Promise<any>): Job {
+        return this.manager.startJob(name, work);
+    }
+
+    getAll(input?: JobListInput): JobInfo[] {
+        return this.manager.getAll(input);
+    }
+
+    getOne(jobId: string): JobInfo | null {
+        return this.manager.getOne(jobId);
+    }
+}

packages/core/src/service/services/search.service.ts

@@ -1,7 +1,8 @@
 import { Injectable } from '@nestjs/common';
-import { SearchReindexResponse } from '@vendure/common/lib/generated-types';
+import { JobInfo } from '@vendure/common/lib/generated-types';
 
 import { RequestContext } from '../../api/common/request-context';
+import { InternalServerError } from '../../common/error/errors';
 import { Logger } from '../../config/logger/vendure-logger';
 
 /**
@@ -17,14 +18,10 @@ import { Logger } from '../../config/logger/vendure-logger';
  */
 @Injectable()
 export class SearchService {
-    async reindex(ctx: RequestContext): Promise<SearchReindexResponse> {
+    async reindex(ctx: RequestContext): Promise<JobInfo> {
         if (!process.env.CI) {
             Logger.warn(`The SearchService should be overridden by an appropriate search plugin.`);
         }
-        return {
-            indexedItemCount: 0,
-            success: false,
-            timeTaken: 0,
-        };
+        throw new InternalServerError('error.not-implemented');
     }
 }