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, VideoDetails } from '../../../PeerTube/shared/models/videos/video.model'
import { indexVideos, listIndexInstances, refreshVideosIndex } from '../elastic-search-videos'
import { IndexableVideo, IndexableDoc } from '../../types/video.model'
import { inspect } from 'util'
import { getRemovedHosts, listIndexInstancesHost } from '../instances-index'
import { elasticSearch } from '../../helpers/elastic-search'
import { AsyncQueue, queue } from 'async'

type GetVideoQueueParam = { host: string, uuid: string }

export class VideosIndexer extends AbstractScheduler {

  private static instance: AbstractScheduler

  protected schedulerIntervalMs = SCHEDULER_INTERVALS_MS.videosIndexer

  private readonly getVideoQueue: AsyncQueue<GetVideoQueueParam>

  private constructor () {
    super()

    this.getVideoQueue = queue<GetVideoQueueParam, Error>((task, cb) => {
      this.indexSpecificVideo(task.host, task.uuid)
        .then(() => cb())
        .catch(err => {
          logger.error('Error in index specific video.', { err: inspect(err) })
          cb()
        })
    })
  }

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

  private async runVideosIndexer () {
    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 host of indexHosts) {
      try {
        let videos: IndexableVideo[] = []
        let start = 0

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

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

          if (videos.length !== 0) {
            const { created } = await indexVideos(videos)

            // Fetch complete video foreach created video (to get tags)
            for (const c of created) {
              this.getVideoQueue.push({ uuid: c, host })
            }
          }
        } while (videos.length === INDEXER_COUNT.VIDEOS && start < 500)

        logger.info('Added video data from %s.', host)
      } catch (err) {
        console.error(inspect(err, { depth: 10 }))
        logger.warn('Cannot index videos from %s.', host, { 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 => this.prepareVideoForDB(v, host))
  }

  private async getVideo (host: string, uuid: string): Promise<IndexableVideo> {
    const url = 'https://' + host + '/api/v1/videos/' + uuid

    const res = await doRequest<VideoDetails>({
      uri: url,
      json: true
    })

    return this.prepareVideoForDB(res.body, host)
  }

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

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

    return elasticSearch.delete_by_query({
      index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS,
      body: {
        query: {
          bool: {
            filter: {
              terms: {
                host: hosts
              }
            }
          }
        }
      }
    })
  }

  private async indexSpecificVideo (host: string, uuid: string) {
    const video = await this.getVideo(host, uuid)

    logger.info('Indexing specific video %s of %s.', uuid, host)

    await indexVideos([ video ])
  }

  private prepareVideoForDB <T extends Video> (video: T, host: string): T & IndexableDoc {
    return Object.assign(video, { elasticSearchId: host + video.id, host })
  }

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