sepia-search-motore-di-ricerca-video/server/lib/schedulers/videos-indexer.ts

import { AbstractScheduler } from './abstract-scheduler'
import { CONFIG, INDEXER_COUNT, SCHEDULER_INTERVALS_MS } from '../../initializers/constants'
import { doRequest } from '../../helpers/requests'
import { logger } from '../../helpers/logger'
import { ResultList } from '../../../PeerTube/shared/models/result-list.model'
import { Video } from '../../../PeerTube/shared/models/videos/video.model'
import { indexVideos, listIndexInstances, refreshVideosIndex } from '../elastic-search-videos'
import { IndexableVideo } from '../../types/video.model'
import { inspect } from 'util'
import { getRemovedHosts, listIndexInstancesHost } from '../instances-index'
import { elasticSearch } from '../../helpers/elastic-search'

export class VideosIndexer extends AbstractScheduler {

  private static instance: AbstractScheduler

  protected schedulerIntervalMs = SCHEDULER_INTERVALS_MS.videosIndexer

  private constructor () {
    super()
  }

  protected async internalExecute () {
    return this.indexVideos()
  }

  private async indexVideos () {
    const dbHosts = await listIndexInstances()
    const indexHosts = (await listIndexInstancesHost()).filter(h => h === 'peertube.cpy.re')

    const hostsToRemove = getRemovedHosts(dbHosts, indexHosts)
    await this.removeVideosFromHosts(hostsToRemove)

    for (const instance of indexHosts) {
      try {
        let videos: IndexableVideo[] = []
        let start = 0

        do {
          videos = await this.getVideos(instance, start)
          start += videos.length

          logger.debug('Getting %d results from %s (from = %d).', videos.length, instance, start)

          if (videos.length !== 0) {
            await indexVideos(videos)
          }
        } while (videos.length === INDEXER_COUNT.VIDEOS && start < 500)

        logger.info('Added video data from %s.', instance)
      } catch (err) {
        console.error(inspect(err, { depth: 10 }))
        logger.warn('Cannot index videos from %s.', instance, { err })
      }
    }

    await refreshVideosIndex()
  }

  private async getVideos (host: string, start: number): Promise<IndexableVideo[]> {
    const url = 'https://' + host + '/api/v1/videos'

    const res = await doRequest<ResultList<Video>>({
      uri: url,
      qs: {
        start,
        filter: 'local',
        skipCount: true,
        count: INDEXER_COUNT.VIDEOS
      },
      json: true
    })

    return res.body.data.map(v => Object.assign(v, { elasticSearchId: host + v.id, host }))
  }

  private removeVideosFromHosts (hosts: string[]) {
    if (hosts.length === 0) return

    logger.info('Will remove videos from hosts.', { hosts })

    const should = hosts.map(host => ({ term: { host } }))

    return elasticSearch.delete_by_query({
      index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS,
      body: {
        query: {
          bool: {
            should
          }
        }
      }
    })
  }

  static get Instance () {
    return this.instance || (this.instance = new this())
  }
}