⚡ run subscriptions job in batches (#766)

* 🔊 log process error * ✨ measure api performance, add statistics endpoint * log subscriptions job progress * 🚚 rename plugins to not match module name * ⚡️ run subscriptions job in batches
ViewTube · Jul 23, 2021 · 1d88911 · 1d88911
1 parent 193c1ee
commit 1d88911
Show file tree

Hide file tree

Showing 24 changed files with 66 additions and 38 deletions.
diff --git a/client/components/About.vue b/client/components/About.vue
@@ -78,7 +78,7 @@ import BadgeButton from '@/components/buttons/BadgeButton.vue';
 import InvidiousLicense from '@/components/licenses/Invidious.vue';
 import { computed, defineComponent, onMounted, ref } from '@nuxtjs/composition-api';
 import { useAccessor } from '@/store';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 
 export default defineComponent({
   name: 'About',

diff --git a/client/components/Comment.vue b/client/components/Comment.vue
@@ -90,7 +90,7 @@ import BadgeButton from '@/components/buttons/BadgeButton.vue';
 import Invidious from '@/plugins/services/invidious';
 import { defineComponent, ref, useRoute } from '@nuxtjs/composition-api';
 import { useAccessor } from '@/store';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 import { useImgProxy } from '@/plugins/proxy';
 
 export default defineComponent({

diff --git a/client/components/SearchAutoComplete.vue b/client/components/SearchAutoComplete.vue
@@ -18,7 +18,7 @@
 
 <script lang="ts">
 import { defineComponent, ref, watch } from '@nuxtjs/composition-api';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 import { useAccessor } from '@/store';
 
 export default defineComponent({

diff --git a/client/components/buttons/SubscribeButton.vue b/client/components/buttons/SubscribeButton.vue
@@ -24,7 +24,7 @@
 
 <script lang="ts">
 import { defineComponent, onMounted, ref } from '@nuxtjs/composition-api';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 import { useAccessor } from '@/store/index';
 
 export default defineComponent({

diff --git a/client/components/history/HistoryList.vue b/client/components/history/HistoryList.vue
@@ -45,7 +45,7 @@
 import { defineComponent } from '@nuxtjs/composition-api';
 import humanizeDuration from 'humanize-duration';
 import DeleteIcon from 'vue-material-design-icons/Delete.vue';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 import { useAccessor } from '@/store';
 import BadgeButton from '@/components/buttons/BadgeButton.vue';
 

diff --git a/client/components/licenses/Invidious.vue b/client/components/licenses/Invidious.vue
@@ -6,7 +6,7 @@
 
 <script lang="ts">
 import { defineComponent, ref, useFetch } from '@nuxtjs/composition-api';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 
 export default defineComponent({
   name: 'InvidiousLicense',

diff --git a/client/components/popup/SubscriptionImport.vue b/client/components/popup/SubscriptionImport.vue
@@ -138,7 +138,7 @@ import SubscriptionConverter from '@/plugins/services/subscriptionConverter';
 import Spinner from '@/components/Spinner.vue';
 import '@/assets/styles/popup.scss';
 import { computed, defineComponent, ref } from '@nuxtjs/composition-api';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 import { useAccessor } from '@/store/index';
 
 class ChannelDto {

diff --git a/client/components/videoplayer/helpers/index.ts b/client/components/videoplayer/helpers/index.ts
@@ -11,13 +11,13 @@ import { commons } from '@/plugins/commons';
 import { SponsorBlock } from '@/plugins/services/sponsorBlock';
 import { SponsorBlockSegmentsDto } from '@/plugins/shared';
 import { useAccessor } from '@/store';
+import { useFormatting } from '@/plugins/formatting';
+import { useAxios } from '@/plugins/axiosPlugin';
+import { useImgProxy } from '@/plugins/proxy';
 import { MediaMetadataHelper } from './mediaMetadata';
 import { calculateSeekPercentage, matchSeekProgressPercentage, seekbarFunctions } from './seekbar';
 import { parseChapters } from './chapters';
 import { destroyInstance, initializeHlsStream, isHlsNative, isHlsSupported } from './hlsHelper';
-import { useFormatting } from '~/plugins/formatting';
-import { useAxios } from '~/plugins/axios';
-import { useImgProxy } from '~/plugins/proxy';
 
 export const videoPlayerSetup = (props: any, emit: Function) => {
   const accessor = useAccessor();

diff --git a/client/components/watch/VideoLoadingTemplate.vue b/client/components/watch/VideoLoadingTemplate.vue
@@ -14,7 +14,7 @@
 <script lang="ts">
 import { computed, defineComponent } from '@nuxtjs/composition-api';
 import Spinner from '@/components/Spinner.vue';
-import { useImgProxy } from '~/plugins/proxy';
+import { useImgProxy } from '@/plugins/proxy';
 
 export default defineComponent({
   name: 'VideoLoadingTemplate',

diff --git a/client/pages/history.vue b/client/pages/history.vue
@@ -54,7 +54,7 @@ import SectionTitle from '@/components/SectionTitle.vue';
 import RestartOffIcon from 'vue-material-design-icons/RestartOff.vue';
 import Pagination from '@/components/pagination/Pagination.vue';
 import HistoryList from '@/components/history/HistoryList.vue';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 import { useAccessor } from '@/store';
 import BadgeButton from '@/components/buttons/BadgeButton.vue';
 import SmallSearchBox from '@/components/SmallSearchBox.vue';

diff --git a/client/pages/index.vue b/client/pages/index.vue
@@ -48,7 +48,7 @@ import ViewTubeApi from '@/plugins/services/viewTubeApi';
 import BadgeButton from '@/components/buttons/BadgeButton.vue';
 import { defineComponent, ref, useFetch, useMeta } from '@nuxtjs/composition-api';
 import { useAccessor } from '@/store';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 
 export default defineComponent({
   name: 'Home',

diff --git a/client/pages/playlist.vue b/client/pages/playlist.vue
@@ -73,9 +73,9 @@ import EyeIcon from 'vue-material-design-icons/EyeOutline.vue';
 import EyeClosedIcon from 'vue-material-design-icons/EyeOffOutline.vue';
 import CountIcon from 'vue-material-design-icons/Counter.vue';
 import CalendarIcon from 'vue-material-design-icons/CalendarClock.vue';
-import { useAxios } from '~/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
+import { useImgProxy } from '@/plugins/proxy';
 import { useAccessor } from '~/store';
-import { useImgProxy } from '~/plugins/proxy';
 
 export default defineComponent({
   name: 'Playlist',

diff --git a/client/pages/profile/index.vue b/client/pages/profile/index.vue
@@ -136,7 +136,7 @@ import {
 } from '@nuxtjs/composition-api';
 import HistoryList from '@/components/history/HistoryList.vue';
 import { useAccessor } from '@/store';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 
 export default defineComponent({
   name: 'Profile',

diff --git a/client/pages/results.vue b/client/pages/results.vue
@@ -65,7 +65,7 @@ import {
   watch
 } from '@nuxtjs/composition-api';
 import { useAccessor } from '@/store';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 
 export default defineComponent({
   name: 'Search',

diff --git a/client/pages/subscriptions/index.vue b/client/pages/subscriptions/index.vue
@@ -99,7 +99,7 @@ import {
   watch
 } from '@nuxtjs/composition-api';
 import { useAccessor } from '@/store';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 
 export default defineComponent({
   name: 'Subscriptions',

diff --git a/client/pages/subscriptions/manage.vue b/client/pages/subscriptions/manage.vue
@@ -73,7 +73,7 @@ import {
   watch
 } from '@nuxtjs/composition-api';
 import { useAccessor } from '@/store';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 import SmallSearchBox from '@/components/SmallSearchBox.vue';
 
 export default defineComponent({

diff --git a/client/pages/watch.vue b/client/pages/watch.vue
@@ -192,7 +192,7 @@ import {
   watch
 } from '@nuxtjs/composition-api';
 import { useAccessor } from '@/store';
-import { useAxios } from '@/plugins/axios';
+import { useAxios } from '@/plugins/axiosPlugin';
 import { useImgProxy } from '@/plugins/proxy';
 import VideoLoadingTemplate from '@/components/watch/VideoLoadingTemplate.vue';
 import { Result } from 'ytpl';

diff --git a/client/plugins/axios.ts → client/plugins/axiosPlugin.ts b/client/plugins/axios.ts → client/plugins/axiosPlugin.ts
diff --git a/client/plugins/vue-datepicker.ts → client/plugins/vueDatepickerPlugin.ts b/client/plugins/vue-datepicker.ts → client/plugins/vueDatepickerPlugin.ts
diff --git a/client/store/instances.ts b/client/store/instances.ts
@@ -1,5 +1,5 @@
 import { getterTree, mutationTree, actionTree } from 'typed-vuex';
-import { commons } from '~/plugins/commons';
+import { commons } from '@/plugins/commons';
 
 export const state = () => ({
   currentInstance: '' as string,

diff --git a/nuxt.config.ts b/nuxt.config.ts
@@ -106,7 +106,7 @@ const config: NuxtConfig = {
     '@/plugins/directives/index',
     '@/plugins/formatting',
     '@/plugins/shared',
-    { src: '~/plugins/vue-datepicker', mode: 'client' },
+    { src: '@/plugins/vueDatepickerPlugin', mode: 'client' },
     { src: '@/plugins/localStorage', mode: 'client' }
   ],
 

diff --git a/server/user/subscriptions/subscriptions-job.helper.ts b/server/user/subscriptions/subscriptions-job.helper.ts
@@ -8,9 +8,12 @@ import humanizeDuration from 'humanize-duration';
 // import { Common } from 'server/core/common';
 import Consola from 'consola';
 import { Common } from 'server/core/common';
+import { Job } from 'bull';
 
 export const runSubscriptionsJob = async (
-  uniqueChannelIds: Array<string>
+  uniqueChannelIds: Array<string>,
+  job: Job = null
+
 ): Promise<{
   channelResultArray: Array<ChannelBasicInfoDto>;
   videoResultArray: Array<VideoBasicInfoDto>;
@@ -30,16 +33,32 @@ export const runSubscriptionsJob = async (
     });
   };
 
+  const channelIdBatches = [];
+  uniqueChannelIds = [].concat(...uniqueChannelIds);
+
+  while (uniqueChannelIds.length) {
+    channelIdBatches.push(uniqueChannelIds.splice(0, 100));
+  }
+
+  console.log(channelIdBatches.length);
+
   let i = 0;
 
-  await uniqueChannelIds
-    .reduce(async (previousPromise: Promise<void>, nextString: string) => {
+  await channelIdBatches
+    .reduce(async (previousPromise: Promise<void>, nextBatch: Array<string>) => {
       await previousPromise;
-      console.log(`${i} of ${uniqueChannelIds.length}`);
+      const jobProgress = Math.floor((i / channelIdBatches.length) * 100);
+      await job.progress(jobProgress);
       i++;
-      return getFeedPromise(nextString);
+      return Promise.allSettled(
+        nextBatch.map(val => {
+          return getFeedPromise(val);
+        })
+      );
     }, Promise.resolve())
-    .catch(() => {});
+    .catch(error => {
+      console.log('job error', error);
+    });
 
   if (videoRawResultArray.length > 0) {
     videoResultArray = videoRawResultArray.reduce(

diff --git a/server/user/subscriptions/subscriptions.processor.ts b/server/user/subscriptions/subscriptions.processor.ts
@@ -1,4 +1,11 @@
-import { OnQueueActive, OnQueueCompleted, OnQueueError, Process, Processor } from '@nestjs/bull';
+import {
+  OnQueueActive,
+  OnQueueCompleted,
+  OnQueueError,
+  OnQueueProgress,
+  Process,
+  Processor
+} from '@nestjs/bull';
 import { InjectModel } from '@nestjs/mongoose';
 import { Job } from 'bull';
 import { Model } from 'mongoose';
@@ -28,9 +35,9 @@ export class SubscriptionsProcessor {
     const uniqueChannelIds = [...new Set(channelIds)];
     let subscriptionResults = null;
     try {
-      subscriptionResults = await runSubscriptionsJob(uniqueChannelIds);
+      subscriptionResults = await runSubscriptionsJob(uniqueChannelIds, job);
     } catch (error) {
-      console.log(error);
+      throw new Error(error);
     }
 
     if (subscriptionResults) {
@@ -62,6 +69,10 @@ export class SubscriptionsProcessor {
     return null;
   }
 
+  @OnQueueProgress()
+  onProgress(job: Job, progress: number) {
+    console.log(`${job.name}: ${progress}%`);
+  }
   @OnQueueError()
   onError(error: Error) {
     console.log(error);

diff --git a/server/user/subscriptions/subscriptions.service.ts b/server/user/subscriptions/subscriptions.service.ts
@@ -37,16 +37,14 @@ export class SubscriptionsService {
     private notificationsService: NotificationsService
   ) {}
 
-  @Cron(CronExpression.EVERY_30_MINUTES)
+  @Cron(CronExpression.EVERY_HOUR)
+  // @Cron(new Date(Date.now() + 60 * 1000))
   async collectSubscriptionsJob(): Promise<void> {
     const userSubscriptions = await this.subscriptionModel.find().lean(true).exec();
 
-    this.subscriptionsQueue.add(
-      {
-        userSubscriptions
-      },
-      {}
-    );
+    this.subscriptionsQueue.add({
+      userSubscriptions
+    });
   }
 
   async saveChannelBasicInfo(channel: ChannelBasicInfoDto): Promise<ChannelBasicInfoDto | null> {