Add most followed channels queries (#17)

* Add most followed channels queries

* PR FIX

* PR FIX 2

* PR FIX 3

* PR FIX 4

schema.graphql

@@ -40,6 +40,12 @@ type Query {
   """Get views count for a single channel"""
   channelViews(channelId: ID!): EntityViewsInfo
 
+  """Get list of most followed channels"""
+  mostFollowedChannels(limit: Int, period: Int!): [ChannelFollowsInfo!]!
+
+  """Get list of most followed channels of all time"""
+  mostFollowedChannelsAllTime(limit: Int!): [ChannelFollowsInfo!]
+
   """Get most viewed list of categories"""
   mostViewedCategories(limit: Int, period: Int!): [EntityViewsInfo!]
 

src/aggregates/follows.ts

@@ -6,12 +6,39 @@ import {
   UnsequencedChannelEvent,
 } from '../models/ChannelEvent'
 
+import { ChannelFollowsInfo } from '../entities/ChannelFollowsInfo'
+
 type ChannelEventsAggregationResult = {
   events?: ChannelEvent[]
 }[]
 
 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 addOrRemoveFollowEvent(eventType: ChannelEventType, { channelId, timestamp }: UnsequencedChannelEvent): void {
+    if (eventType === ChannelEventType.FollowChannel) {
+      this.allChannelFollowEvents = [...this.allChannelFollowEvents, { channelId, timestamp }]
+    }
+    if (eventType === ChannelEventType.UnfollowChannel) {
+      const index = this.allChannelFollowEvents.findIndex((item) => item.channelId === channelId)
+      if (index >= 0) {
+        this.allChannelFollowEvents.splice(index, 1)
+      }
+    }
+  }
 
   public channelFollows(channelId: string): number | null {
     return this.channelFollowsMap[channelId] ?? null
@@ -21,6 +48,14 @@ export class FollowsAggregate implements GenericAggregate<ChannelEvent> {
     return Object.freeze(this.channelFollowsMap)
   }
 
+  public getAllFollowEvents() {
+    return this.allChannelFollowEvents
+  }
+
+  public getAllChannelFollows() {
+    return this.allChannelFollows
+  }
+
   public static async Build(): Promise<FollowsAggregate> {
     const aggregation: ChannelEventsAggregationResult = await ChannelEventsBucketModel.aggregate([
       { $unwind: '$events' },
@@ -38,17 +73,22 @@ export class FollowsAggregate implements GenericAggregate<ChannelEvent> {
   }
 
   public applyEvent(event: UnsequencedChannelEvent) {
-    const currentChannelFollows = this.channelFollowsMap[event.channelId] || 0
+    const { channelId, type } = event
+    const currentChannelFollows = this.channelFollowsMap[channelId] || 0
 
     switch (event.type) {
       case ChannelEventType.FollowChannel:
-        this.channelFollowsMap[event.channelId] = currentChannelFollows + 1
+        this.channelFollowsMap[channelId] = currentChannelFollows + 1
+        this.addOrUpdateFollows(channelId, ChannelEventType.FollowChannel)
+        this.addOrRemoveFollowEvent(ChannelEventType.FollowChannel, event)
         break
       case ChannelEventType.UnfollowChannel:
-        this.channelFollowsMap[event.channelId] = Math.max(currentChannelFollows - 1, 0)
+        this.channelFollowsMap[channelId] = Math.max(currentChannelFollows - 1, 0)
+        this.addOrUpdateFollows(channelId, ChannelEventType.UnfollowChannel)
+        this.addOrRemoveFollowEvent(ChannelEventType.UnfollowChannel, event)
         break
       default:
-        console.error(`Parsing unknown channel event: ${event.type}`)
+        console.error(`Parsing unknown channel event: ${type}`)
     }
   }
 }

src/resolvers/followsInfo.ts

@@ -1,7 +1,11 @@
-import { Args, ArgsType, Ctx, Field, ID, Mutation, Query, Resolver } from 'type-graphql'
+import { Args, ArgsType, Ctx, Field, ID, Int, Mutation, Query, Resolver } from 'type-graphql'
+import { Min, Max } 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
 
 @ArgsType()
 class ChannelFollowsArgs {
@@ -9,6 +13,25 @@ class ChannelFollowsArgs {
   channelId: string
 }
 
+@ArgsType()
+class MostFollowedChannelsArgs {
+  @Field(() => Int)
+  @Min(1)
+  @Max(MAXIMUM_PERIOD)
+  period: number
+
+  @Field(() => Int, { nullable: true })
+  limit?: number
+}
+
+@ArgsType()
+class MostFollowedChannelsAllTimeArgs {
+  @Field(() => Int)
+  @Min(1)
+  @Max(200)
+  limit: number
+}
+
 @ArgsType()
 class BatchedChannelFollowsArgs {
   @Field(() => [ID])
@@ -31,6 +54,22 @@ export class ChannelFollowsInfosResolver {
     return getFollowsInfo(channelId, ctx)
   }
 
+  @Query(() => [ChannelFollowsInfo], { description: 'Get list of most followed channels' })
+  async mostFollowedChannels(
+    @Args() { period, limit }: MostFollowedChannelsArgs,
+    @Ctx() ctx: OrionContext
+  ): Promise<ChannelFollowsInfo[]> {
+    return mapMostFollowedArray(buildMostFollowedChannelsArray(ctx, period), limit)
+  }
+
+  @Query(() => [ChannelFollowsInfo], { nullable: true, description: 'Get list of most followed channels of all time' })
+  async mostFollowedChannelsAllTime(
+    @Args() { limit }: MostFollowedChannelsAllTimeArgs,
+    @Ctx() ctx: OrionContext
+  ): Promise<ChannelFollowsInfo[]> {
+    return sortAndLimitFollows(ctx.followsAggregate.getAllChannelFollows(), limit)
+  }
+
   @Query(() => [ChannelFollowsInfo], { description: 'Get follows counts for a list of channels', nullable: 'items' })
   async batchedChannelFollows(
     @Args() { channelIdList }: BatchedChannelFollowsArgs,
@@ -73,6 +112,37 @@ 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) => {
+  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

@@ -5,6 +5,8 @@ import { EntityViewsInfo } from '../entities/EntityViewsInfo'
 import { saveVideoEvent, VideoEventType, UnsequencedVideoEvent } from '../models/VideoEvent'
 import { OrionContext } from '../types'
 
+const MAXIMUM_PERIOD = 30
+
 @ArgsType()
 class VideoViewsArgs {
   @Field(() => ID)
@@ -21,7 +23,7 @@ class BatchedVideoViewsArgs {
 class MostViewedVideosArgs {
   @Field(() => Int)
   @Min(1)
-  @Max(30)
+  @Max(MAXIMUM_PERIOD)
   period: number
 
   @Field(() => Int, { nullable: true })
@@ -32,7 +34,7 @@ class MostViewedVideosArgs {
 class MostViewedChannelArgs {
   @Field(() => Int)
   @Min(1)
-  @Max(30)
+  @Max(MAXIMUM_PERIOD)
   period: number
 
   @Field(() => Int, { nullable: true })
@@ -55,7 +57,7 @@ class BatchedChannelViewsArgs {
 class MostViewedCategoriesArgs {
   @Field(() => Int)
   @Min(1)
-  @Max(30)
+  @Max(MAXIMUM_PERIOD)
   period: number
 
   @Field(() => Int, { nullable: true })

tests/follows.test.ts

@@ -14,6 +14,12 @@ import {
   UNFOLLOW_CHANNEL,
   UnfollowChannel,
   UnfollowChannelArgs,
+  GET_MOST_FOLLOWED_CHANNELS,
+  GetMostFollowedChannels,
+  GetMostFollowedChannelsArgs,
+  GET_MOST_FOLLOWED_CHANNELS_ALL_TIME,
+  GetMostFollowedChannelsAllTime,
+  GetMostFollowedChannelsAllTimeArgs,
 } from './queries/follows'
 import { ChannelFollowsInfo } from '../src/entities/ChannelFollowsInfo'
 import { ChannelEventsBucketModel } from '../src/models/ChannelEvent'
@@ -71,10 +77,35 @@ describe('Channel follows resolver', () => {
     return channelFollowsResponse.data?.channelFollows
   }
 
+  const getMostFollowedChannels = async (period: number) => {
+    const mostFollowedChannelsResponse = await query<GetMostFollowedChannels, GetMostFollowedChannelsArgs>({
+      query: GET_MOST_FOLLOWED_CHANNELS,
+      variables: { period },
+    })
+    expect(mostFollowedChannelsResponse.errors).toBeUndefined()
+    return mostFollowedChannelsResponse.data?.mostFollowedChannels
+  }
+
+  const getMostFollowedChannelsAllTime = async (limit: number) => {
+    const mostFollowedChannelsAllTimeResponse = await query<
+      GetMostFollowedChannelsAllTime,
+      GetMostFollowedChannelsAllTimeArgs
+    >({
+      query: GET_MOST_FOLLOWED_CHANNELS_ALL_TIME,
+      variables: { limit },
+    })
+    expect(mostFollowedChannelsAllTimeResponse.errors).toBeUndefined()
+    return mostFollowedChannelsAllTimeResponse.data?.mostFollowedChannelsAllTime
+  }
+
   it('should return null for unknown channel follows', async () => {
     const channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
+    const mostFollowedChannels = await getMostFollowedChannels(10)
+    const mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
 
     expect(channelFollows).toBeNull()
+    expect(mostFollowedChannels).toHaveLength(0)
+    expect(mostFollowedChannelsAllTime).toHaveLength(0)
   })
 
   it('should properly handle channel follow', async () => {
@@ -87,7 +118,11 @@ describe('Channel follows resolver', () => {
     expect(addChannelFollowData).toEqual(expectedChannelFollows)
 
     let channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
+    let mostFollowedChannels = await getMostFollowedChannels(10)
+    let mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
+    expect(mostFollowedChannels).toEqual([expectedChannelFollows])
+    expect(mostFollowedChannelsAllTime).toEqual([expectedChannelFollows])
 
     expectedChannelFollows.follows++
 
@@ -95,7 +130,11 @@ describe('Channel follows resolver', () => {
     expect(addChannelFollowData).toEqual(expectedChannelFollows)
 
     channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
+    mostFollowedChannels = await getMostFollowedChannels(10)
+    mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
+    expect(mostFollowedChannels).toEqual([expectedChannelFollows])
+    expect(mostFollowedChannelsAllTime).toEqual([expectedChannelFollows])
   })
 
   it('should properly handle channel unfollow', async () => {
@@ -111,7 +150,11 @@ describe('Channel follows resolver', () => {
     await followChannel(FIRST_CHANNEL_ID)
 
     let channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
+    let mostFollowedChannels = await getMostFollowedChannels(10)
+    let mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
+    expect(mostFollowedChannels).toEqual([expectedChannelFollows])
+    expect(mostFollowedChannelsAllTime).toEqual([expectedChannelFollows])
 
     expectedChannelFollows.follows--
 
@@ -119,7 +162,11 @@ describe('Channel follows resolver', () => {
     expect(unfollowChannelData).toEqual(expectedChannelFollows)
 
     channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
+    mostFollowedChannels = await getMostFollowedChannels(10)
+    mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
+    expect(mostFollowedChannels).toEqual([expectedChannelFollows])
+    expect(mostFollowedChannelsAllTime).toEqual([expectedChannelFollows])
   })
 
   it('should keep the channel follows non-negative', async () => {
@@ -136,7 +183,11 @@ describe('Channel follows resolver', () => {
     await unfollowChannel(FIRST_CHANNEL_ID)
 
     const channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
+    const mostFollowedChannels = await getMostFollowedChannels(10)
+    const mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
+    expect(mostFollowedChannels).toHaveLength(0)
+    expect(mostFollowedChannelsAllTime).toEqual([expectedChannelFollows])
   })
 
   it('should distinct follows of separate channels', async () => {
@@ -161,9 +212,13 @@ describe('Channel follows resolver', () => {
 
     const firstChannelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
     const secondChannelFollows = await getChannelFollows(SECOND_CHANNEL_ID)
+    const mostFollowedChannels = await getMostFollowedChannels(10)
+    const mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
 
     expect(firstChannelFollows).toEqual(expectedFirstChannelFollows)
     expect(secondChannelFollows).toEqual(expectedSecondChannelFollows)
+    expect(mostFollowedChannels).toEqual([expectedFirstChannelFollows, expectedSecondChannelFollows])
+    expect(mostFollowedChannelsAllTime).toEqual([expectedFirstChannelFollows, expectedSecondChannelFollows])
   })
 
   it('should properly rebuild the aggregate', async () => {
@@ -179,9 +234,13 @@ 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 mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
 
       expect(firstChannelFollows).toEqual(expectedFirstChannelFollows)
       expect(secondChannelFollows).toEqual(expectedSecondChannelFollows)
+      expect(mostFollowedChannels).toEqual([expectedSecondChannelFollows, expectedFirstChannelFollows])
+      expect(mostFollowedChannelsAllTime).toEqual([expectedSecondChannelFollows, expectedFirstChannelFollows])
     }
 
     await followChannel(FIRST_CHANNEL_ID)
@@ -220,6 +279,10 @@ describe('Channel follows resolver', () => {
     }
 
     const channelFollows = await getChannelFollows(FIRST_CHANNEL_ID)
+    const mostFollowedChannels = await getMostFollowedChannels(10)
+    const mostFollowedChannelsAllTime = await getMostFollowedChannelsAllTime(10)
     expect(channelFollows).toEqual(expectedChannelFollows)
+    expect(mostFollowedChannels).toEqual([expectedChannelFollows])
+    expect(mostFollowedChannelsAllTime).toEqual([expectedChannelFollows])
   })
 })

tests/queries/follows.ts

@@ -9,12 +9,41 @@ export const GET_CHANNEL_FOLLOWS = gql`
     }
   }
 `
+export const GET_MOST_FOLLOWED_CHANNELS = gql`
+  query GetMostFollowedChannels($period: Int!) {
+    mostFollowedChannels(period: $period) {
+      id
+      follows
+    }
+  }
+`
+
+export const GET_MOST_FOLLOWED_CHANNELS_ALL_TIME = gql`
+  query GetMostFollowedChannelsAllTime($limit: Int!) {
+    mostFollowedChannelsAllTime(limit: $limit) {
+      id
+      follows
+    }
+  }
+`
 export type GetChannelFollows = {
   channelFollows: ChannelFollowsInfo | null
 }
 export type GetChannelFollowsArgs = {
   channelId: string
 }
+export type GetMostFollowedChannelsArgs = {
+  period: number
+}
+export type GetMostFollowedChannels = {
+  mostFollowedChannels: ChannelFollowsInfo[]
+}
+export type GetMostFollowedChannelsAllTime = {
+  mostFollowedChannelsAllTime: ChannelFollowsInfo[]
+}
+export type GetMostFollowedChannelsAllTimeArgs = {
+  limit: number
+}
 
 export const FOLLOW_CHANNEL = gql`
   mutation FollowChannel($channelId: ID!) {