Most followed channels query performance optimization (#22)

* Most followed channels query performance optimization

* Sort follows on startup

schema.graphql

@@ -41,7 +41,12 @@ type Query {
   channelViews(channelId: ID!): EntityViewsInfo
 
   """Get list of most followed channels"""
-  mostFollowedChannels(limit: Int, period: Int!): [ChannelFollowsInfo!]!
+  mostFollowedChannels(
+    limit: Int
+
+    """timePeriodDays must take one of the following values: 7, 30"""
+    timePeriodDays: Int!
+  ): [ChannelFollowsInfo!]!
 
   """Get list of most followed channels of all time"""
   mostFollowedChannelsAllTime(limit: Int!): [ChannelFollowsInfo!]

src/aggregates/follows.ts

@@ -7,39 +7,92 @@ import {
 } from '../models/ChannelEvent'
 
 import { ChannelFollowsInfo } from '../entities/ChannelFollowsInfo'
+import { mapPeriods } from '../helpers'
+import { differenceInCalendarDays } from 'date-fns'
 
 type ChannelEventsAggregationResult = {
   events?: ChannelEvent[]
 }[]
 
+type TimePeriodEventsData = {
+  sevenDays: Partial<UnsequencedChannelEvent>[]
+  thirtyDays: Partial<UnsequencedChannelEvent>[]
+}
+
+type TimePeriodFollows = {
+  sevenDays: ChannelFollowsInfo[]
+  thirtyDays: ChannelFollowsInfo[]
+}
+
 export class FollowsAggregate implements GenericAggregate<ChannelEvent> {
   private channelFollowsMap: Record<string, number> = {}
-  private allChannelFollowEvents: Partial<UnsequencedChannelEvent>[] = []
   private allChannelFollows: ChannelFollowsInfo[] = []
 
-  private addOrUpdateFollows(id: string, eventType: ChannelEventType): void {
-    const i = this.allChannelFollows.findIndex((element) => element.id === id)
-    if (i > -1) {
-      if (!this.allChannelFollows[i].follows && eventType === ChannelEventType.UnfollowChannel) return
-      this.allChannelFollows[i].follows =
-        eventType === ChannelEventType.FollowChannel
-          ? this.allChannelFollows[i].follows + 1
-          : this.allChannelFollows[i].follows - 1
-    } else this.allChannelFollows.push({ id, follows: eventType === ChannelEventType.UnfollowChannel ? 0 : 1 })
+  private timePeriodEvents: TimePeriodEventsData = {
+    sevenDays: [],
+    thirtyDays: [],
+  }
+
+  private timePeriodChannelFollows: TimePeriodFollows = {
+    sevenDays: [],
+    thirtyDays: [],
+  }
+
+  private addOrUpdateFollows(array: ChannelFollowsInfo[], id: string, shouldAdd = true): void {
+    const followsObject = array.find((element) => element.id === id)
+
+    if (followsObject) {
+      if (!followsObject.follows && !shouldAdd) return
+
+      if (shouldAdd) {
+        followsObject.follows++
+      } else {
+        followsObject.follows--
+      }
+    } else {
+      array.push({ id, follows: shouldAdd ? 1 : 0 })
+    }
+
+    array.sort((a, b) => (a.follows > b.follows ? -1 : 1))
   }
 
-  private addOrRemoveFollowEvent(eventType: ChannelEventType, { channelId, timestamp }: UnsequencedChannelEvent): void {
+  private addOrRemoveFollowEvent(
+    array: Partial<UnsequencedChannelEvent>[],
+    eventType: ChannelEventType,
+    { channelId, timestamp }: UnsequencedChannelEvent
+  ): void {
     if (eventType === ChannelEventType.FollowChannel) {
-      this.allChannelFollowEvents = [...this.allChannelFollowEvents, { channelId, timestamp }]
+      array.push({ channelId, timestamp })
     }
     if (eventType === ChannelEventType.UnfollowChannel) {
-      const index = this.allChannelFollowEvents.findIndex((item) => item.channelId === channelId)
+      const index = array.findIndex((item) => item.channelId === channelId)
       if (index >= 0) {
-        this.allChannelFollowEvents.splice(index, 1)
+        array.splice(index, 1)
       }
     }
   }
 
+  public filterEventsByPeriod(timePeriodDays: 7 | 30) {
+    const mappedPeriod = mapPeriods(timePeriodDays)
+    const followEvents = this.timePeriodEvents[mappedPeriod]
+
+    // find index of first event that should be kept
+    const firstEventToIncludeIdx = followEvents.findIndex(
+      (follow) => follow.timestamp && differenceInCalendarDays(new Date(), follow.timestamp) <= timePeriodDays
+    )
+
+    for (let i = 0; i < firstEventToIncludeIdx; i++) {
+      const { channelId } = followEvents[i]
+
+      if (channelId) {
+        this.addOrUpdateFollows(this.timePeriodChannelFollows[mappedPeriod], channelId, false)
+      }
+    }
+
+    // remove older events
+    this.timePeriodEvents[mappedPeriod] = followEvents.slice(firstEventToIncludeIdx)
+  }
+
   public channelFollows(channelId: string): number | null {
     return this.channelFollowsMap[channelId] ?? null
   }
@@ -48,14 +101,14 @@ export class FollowsAggregate implements GenericAggregate<ChannelEvent> {
     return Object.freeze(this.channelFollowsMap)
   }
 
-  public getAllFollowEvents() {
-    return this.allChannelFollowEvents
-  }
-
   public getAllChannelFollows() {
     return this.allChannelFollows
   }
 
+  public getTimePeriodChannelFollows() {
+    return this.timePeriodChannelFollows
+  }
+
   public static async Build(): Promise<FollowsAggregate> {
     const aggregation: ChannelEventsAggregationResult = await ChannelEventsBucketModel.aggregate([
       { $unwind: '$events' },
@@ -69,23 +122,34 @@ export class FollowsAggregate implements GenericAggregate<ChannelEvent> {
     events.forEach((event) => {
       aggregate.applyEvent(event)
     })
+
+    aggregate.filterEventsByPeriod(7)
+    aggregate.filterEventsByPeriod(30)
+
     return aggregate
   }
 
   public applyEvent(event: UnsequencedChannelEvent) {
-    const { channelId, type } = event
+    const { type, ...eventWithoutType } = event
+    const { channelId } = eventWithoutType
     const currentChannelFollows = this.channelFollowsMap[channelId] || 0
 
-    switch (event.type) {
+    switch (type) {
       case ChannelEventType.FollowChannel:
         this.channelFollowsMap[channelId] = currentChannelFollows + 1
-        this.addOrUpdateFollows(channelId, ChannelEventType.FollowChannel)
-        this.addOrRemoveFollowEvent(ChannelEventType.FollowChannel, event)
+        this.addOrUpdateFollows(this.allChannelFollows, channelId)
+        this.addOrUpdateFollows(this.timePeriodChannelFollows.sevenDays, channelId)
+        this.addOrUpdateFollows(this.timePeriodChannelFollows.thirtyDays, channelId)
+        this.addOrRemoveFollowEvent(this.timePeriodEvents.sevenDays, ChannelEventType.FollowChannel, event)
+        this.addOrRemoveFollowEvent(this.timePeriodEvents.thirtyDays, ChannelEventType.FollowChannel, event)
         break
       case ChannelEventType.UnfollowChannel:
         this.channelFollowsMap[channelId] = Math.max(currentChannelFollows - 1, 0)
-        this.addOrUpdateFollows(channelId, ChannelEventType.UnfollowChannel)
-        this.addOrRemoveFollowEvent(ChannelEventType.UnfollowChannel, event)
+        this.addOrUpdateFollows(this.allChannelFollows, channelId, false)
+        this.addOrUpdateFollows(this.timePeriodChannelFollows.sevenDays, channelId, false)
+        this.addOrUpdateFollows(this.timePeriodChannelFollows.thirtyDays, channelId, false)
+        this.addOrRemoveFollowEvent(this.timePeriodEvents.sevenDays, ChannelEventType.UnfollowChannel, event)
+        this.addOrRemoveFollowEvent(this.timePeriodEvents.thirtyDays, ChannelEventType.UnfollowChannel, event)
         break
       default:
         console.error(`Parsing unknown channel event: ${type}`)

src/aggregates/views.ts

@@ -1,6 +1,6 @@
 import { UnsequencedVideoEvent, VideoEvent, VideoEventsBucketModel, VideoEventType } from '../models/VideoEvent'
 import { EntityViewsInfo } from '../entities/EntityViewsInfo'
-import { mapPeriods } from '../resolvers/viewsInfo'
+import { mapPeriods } from '../helpers'
 import { differenceInCalendarDays } from 'date-fns'
 
 type VideoEventsAggregationResult = {
@@ -42,7 +42,6 @@ export class ViewsAggregate {
     thirtyDays: [],
   }
 
-  private allViewsEvents: Partial<UnsequencedVideoEvent>[] = []
   private allVideoViews: EntityViewsInfo[] = []
   private allChannelViews: EntityViewsInfo[] = []
   private allCategoryViews: EntityViewsInfo[] = []
@@ -101,10 +100,6 @@ export class ViewsAggregate {
     return this.channelViewsMap[channelId] ?? null
   }
 
-  public getAllViewsEvents() {
-    return this.allViewsEvents
-  }
-
   public getVideoViewsMap() {
     return Object.freeze(this.videoViewsMap)
   }

src/helpers/index.ts

@@ -0,0 +1 @@
+export { mapPeriods } from './period'

src/helpers/period.ts

@@ -0,0 +1 @@
+export const mapPeriods = (period: number) => (period === 7 ? 'sevenDays' : 'thirtyDays')

src/resolvers/followsInfo.ts

@@ -1,11 +1,9 @@
 import { Args, ArgsType, Ctx, Field, ID, Int, Mutation, Query, Resolver } from 'type-graphql'
-import { Min, Max } from 'class-validator'
+import { Min, Max, IsIn } from 'class-validator'
 import { ChannelFollowsInfo } from '../entities/ChannelFollowsInfo'
 import { ChannelEventType, saveChannelEvent, UnsequencedChannelEvent } from '../models/ChannelEvent'
 import { OrionContext } from '../types'
-import { differenceInCalendarDays } from 'date-fns'
-
-const MAXIMUM_PERIOD = 30
+import { mapPeriods } from '../helpers'
 
 @ArgsType()
 class ChannelFollowsArgs {
@@ -14,11 +12,12 @@ class ChannelFollowsArgs {
 }
 
 @ArgsType()
-class MostFollowedChannelsArgs {
-  @Field(() => Int)
-  @Min(1)
-  @Max(MAXIMUM_PERIOD)
-  period: number
+class MostFollowedArgs {
+  @Field(() => Int, {
+    description: 'timePeriodDays must take one of the following values: 7, 30',
+  })
+  @IsIn([7, 30])
+  timePeriodDays: 7 | 30
 
   @Field(() => Int, { nullable: true })
   limit?: number
@@ -56,10 +55,11 @@ export class ChannelFollowsInfosResolver {
 
   @Query(() => [ChannelFollowsInfo], { description: 'Get list of most followed channels' })
   async mostFollowedChannels(
-    @Args() { period, limit }: MostFollowedChannelsArgs,
+    @Args() { timePeriodDays, limit }: MostFollowedArgs,
     @Ctx() ctx: OrionContext
   ): Promise<ChannelFollowsInfo[]> {
-    return mapMostFollowedArray(buildMostFollowedChannelsArray(ctx, period), limit)
+    ctx.followsAggregate.filterEventsByPeriod(timePeriodDays)
+    return limitFollows(ctx.followsAggregate.getTimePeriodChannelFollows()[mapPeriods(timePeriodDays)], limit)
   }
 
   @Query(() => [ChannelFollowsInfo], { nullable: true, description: 'Get list of most followed channels of all time' })
@@ -67,7 +67,7 @@ export class ChannelFollowsInfosResolver {
     @Args() { limit }: MostFollowedChannelsAllTimeArgs,
     @Ctx() ctx: OrionContext
   ): Promise<ChannelFollowsInfo[]> {
-    return sortAndLimitFollows(ctx.followsAggregate.getAllChannelFollows(), limit)
+    return limitFollows(ctx.followsAggregate.getAllChannelFollows(), limit)
   }
 
   @Query(() => [ChannelFollowsInfo], { description: 'Get follows counts for a list of channels', nullable: 'items' })
@@ -112,37 +112,10 @@ export class ChannelFollowsInfosResolver {
   }
 }
 
-const mapMostFollowedArray = (follows: Record<string, number>, limit?: number) =>
-  follows
-    ? Object.keys(follows)
-        .map((id) => ({ id, follows: follows[id] }))
-        .sort((a, b) => (a.follows > b.follows ? -1 : 1))
-        .slice(0, limit)
-    : []
-
-const sortAndLimitFollows = (follows: ChannelFollowsInfo[], limit: number) => {
+const limitFollows = (follows: ChannelFollowsInfo[], limit?: number) => {
   return follows.sort((a, b) => (a.follows > b.follows ? -1 : 1)).slice(0, limit)
 }
 
-const filterAllFollowsByPeriod = (ctx: OrionContext, period: number): Partial<UnsequencedChannelEvent>[] => {
-  const follows = ctx.followsAggregate.getAllFollowEvents()
-  const filteredFollows = []
-
-  for (let i = follows.length - 1; i >= 0; i--) {
-    const { timestamp } = follows[i]
-    if (timestamp && differenceInCalendarDays(new Date(), timestamp) > period) break
-    filteredFollows.push(follows[i])
-  }
-
-  return filteredFollows
-}
-
-const buildMostFollowedChannelsArray = (ctx: OrionContext, period: number) =>
-  filterAllFollowsByPeriod(ctx, period).reduce(
-    (entity: Record<string, number>, { channelId = '' }) => ({ ...entity, [channelId]: (entity[channelId] || 0) + 1 }),
-    {}
-  )
-
 const getFollowsInfo = (channelId: string, ctx: OrionContext): ChannelFollowsInfo | null => {
   const follows = ctx.followsAggregate.channelFollows(channelId)
   if (follows != null) {

src/resolvers/viewsInfo.ts

@@ -3,8 +3,7 @@ import { Min, Max, IsIn } from 'class-validator'
 import { EntityViewsInfo } from '../entities/EntityViewsInfo'
 import { saveVideoEvent, VideoEventType, UnsequencedVideoEvent } from '../models/VideoEvent'
 import { OrionContext } from '../types'
-
-export const mapPeriods = (period: number) => (period === 7 ? 'sevenDays' : 'thirtyDays')
+import { mapPeriods } from '../helpers'
 
 @ArgsType()
 class VideoViewsArgs {

tests/follows.test.ts

@@ -77,10 +77,10 @@ describe('Channel follows resolver', () => {
     return channelFollowsResponse.data?.channelFollows
   }
 
-  const getMostFollowedChannels = async (period: number) => {
+  const getMostFollowedChannels = async (timePeriodDays: number) => {
     const mostFollowedChannelsResponse = await query<GetMostFollowedChannels, GetMostFollowedChannelsArgs>({
       query: GET_MOST_FOLLOWED_CHANNELS,
-      variables: { period },
+      variables: { timePeriodDays },
     })
     expect(mostFollowedChannelsResponse.errors).toBeUndefined()
     return mostFollowedChannelsResponse.data?.mostFollowedChannels
@@ -100,7 +100,7 @@ describe('Channel follows resolver', () => {
 
   it('should return null for unknown channel follows', async () => {
     const channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
-    const mostFollowedChannels = await getMostFollowedChannels(10)
+    const mostFollowedChannels = await getMostFollowedChannels(30)
     const mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
 
     expect(channelFollows).toBeNull()
@@ -118,7 +118,7 @@ describe('Channel follows resolver', () => {
     expect(addChannelFollowData).toEqual(expectedChannelFollows)
 
     let channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
-    let mostFollowedChannels = await getMostFollowedChannels(10)
+    let mostFollowedChannels = await getMostFollowedChannels(30)
     let mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
     expect(mostFollowedChannels).toEqual([expectedChannelFollows])
@@ -130,7 +130,7 @@ describe('Channel follows resolver', () => {
     expect(addChannelFollowData).toEqual(expectedChannelFollows)
 
     channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
-    mostFollowedChannels = await getMostFollowedChannels(10)
+    mostFollowedChannels = await getMostFollowedChannels(30)
     mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
     expect(mostFollowedChannels).toEqual([expectedChannelFollows])
@@ -150,7 +150,7 @@ describe('Channel follows resolver', () => {
     await followChannel(FIRST_CHANNEL_ID)
 
     let channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
-    let mostFollowedChannels = await getMostFollowedChannels(10)
+    let mostFollowedChannels = await getMostFollowedChannels(30)
     let mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
     expect(mostFollowedChannels).toEqual([expectedChannelFollows])
@@ -162,7 +162,7 @@ describe('Channel follows resolver', () => {
     expect(unfollowChannelData).toEqual(expectedChannelFollows)
 
     channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
-    mostFollowedChannels = await getMostFollowedChannels(10)
+    mostFollowedChannels = await getMostFollowedChannels(30)
     mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
     expect(mostFollowedChannels).toEqual([expectedChannelFollows])
@@ -183,10 +183,10 @@ describe('Channel follows resolver', () => {
     await unfollowChannel(FIRST_CHANNEL_ID)
 
     const channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
-    const mostFollowedChannels = await getMostFollowedChannels(10)
+    const mostFollowedChannels = await getMostFollowedChannels(30)
     const mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
-    expect(mostFollowedChannels).toHaveLength(0)
+    expect(mostFollowedChannels).toEqual([expectedChannelFollows])
     expect(mostFollowedChannelsAllTime).toEqual([expectedChannelFollows])
   })
 
@@ -212,7 +212,7 @@ describe('Channel follows resolver', () => {
 
     const firstChannelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
     const secondChannelFollows = await getChannelFollows(SECOND_CHANNEL_ID)
-    const mostFollowedChannels = await getMostFollowedChannels(10)
+    const mostFollowedChannels = await getMostFollowedChannels(30)
     const mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
 
     expect(firstChannelFollows).toEqual(expectedFirstChannelFollows)
@@ -234,7 +234,7 @@ describe('Channel follows resolver', () => {
     const checkFollows = async () => {
       const firstChannelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
       const secondChannelFollows = await getChannelFollows(SECOND_CHANNEL_ID)
-      const mostFollowedChannels = await getMostFollowedChannels(10)
+      const mostFollowedChannels = await getMostFollowedChannels(30)
       const mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
 
       expect(firstChannelFollows).toEqual(expectedFirstChannelFollows)
@@ -279,7 +279,7 @@ describe('Channel follows resolver', () => {
     }
 
     const channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
-    const mostFollowedChannels = await getMostFollowedChannels(10)
+    const mostFollowedChannels = await getMostFollowedChannels(30)
     const mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
     expect(mostFollowedChannels).toEqual([expectedChannelFollows])

tests/queries/follows.ts

@@ -10,8 +10,8 @@ export const GET_CHANNEL_FOLLOWS = gql`
   }
 `
 export const GET_MOST_FOLLOWED_CHANNELS = gql`
-  query GetMostFollowedChannels($period: Int!) {
-    mostFollowedChannels(period: $period) {
+  query GetMostFollowedChannels($timePeriodDays: Int!) {
+    mostFollowedChannels(timePeriodDays: $timePeriodDays) {
       id
       follows
     }
@@ -33,7 +33,7 @@ export type GetChannelFollowsArgs = {
   channelId: string
 }
 export type GetMostFollowedChannelsArgs = {
-  period: number
+  timePeriodDays: number
 }
 export type GetMostFollowedChannels = {
   mostFollowedChannels: ChannelFollowsInfo[]

tests/server.test.ts

@@ -26,12 +26,10 @@ describe('The server', () => {
   it('should start with empty aggregates', async () => {
     const videoViewsMap = aggregates.viewsAggregate.getVideoViewsMap()
     const channelViewsMap = aggregates.viewsAggregate.getChannelViewsMap()
-    const allViewsEvents = aggregates.viewsAggregate.getAllViewsEvents()
     const channelFollowsMap = aggregates.followsAggregate.getChannelFollowsMap()
 
     expect(videoViewsMap).toEqual({})
     expect(channelViewsMap).toEqual({})
     expect(channelFollowsMap).toEqual({})
-    expect(allViewsEvents).toHaveLength(0)
   })
 })