148 lines
4.4 KiB
TypeScript
148 lines
4.4 KiB
TypeScript
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()
|
|
})
|
|
})
|
|
}
|
|
|
|
scheduleVideoIndexation (host: string, uuid: string) {
|
|
this.getVideoQueue.push({ uuid, host })
|
|
}
|
|
|
|
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.scheduleVideoIndexation(host, c)
|
|
}
|
|
}
|
|
} 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 ], true)
|
|
}
|
|
|
|
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())
|
|
}
|
|
}
|