Filter and delete jobs

apps/api/src/app/admin/queue/queue.controller.ts

@@ -8,10 +8,12 @@ import {
   HttpException,
   Inject,
   Param,
+  Query,
   UseGuards
 } from '@nestjs/common';
 import { REQUEST } from '@nestjs/core';
 import { AuthGuard } from '@nestjs/passport';
+import { JobStatus } from 'bull';
 import { StatusCodes, getReasonPhrase } from 'http-status-codes';
 
 import { QueueService } from './queue.service';
@@ -23,9 +25,32 @@ export class QueueController {
     @Inject(REQUEST) private readonly request: RequestWithUser
   ) {}
 
+  @Delete('job')
+  @UseGuards(AuthGuard('jwt'))
+  public async deleteJobs(
+    @Query('status') filterByStatus?: string
+  ): Promise<void> {
+    if (
+      !hasPermission(
+        this.request.user.permissions,
+        permissions.accessAdminControl
+      )
+    ) {
+      throw new HttpException(
+        getReasonPhrase(StatusCodes.FORBIDDEN),
+        StatusCodes.FORBIDDEN
+      );
+    }
+
+    const status = <JobStatus[]>filterByStatus?.split(',') ?? undefined;
+    return this.queueService.deleteJobs({ status });
+  }
+
   @Get('job')
   @UseGuards(AuthGuard('jwt'))
-  public async getJobs(): Promise<AdminJobs> {
+  public async getJobs(
+    @Query('status') filterByStatus?: string
+  ): Promise<AdminJobs> {
     if (
       !hasPermission(
         this.request.user.permissions,
@@ -38,7 +63,8 @@ export class QueueController {
       );
     }
 
-    return this.queueService.getJobs({});
+    const status = <JobStatus[]>filterByStatus?.split(',') ?? undefined;
+    return this.queueService.getJobs({ status });
   }
 
   @Delete('job/:id')

apps/api/src/app/admin/queue/queue.service.ts

@@ -1,8 +1,11 @@
-import { DATA_GATHERING_QUEUE } from '@ghostfolio/common/config';
+import {
+  DATA_GATHERING_QUEUE,
+  QUEUE_JOB_STATUS_LIST
+} from '@ghostfolio/common/config';
 import { AdminJobs } from '@ghostfolio/common/interfaces';
 import { InjectQueue } from '@nestjs/bull';
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';
-import { Queue } from 'bull';
+import { JobStatus, Queue } from 'bull';
 
 @Injectable()
 export class QueueService {
@@ -15,19 +18,30 @@ export class QueueService {
     return (await this.dataGatheringQueue.getJob(aId))?.remove();
   }
 
+  public async deleteJobs({
+    status = QUEUE_JOB_STATUS_LIST
+  }: {
+    status?: JobStatus[];
+  }) {
+    const jobs = await this.dataGatheringQueue.getJobs(status);
+
+    for (const job of jobs) {
+      try {
+        await job.remove();
+      } catch (error) {
+        Logger.warn(error, 'QueueService');
+      }
+    }
+  }
+
   public async getJobs({
-    limit = 1000
+    limit = 1000,
+    status = QUEUE_JOB_STATUS_LIST
   }: {
     limit?: number;
+    status?: JobStatus[];
   }): Promise<AdminJobs> {
-    const jobs = await this.dataGatheringQueue.getJobs([
+    const jobs = await this.dataGatheringQueue.getJobs(status);
-      'active',
-      'completed',
-      'delayed',
-      'failed',
-      'paused',
-      'waiting'
-    ]);
 
     const jobsWithState = await Promise.all(
       jobs.slice(0, limit).map(async (job) => {

apps/client/src/app/components/admin-jobs/admin-jobs.component.ts

@@ -5,10 +5,13 @@ import {
   OnDestroy,
   OnInit
 } from '@angular/core';
+import { FormBuilder, FormGroup } from '@angular/forms';
 import { AdminService } from '@ghostfolio/client/services/admin.service';
 import { UserService } from '@ghostfolio/client/services/user/user.service';
+import { QUEUE_JOB_STATUS_LIST } from '@ghostfolio/common/config';
 import { getDateWithTimeFormatString } from '@ghostfolio/common/helper';
 import { AdminJobs, User } from '@ghostfolio/common/interfaces';
+import { JobStatus } from 'bull';
 import { Subject } from 'rxjs';
 import { takeUntil } from 'rxjs/operators';
 
@@ -20,7 +23,9 @@ import { takeUntil } from 'rxjs/operators';
 })
 export class AdminJobsComponent implements OnDestroy, OnInit {
   public defaultDateTimeFormat: string;
+  public filterForm: FormGroup;
   public jobs: AdminJobs['jobs'] = [];
+  public statusFilterOptions = QUEUE_JOB_STATUS_LIST;
   public user: User;
 
   private unsubscribeSubject = new Subject<void>();
@@ -31,6 +36,7 @@ export class AdminJobsComponent implements OnDestroy, OnInit {
   public constructor(
     private adminService: AdminService,
     private changeDetectorRef: ChangeDetectorRef,
+    private formBuilder: FormBuilder,
     private userService: UserService
   ) {
     this.userService.stateChanged
@@ -50,6 +56,17 @@ export class AdminJobsComponent implements OnDestroy, OnInit {
    * Initializes the controller
    */
   public ngOnInit() {
+    this.filterForm = this.formBuilder.group({
+      status: []
+    });
+
+    this.filterForm.valueChanges
+      .pipe(takeUntil(this.unsubscribeSubject))
+      .subscribe(() => {
+        const currentFilter = this.filterForm.get('status').value;
+        this.fetchJobs(currentFilter ? [currentFilter] : undefined);
+      });
+
     this.fetchJobs();
   }
 
@@ -62,6 +79,16 @@ export class AdminJobsComponent implements OnDestroy, OnInit {
       });
   }
 
+  public onDeleteJobs() {
+    this.adminService
+      .deleteJobs({})
+      .pipe(takeUntil(this.unsubscribeSubject))
+      .subscribe(() => {
+        const currentFilter = this.filterForm.get('status').value;
+        this.fetchJobs(currentFilter ? [currentFilter] : undefined);
+      });
+  }
+
   public onViewStacktrace(aStacktrace: AdminJobs['jobs'][0]['stacktrace']) {
     alert(JSON.stringify(aStacktrace, null, '  '));
   }
@@ -71,9 +98,9 @@ export class AdminJobsComponent implements OnDestroy, OnInit {
     this.unsubscribeSubject.complete();
   }
 
-  private fetchJobs() {
+  private fetchJobs(aStatus?: JobStatus[]) {
     this.adminService
-      .fetchJobs()
+      .fetchJobs({ status: aStatus })
       .pipe(takeUntil(this.unsubscribeSubject))
       .subscribe(({ jobs }) => {
         this.jobs = jobs;

apps/client/src/app/components/admin-jobs/admin-jobs.html

@@ -1,6 +1,26 @@
 <div class="container">
   <div class="row">
     <div class="col">
+      <form class="align-items-center d-flex" [formGroup]="filterForm">
+        <mat-form-field appearance="outline" class="flex-grow-1">
+          <mat-select formControlName="status">
+            <mat-option></mat-option>
+            <mat-option
+              *ngFor="let statusFilterOption of statusFilterOptions"
+              [value]="statusFilterOption"
+              >{{ statusFilterOption }}</mat-option
+            >
+          </mat-select>
+        </mat-form-field>
+        <button
+          class="ml-1"
+          color="warn"
+          mat-flat-button
+          (click)="onDeleteJobs()"
+        >
+          <span i18n>Delete Jobs</span>
+        </button>
+      </form>
       <table class="gf-table w-100">
         <thead>
           <tr class="mat-header-row">
@@ -47,6 +67,10 @@
                 {{ job.finishedOn | date: defaultDateTimeFormat }}
               </td>
               <td class="mat-cell px-1 py-2">
+                <ion-icon
+                  *ngIf="job.state === 'active'"
+                  name="play-outline"
+                ></ion-icon>
                 <ion-icon
                   *ngIf="job.state === 'completed'"
                   class="text-success"
@@ -61,6 +85,14 @@
                   class="text-danger"
                   name="alert-circle-outline"
                 ></ion-icon>
+                <ion-icon
+                  *ngIf="job.state === 'paused'"
+                  name="pause-outline"
+                ></ion-icon>
+                <ion-icon
+                  *ngIf="job.state === 'waiting'"
+                  name="cafe-outline"
+                ></ion-icon>
               </td>
               <td class="mat-cell px-1 py-2">
                 <button

apps/client/src/app/components/admin-jobs/admin-jobs.module.ts

@@ -1,13 +1,22 @@
 import { CommonModule } from '@angular/common';
 import { CUSTOM_ELEMENTS_SCHEMA, NgModule } from '@angular/core';
+import { FormsModule, ReactiveFormsModule } from '@angular/forms';
 import { MatButtonModule } from '@angular/material/button';
 import { MatMenuModule } from '@angular/material/menu';
+import { MatSelectModule } from '@angular/material/select';
 
 import { AdminJobsComponent } from './admin-jobs.component';
 
 @NgModule({
   declarations: [AdminJobsComponent],
-  imports: [CommonModule, MatButtonModule, MatMenuModule],
+  imports: [
+    CommonModule,
+    FormsModule,
+    MatButtonModule,
+    MatMenuModule,
+    MatSelectModule,
+    ReactiveFormsModule
+  ],
   schemas: [CUSTOM_ELEMENTS_SCHEMA]
 })
 export class GfAdminJobsModule {}

apps/client/src/app/services/admin.service.ts

@@ -1,4 +1,4 @@
-import { HttpClient } from '@angular/common/http';
+import { HttpClient, HttpParams } from '@angular/common/http';
 import { Injectable } from '@angular/core';
 import { UpdateMarketDataDto } from '@ghostfolio/api/app/admin/update-market-data.dto';
 import { IDataProviderHistoricalResponse } from '@ghostfolio/api/services/interfaces/interfaces';
@@ -9,6 +9,7 @@ import {
   UniqueAsset
 } from '@ghostfolio/common/interfaces';
 import { DataSource, MarketData } from '@prisma/client';
+import { JobStatus } from 'bull';
 import { format, parseISO } from 'date-fns';
 import { Observable, map } from 'rxjs';
 
@@ -22,6 +23,18 @@ export class AdminService {
     return this.http.delete<void>(`/api/v1/admin/queue/job/${aId}`);
   }
 
+  public deleteJobs({ status }: { status?: JobStatus[] }) {
+    let params = new HttpParams();
+
+    if (status?.length > 0) {
+      params = params.append('status', status.join(','));
+    }
+
+    return this.http.delete<void>('/api/v1/admin/queue/job', {
+      params
+    });
+  }
+
   public deleteProfileData({ dataSource, symbol }: UniqueAsset) {
     return this.http.delete<void>(
       `/api/v1/admin/profile-data/${dataSource}/${symbol}`
@@ -47,20 +60,28 @@ export class AdminService {
       );
   }
 
-  public fetchJobs() {
+  public fetchJobs({ status }: { status?: JobStatus[] }) {
-    return this.http.get<AdminJobs>(`/api/v1/admin/queue/job`);
+    let params = new HttpParams();
+
+    if (status?.length > 0) {
+      params = params.append('status', status.join(','));
+    }
+
+    return this.http.get<AdminJobs>('/api/v1/admin/queue/job', {
+      params
+    });
   }
 
   public gather7Days() {
-    return this.http.post<void>(`/api/v1/admin/gather`, {});
+    return this.http.post<void>('/api/v1/admin/gather', {});
   }
 
   public gatherMax() {
-    return this.http.post<void>(`/api/v1/admin/gather/max`, {});
+    return this.http.post<void>('/api/v1/admin/gather/max', {});
   }
 
   public gatherProfileData() {
-    return this.http.post<void>(`/api/v1/admin/gather/profile-data`, {});
+    return this.http.post<void>('/api/v1/admin/gather/profile-data', {});
   }
 
   public gatherProfileDataBySymbol({ dataSource, symbol }: UniqueAsset) {

libs/common/src/lib/config.ts

@@ -1,4 +1,5 @@
 import { DataSource } from '@prisma/client';
+import { JobStatus } from 'bull';
 
 import { ToggleOption } from './types';
 
@@ -60,4 +61,13 @@ export const PROPERTY_SLACK_COMMUNITY_USERS = 'SLACK_COMMUNITY_USERS';
 export const PROPERTY_STRIPE_CONFIG = 'STRIPE_CONFIG';
 export const PROPERTY_SYSTEM_MESSAGE = 'SYSTEM_MESSAGE';
 
+export const QUEUE_JOB_STATUS_LIST = <JobStatus[]>[
+  'active',
+  'completed',
+  'delayed',
+  'failed',
+  'paused',
+  'waiting'
+];
+
 export const UNKNOWN_KEY = 'UNKNOWN';