Index and serve channels

This commit is contained in:
Chocobozzz 2020-02-19 15:39:35 +01:00
parent 23f0d843c3
commit 223c18c7c0
No known key found for this signature in database
GPG Key ID: 583A612D890159BE
20 changed files with 623 additions and 218 deletions

View File

@ -12,6 +12,7 @@ elastic_search:
port: 9200 port: 9200
indexes: indexes:
videos: 'peertube-index-videos' videos: 'peertube-index-videos'
channels: 'peertube-index-channels'
log: log:
level: 'debug' # debug/info/warning/error level: 'debug' # debug/info/warning/error

View File

@ -1,3 +1,4 @@
elastic_search: elastic_search:
indexes: indexes:
videos: 'peertube-index-videos-test1' videos: 'peertube-index-videos-test1'
channels: 'peertube-index-channels-test1'

View File

@ -1,8 +1,4 @@
import { isTestInstance } from './server/helpers/core-utils' import { isTestInstance } from './server/helpers/core-utils'
if (isTestInstance()) {
require('source-map-support').install()
}
import * as bodyParser from 'body-parser' import * as bodyParser from 'body-parser'
import * as express from 'express' import * as express from 'express'
import * as cors from 'cors' import * as cors from 'cors'
@ -12,6 +8,11 @@ import { logger } from './server/helpers/logger'
import { API_VERSION, CONFIG } from './server/initializers/constants' import { API_VERSION, CONFIG } from './server/initializers/constants'
import { VideosIndexer } from './server/lib/schedulers/videos-indexer' import { VideosIndexer } from './server/lib/schedulers/videos-indexer'
import { initVideosIndex } from './server/lib/elastic-search-videos' import { initVideosIndex } from './server/lib/elastic-search-videos'
import { initChannelsIndex } from './server/lib/elastic-search-channels'
if (isTestInstance()) {
require('source-map-support').install()
}
const app = express() const app = express()
@ -57,7 +58,10 @@ app.listen(CONFIG.LISTEN.PORT, async () => {
logger.info('Server listening on port %d', CONFIG.LISTEN.PORT) logger.info('Server listening on port %d', CONFIG.LISTEN.PORT)
try { try {
await initVideosIndex() await Promise.all([
initVideosIndex(),
initChannelsIndex()
])
} catch (err) { } catch (err) {
logger.error('Cannot init videos index.', { err }) logger.error('Cannot init videos index.', { err })
process.exit(-1) process.exit(-1)
@ -65,4 +69,5 @@ app.listen(CONFIG.LISTEN.PORT, async () => {
VideosIndexer.Instance.enable() VideosIndexer.Instance.enable()
VideosIndexer.Instance.execute() VideosIndexer.Instance.execute()
.catch(err => logger.error('Cannot run video indexer', { err }))
}) })

View File

@ -1,10 +1,12 @@
import * as express from 'express' import * as express from 'express'
import { badRequest } from '../../helpers/utils' import { badRequest } from '../../helpers/utils'
import { searchRouter } from './search-videos' import { searchVideosRouter } from './search-videos'
import { searchChannelsRouter } from './search-channels'
const apiRouter = express.Router() const apiRouter = express.Router()
apiRouter.use('/', searchRouter) apiRouter.use('/', searchVideosRouter)
apiRouter.use('/', searchChannelsRouter)
apiRouter.use('/ping', pong) apiRouter.use('/ping', pong)
apiRouter.use('/*', badRequest) apiRouter.use('/*', badRequest)

View File

@ -0,0 +1,34 @@
import * as express from 'express'
import { paginationValidator } from '../../middlewares/validators/pagination'
import { setDefaultPagination } from '../../middlewares/pagination'
import { asyncMiddleware } from '../../middlewares/async'
import { channelsSearchSortValidator } from '../../middlewares/validators/sort'
import { videoChannelsSearchValidator } from '../../middlewares/validators/search'
import { setDefaultSearchSort } from '../../middlewares/sort'
import { formatChannelForAPI, queryChannels } from '../../lib/elastic-search-channels'
const searchChannelsRouter = express.Router()
searchChannelsRouter.get('/search/video-channels',
paginationValidator,
setDefaultPagination,
channelsSearchSortValidator,
setDefaultSearchSort,
videoChannelsSearchValidator,
asyncMiddleware(searchChannels)
)
// ---------------------------------------------------------------------------
export { searchChannelsRouter }
// ---------------------------------------------------------------------------
async function searchChannels (req: express.Request, res: express.Response) {
const resultList = await queryChannels(req.query)
return res.json({
total: resultList.total,
data: resultList.data.map(v => formatChannelForAPI(v, req.query.fromHost))
})
}

View File

@ -7,9 +7,9 @@ import { videosSearchSortValidator } from '../../middlewares/validators/sort'
import { commonVideosFiltersValidator, videosSearchValidator } from '../../middlewares/validators/search' import { commonVideosFiltersValidator, videosSearchValidator } from '../../middlewares/validators/search'
import { setDefaultSearchSort } from '../../middlewares/sort' import { setDefaultSearchSort } from '../../middlewares/sort'
const searchRouter = express.Router() const searchVideosRouter = express.Router()
searchRouter.get('/search/videos', searchVideosRouter.get('/search/videos',
paginationValidator, paginationValidator,
setDefaultPagination, setDefaultPagination,
videosSearchSortValidator, videosSearchSortValidator,
@ -21,7 +21,7 @@ searchRouter.get('/search/videos',
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
export { searchRouter } export { searchVideosRouter }
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------

View File

@ -1,24 +1,120 @@
import { Client } from '@elastic/elasticsearch' import { ApiResponse, Client } from '@elastic/elasticsearch'
import { CONFIG } from '../initializers/constants' import { CONFIG } from '../initializers/constants'
import { logger } from './logger'
import { flatMap } from 'lodash'
import { IndexableDoc } from '../types/elastic-search.model'
const elasticSearch = new Client({ node: 'http://' + CONFIG.ELASTIC_SEARCH.HOSTNAME + ':' + CONFIG.ELASTIC_SEARCH.PORT }) const elasticSearch = new Client({ node: 'http://' + CONFIG.ELASTIC_SEARCH.HOSTNAME + ':' + CONFIG.ELASTIC_SEARCH.PORT })
function buildSort (value: string) { function buildSort (value: string) {
let field: string let sortField: string
let direction: 'asc' | 'desc' let direction: 'asc' | 'desc'
if (value.substring(0, 1) === '-') { if (value.substring(0, 1) === '-') {
direction = 'desc' direction = 'desc'
field = value.substring(1) sortField = value.substring(1)
} else { } else {
direction = 'asc' direction = 'asc'
field = value sortField = value
} }
return { direction, field } const field = sortField === 'match'
? '_score'
: sortField
return [
{
[field]: { order: direction }
}
]
}
function buildIndex (name: string, mapping: object) {
logger.info('Initialize %s Elastic Search index.', name)
return elasticSearch.indices.create({
index: name,
body: {
settings: {
number_of_shards: 1,
number_of_replicas: 1
},
mappings: {
properties: mapping
}
}
}).catch(err => {
if (err.name === 'ResponseError' && err.meta?.body?.error.root_cause[0]?.type === 'resource_already_exists_exception') return
throw err
})
}
async function indexDocuments <T extends IndexableDoc> (options: {
objects: T[]
formatter: (o: T) => any
replace: boolean
index: string
}) {
const { objects, formatter, replace, index } = options
const elIdIndex: { [elId: string]: T } = {}
for (const object of objects) {
elIdIndex[object.elasticSearchId] = object
}
const method = replace ? 'index' : 'update'
const body = flatMap(objects, v => {
const doc = formatter(v)
const options = replace
? doc
: { doc, doc_as_upsert: true }
return [
{
[method]: {
_id: v.elasticSearchId,
_index: index
}
},
options
]
})
const result = await elasticSearch.bulk({
index,
body
})
const resultBody = result.body
if (resultBody.errors === true) {
const msg = 'Cannot insert data in elastic search.'
logger.error(msg, { err: resultBody })
throw new Error(msg)
}
const created: T[] = result.body.items
.map(i => i[method])
.filter(i => i.result === 'created')
.map(i => elIdIndex[i._id])
return { created }
}
function extractQueryResult (result: ApiResponse<any, any>) {
const hits = result.body.hits
return { total: hits.total.value, data: hits.hits.map(h => h._source) }
} }
export { export {
elasticSearch, elasticSearch,
buildSort indexDocuments,
buildSort,
extractQueryResult,
buildIndex
} }

View File

@ -11,7 +11,8 @@ const CONFIG = {
HOSTNAME: config.get<string>('elastic_search.hostname'), HOSTNAME: config.get<string>('elastic_search.hostname'),
PORT: config.get<number>('elastic_search.port'), PORT: config.get<number>('elastic_search.port'),
INDEXES: { INDEXES: {
VIDEOS: config.get<string>('elastic_search.indexes.videos') VIDEOS: config.get<string>('elastic_search.indexes.videos'),
CHANNELS: config.get<string>('elastic_search.indexes.channels')
} }
}, },
LOG: { LOG: {
@ -23,7 +24,8 @@ const CONFIG = {
} }
const SORTABLE_COLUMNS = { const SORTABLE_COLUMNS = {
VIDEOS_SEARCH: [ 'name', 'duration', 'createdAt', 'publishedAt', 'views', 'likes', 'match' ] VIDEOS_SEARCH: [ 'name', 'duration', 'createdAt', 'publishedAt', 'views', 'likes', 'match' ],
CHANNELS_SEARCH: [ 'match', 'displayName', 'createdAt' ]
} }
const PAGINATION_COUNT_DEFAULT = 20 const PAGINATION_COUNT_DEFAULT = 20

View File

@ -0,0 +1,43 @@
import { Avatar } from '@shared/models'
function formatAvatarForAPI (obj: { avatar?: Avatar }) {
if (!obj.avatar) return null
return {
path: obj.avatar.path,
createdAt: obj.avatar.createdAt,
updatedAt: obj.avatar.updatedAt
}
}
function formatAvatarForDB (obj: { avatar?: Avatar }) {
if (!obj.avatar) return null
return {
path: obj.avatar.path,
createdAt: obj.avatar.createdAt,
updatedAt: obj.avatar.updatedAt
}
}
function buildAvatarMapping () {
return {
path: {
type: 'keyword'
},
createdAt: {
type: 'date',
format: 'date_optional_time'
},
updatedAt: {
type: 'date',
format: 'date_optional_time'
}
}
}
export {
formatAvatarForAPI,
formatAvatarForDB,
buildAvatarMapping
}

View File

@ -0,0 +1,213 @@
import { CONFIG } from '../initializers/constants'
import { VideoChannel } from '@shared/models'
import { buildIndex, buildSort, elasticSearch, extractQueryResult, indexDocuments } from '../helpers/elastic-search'
import { logger } from '../helpers/logger'
import { DBChannel, IndexableChannel } from '../types/channel.model'
import { ChannelsSearchQuery } from '../types/channel-search.model'
import { buildAvatarMapping, formatAvatarForAPI, formatAvatarForDB } from './elastic-search-avatar'
function initChannelsIndex () {
return buildIndex(CONFIG.ELASTIC_SEARCH.INDEXES.CHANNELS, buildChannelsMapping())
}
async function indexChannels (channels: IndexableChannel[], replace = false) {
return indexDocuments({
objects: channels,
formatter: c => formatChannelForDB(c),
replace,
index: CONFIG.ELASTIC_SEARCH.INDEXES.CHANNELS
})
}
function refreshChannelsIndex () {
return elasticSearch.indices.refresh({ index: CONFIG.ELASTIC_SEARCH.INDEXES.CHANNELS })
}
async function queryChannels (search: ChannelsSearchQuery) {
const bool: any = {}
if (search.search) {
Object.assign(bool, {
must: [
{
multi_match: {
query: search.search,
fields: [ 'name', 'displayName', 'description' ],
fuzziness: 'AUTO'
}
}
]
})
}
const body = {
from: search.start,
size: search.count,
sort: buildSort(search.sort),
query: { bool }
}
logger.debug('Will query Elastic Search for channels.', { body })
const res = await elasticSearch.search({
index: CONFIG.ELASTIC_SEARCH.INDEXES.CHANNELS,
body
})
return extractQueryResult(res)
}
export {
initChannelsIndex,
indexChannels,
refreshChannelsIndex,
formatChannelForAPI,
queryChannels
}
// ############################################################################
function formatChannelForDB (c: IndexableChannel): DBChannel {
return {
id: c.id,
url: c.url,
name: c.name,
host: c.host,
avatar: formatAvatarForDB(c),
displayName: c.displayName,
indexedAt: new Date(),
followingCount: c.followingCount,
followersCount: c.followersCount,
createdAt: c.createdAt,
updatedAt: c.updatedAt,
description: c.description,
support: c.support,
ownerAccount: {
id: c.ownerAccount.id,
url: c.ownerAccount.url,
displayName: c.ownerAccount.displayName,
description: c.ownerAccount.description,
name: c.ownerAccount.name,
host: c.ownerAccount.host,
followingCount: c.ownerAccount.followingCount,
followersCount: c.ownerAccount.followersCount,
createdAt: c.ownerAccount.createdAt,
updatedAt: c.ownerAccount.updatedAt,
avatar: formatAvatarForDB(c.ownerAccount)
}
}
}
function formatChannelForAPI (c: DBChannel, fromHost?: string): VideoChannel {
return {
id: c.id,
url: c.url,
name: c.name,
host: c.host,
followingCount: c.followingCount,
followersCount: c.followersCount,
createdAt: c.createdAt,
updatedAt: c.updatedAt,
avatar: formatAvatarForAPI(c),
displayName: c.displayName,
description: c.description,
support: c.support,
isLocal: fromHost === c.host,
ownerAccount: {
id: c.ownerAccount.id,
url: c.ownerAccount.url,
displayName: c.ownerAccount.displayName,
description: c.ownerAccount.description,
name: c.ownerAccount.name,
host: c.ownerAccount.host,
followingCount: c.ownerAccount.followingCount,
followersCount: c.ownerAccount.followersCount,
createdAt: c.ownerAccount.createdAt,
updatedAt: c.ownerAccount.updatedAt,
avatar: formatAvatarForAPI(c.ownerAccount)
}
}
}
function buildChannelOrAccountCommonMapping () {
return {
id: {
type: 'long'
},
url: {
type: 'keyword'
},
name: {
type: 'text',
fields: {
raw: {
type: 'keyword'
}
}
},
host: {
type: 'keyword'
},
displayName: {
type: 'text'
},
avatar: {
properties: buildAvatarMapping()
},
followingCount: {
type: 'long'
},
followersCount: {
type: 'long'
},
createdAt: {
type: 'date',
format: 'date_optional_time'
},
updatedAt: {
type: 'date',
format: 'date_optional_time'
},
description: {
type: 'text'
}
}
}
function buildChannelsMapping () {
const base = buildChannelOrAccountCommonMapping()
Object.assign(base, {
support: {
type: 'keyword'
},
ownerAccount: {
properties: buildChannelOrAccountCommonMapping()
}
})
return base
}

View File

@ -0,0 +1,35 @@
import { elasticSearch } from '../helpers/elastic-search'
import { CONFIG } from '../initializers/constants'
import { getRemovedHosts, listIndexInstancesHost } from './instances-index'
async function listIndexInstances () {
const res = await elasticSearch.search({
index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS,
body: {
size: 0,
aggs: {
hosts: {
terms: {
field: 'host'
}
}
}
}
})
return res.body.aggregations.hosts.buckets.map(b => b.key)
}
async function buildInstanceHosts () {
const indexHosts = (await listIndexInstancesHost()).filter(h => h === 'peertube.cpy.re')
const dbHosts = await listIndexInstances()
const removedHosts = getRemovedHosts(dbHosts, indexHosts)
return { indexHosts, removedHosts }
}
export {
listIndexInstances,
buildInstanceHosts
}

View File

@ -1,100 +1,47 @@
import { CONFIG } from '../initializers/constants' import { CONFIG } from '../initializers/constants'
import { DBVideo, DBVideoDetails, IndexableVideo, IndexableVideoDetails } from '../types/video.model' import { DBVideo, DBVideoDetails, IndexableVideo, IndexableVideoDetails } from '../types/video.model'
import { flatMap } from 'lodash' import { Video } from '@shared/models'
import { Avatar, Video } from '@shared/models' import { buildIndex, buildSort, elasticSearch, extractQueryResult, indexDocuments } from '../helpers/elastic-search'
import { buildSort, elasticSearch } from '../helpers/elastic-search'
import { VideosSearchQuery } from '../types/video-search.model' import { VideosSearchQuery } from '../types/video-search.model'
import { logger } from '../helpers/logger' import { logger } from '../helpers/logger'
import { buildAvatarMapping, formatAvatarForAPI, formatAvatarForDB } from './elastic-search-avatar'
function initVideosIndex () { function initVideosIndex () {
logger.info('Initialize %s Elastic Search index.', CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS) return buildIndex(CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS, buildVideosMapping())
return elasticSearch.indices.create({
index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS,
body: {
settings: {
number_of_shards: 1,
number_of_replicas: 1
},
mappings: {
properties: buildVideosMapping()
}
}
}).catch(err => {
if (err.name === 'ResponseError' && err.meta?.body?.error.root_cause[0]?.type === 'resource_already_exists_exception') return
throw err
})
} }
async function indexVideos (videos: IndexableVideo[], replace = false) { async function indexVideos (videos: IndexableVideo[], replace = false) {
const elIdIndex: { [elId: string]: string } = {} return indexDocuments({
objects: videos,
for (const video of videos) { formatter: v => formatVideoForDB(v),
elIdIndex[video.elasticSearchId] = video.uuid replace,
} index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS
const method = replace ? 'index' : 'update'
const body = flatMap(videos, v => {
const doc = formatVideoForDB(v)
const options = replace
? doc
: { doc, doc_as_upsert: true }
return [
{
[method]: {
_id: v.elasticSearchId,
_index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS
}
},
options
]
}) })
const result = await elasticSearch.bulk({
index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS,
body
})
const resultBody = result.body
if (resultBody.errors === true) {
const msg = 'Cannot insert data in elastic search.'
logger.error(msg, { err: resultBody })
throw new Error(msg)
}
const created: string[] = result.body.items
.map(i => i[method])
.filter(i => i.result === 'created')
.map(i => elIdIndex[i._id])
return { created }
} }
function refreshVideosIndex () { function refreshVideosIndex () {
return elasticSearch.indices.refresh({ index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS }) return elasticSearch.indices.refresh({ index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS })
} }
async function listIndexInstances () { function removeVideosFromHosts (hosts: string[]) {
const res = await elasticSearch.search({ if (hosts.length === 0) return
logger.info('Will remove videos from hosts.', { hosts })
return elasticSearch.delete_by_query({
index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS, index: CONFIG.ELASTIC_SEARCH.INDEXES.VIDEOS,
body: { body: {
size: 0, query: {
aggs: { bool: {
hosts: { filter: {
terms: { terms: {
field: 'host' host: hosts
}
} }
} }
} }
} }
}) })
return res.body.aggregations.hosts.buckets.map(b => b.key)
} }
async function queryVideos (search: VideosSearchQuery) { async function queryVideos (search: VideosSearchQuery) {
@ -230,7 +177,7 @@ async function queryVideos (search: VideosSearchQuery) {
const body = { const body = {
from: search.start, from: search.start,
size: search.count, size: search.count,
sort: buildVideosSort(search.sort), sort: buildSort(search.sort),
query: { bool } query: { bool }
} }
@ -241,35 +188,20 @@ async function queryVideos (search: VideosSearchQuery) {
body body
}) })
const hits = res.body.hits return extractQueryResult(res)
return { total: hits.total.value, data: hits.hits.map(h => h._source) }
} }
export { export {
indexVideos, indexVideos,
queryVideos, queryVideos,
refreshVideosIndex, refreshVideosIndex,
removeVideosFromHosts,
initVideosIndex, initVideosIndex,
listIndexInstances,
formatVideoForAPI formatVideoForAPI
} }
// ############################################################################ // ############################################################################
function buildVideosSort (sort: string) {
const { direction, field: sortField } = buildSort(sort)
const field = sortField === 'match'
? '_score'
: sortField
return [
{
[field]: { order: direction }
}
]
}
function formatVideoForDB (v: IndexableVideo | IndexableVideoDetails): DBVideo | DBVideoDetails { function formatVideoForDB (v: IndexableVideo | IndexableVideoDetails): DBVideo | DBVideoDetails {
return { return {
id: v.id, id: v.id,
@ -336,16 +268,6 @@ function formatVideoForDB (v: IndexableVideo | IndexableVideoDetails): DBVideo |
} }
} }
function formatAvatarForDB (obj: { avatar?: Avatar }) {
if (!obj.avatar) return null
return {
path: obj.avatar.path,
createdAt: obj.avatar.createdAt,
updatedAt: obj.avatar.updatedAt
}
}
function formatVideoForAPI (v: DBVideo, fromHost?: string): Video { function formatVideoForAPI (v: DBVideo, fromHost?: string): Video {
return { return {
id: v.id, id: v.id,
@ -409,16 +331,6 @@ function formatVideoForAPI (v: DBVideo, fromHost?: string): Video {
} }
} }
function formatAvatarForAPI (obj: { avatar?: Avatar }) {
if (!obj.avatar) return null
return {
path: obj.avatar.path,
createdAt: obj.avatar.createdAt,
updatedAt: obj.avatar.updatedAt
}
}
function buildChannelOrAccountMapping () { function buildChannelOrAccountMapping () {
return { return {
id: { id: {
@ -444,19 +356,7 @@ function buildChannelOrAccountMapping () {
}, },
avatar: { avatar: {
properties: { properties: buildAvatarMapping()
path: {
type: 'keyword'
},
createdAt: {
type: 'date',
format: 'date_optional_time'
},
updatedAt: {
type: 'date',
format: 'date_optional_time'
}
}
} }
} }
} }

View File

@ -0,0 +1,60 @@
import { IndexableVideo } from '../types/video.model'
import { doRequest } from '../helpers/requests'
import { ResultList, Video, VideoChannel, VideoDetails } from '@shared/models'
import { IndexableChannel } from '../types/channel.model'
import { INDEXER_COUNT } from '../initializers/constants'
import { IndexableDoc } from '../types/elastic-search.model'
async function 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 prepareVideoForDB(res.body, host)
}
async function getChannel (host: string, name: string): Promise<IndexableChannel> {
const url = 'https://' + host + '/api/v1/video-channels/' + name
const res = await doRequest<VideoChannel>({
uri: url,
json: true
})
return prepareChannelForDB(res.body, host)
}
async function 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 => prepareVideoForDB(v, host))
}
function prepareVideoForDB <T extends Video> (video: T, host: string): T & IndexableDoc {
return Object.assign(video, { elasticSearchId: host + video.id, host })
}
function prepareChannelForDB <T extends VideoChannel> (channel: T, host: string): T & IndexableDoc {
return Object.assign(channel, { elasticSearchId: host + channel.id, host })
}
export {
getVideo,
getChannel,
getVideos,
prepareChannelForDB
}

View File

@ -1,17 +1,16 @@
import { AbstractScheduler } from './abstract-scheduler' import { AbstractScheduler } from './abstract-scheduler'
import { CONFIG, INDEXER_COUNT, SCHEDULER_INTERVALS_MS } from '../../initializers/constants' import { INDEXER_COUNT, SCHEDULER_INTERVALS_MS } from '../../initializers/constants'
import { doRequest } from '../../helpers/requests'
import { logger } from '../../helpers/logger' import { logger } from '../../helpers/logger'
import { ResultList } from '../../../PeerTube/shared/models/result-list.model' import { indexVideos, refreshVideosIndex, removeVideosFromHosts } from '../elastic-search-videos'
import { Video, VideoDetails } from '../../../PeerTube/shared/models/videos/video.model' import { IndexableVideo } from '../../types/video.model'
import { indexVideos, listIndexInstances, refreshVideosIndex } from '../elastic-search-videos'
import { IndexableVideo, IndexableDoc } from '../../types/video.model'
import { inspect } from 'util' import { inspect } from 'util'
import { getRemovedHosts, listIndexInstancesHost } from '../instances-index'
import { elasticSearch } from '../../helpers/elastic-search'
import { AsyncQueue, queue } from 'async' import { AsyncQueue, queue } from 'async'
import { buildInstanceHosts } from '../elastic-search-instances'
import { getChannel, getVideo, getVideos } from '../peertube-instance'
import { indexChannels, refreshChannelsIndex } from '../elastic-search-channels'
type GetVideoQueueParam = { host: string, uuid: string } type GetVideoQueueParam = { host: string, uuid: string }
type GetChannelQueueParam = { host: string, name: string }
export class VideosIndexer extends AbstractScheduler { export class VideosIndexer extends AbstractScheduler {
@ -19,12 +18,13 @@ export class VideosIndexer extends AbstractScheduler {
protected schedulerIntervalMs = SCHEDULER_INTERVALS_MS.videosIndexer protected schedulerIntervalMs = SCHEDULER_INTERVALS_MS.videosIndexer
private readonly getVideoQueue: AsyncQueue<GetVideoQueueParam> private readonly indexVideoQueue: AsyncQueue<GetVideoQueueParam>
private readonly indexChannelQueue: AsyncQueue<GetChannelQueueParam>
private constructor () { private constructor () {
super() super()
this.getVideoQueue = queue<GetVideoQueueParam, Error>((task, cb) => { this.indexVideoQueue = queue<GetVideoQueueParam, Error>((task, cb) => {
this.indexSpecificVideo(task.host, task.uuid) this.indexSpecificVideo(task.host, task.uuid)
.then(() => cb()) .then(() => cb())
.catch(err => { .catch(err => {
@ -32,10 +32,28 @@ export class VideosIndexer extends AbstractScheduler {
cb() cb()
}) })
}) })
this.indexChannelQueue = queue<GetChannelQueueParam, Error>((task, cb) => {
this.indexSpecificChannel(task.host, task.name)
.then(() => cb())
.catch(err => {
logger.error('Error in index specific channel.', { err: inspect(err) })
cb()
})
})
this.indexChannelQueue.drain(async () => {
logger.info('Refresh channels index.')
await refreshChannelsIndex()
})
} }
scheduleVideoIndexation (host: string, uuid: string) { scheduleVideoIndexation (host: string, uuid: string) {
this.getVideoQueue.push({ uuid, host }) this.indexVideoQueue.push({ uuid, host })
}
scheduleChannelIndexation (host: string, name: string) {
this.indexChannelQueue.push({ name, host })
} }
protected async internalExecute () { protected async internalExecute () {
@ -43,11 +61,10 @@ export class VideosIndexer extends AbstractScheduler {
} }
private async runVideosIndexer () { private async runVideosIndexer () {
const dbHosts = await listIndexInstances() const { indexHosts, removedHosts } = await buildInstanceHosts()
const indexHosts = (await listIndexInstancesHost()).filter(h => h === 'peertube.cpy.re') const channelsToSync = new Set<string>()
const hostsToRemove = getRemovedHosts(dbHosts, indexHosts) await removeVideosFromHosts(removedHosts)
await this.removeVideosFromHosts(hostsToRemove)
for (const host of indexHosts) { for (const host of indexHosts) {
try { try {
@ -55,7 +72,7 @@ export class VideosIndexer extends AbstractScheduler {
let start = 0 let start = 0
do { do {
videos = await this.getVideos(host, start) videos = await getVideos(host, start)
start += videos.length start += videos.length
logger.debug('Getting %d results from %s (from = %d).', videos.length, host, start) logger.debug('Getting %d results from %s (from = %d).', videos.length, host, start)
@ -65,12 +82,16 @@ export class VideosIndexer extends AbstractScheduler {
// Fetch complete video foreach created video (to get tags) // Fetch complete video foreach created video (to get tags)
for (const c of created) { for (const c of created) {
this.scheduleVideoIndexation(host, c) this.scheduleVideoIndexation(host, c.uuid)
} }
videos.forEach(v => channelsToSync.add(v.channel.name))
} }
} while (videos.length === INDEXER_COUNT.VIDEOS && start < 500) } while (videos.length === INDEXER_COUNT.VIDEOS && start < 500)
logger.info('Added video data from %s.', host) logger.info('Added video data from %s.', host)
channelsToSync.forEach(c => this.scheduleChannelIndexation(host, c))
} catch (err) { } catch (err) {
console.error(inspect(err, { depth: 10 })) console.error(inspect(err, { depth: 10 }))
logger.warn('Cannot index videos from %s.', host, { err }) logger.warn('Cannot index videos from %s.', host, { err })
@ -80,65 +101,20 @@ export class VideosIndexer extends AbstractScheduler {
await refreshVideosIndex() 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) { private async indexSpecificVideo (host: string, uuid: string) {
const video = await this.getVideo(host, uuid) const video = await getVideo(host, uuid)
logger.info('Indexing specific video %s of %s.', uuid, host) logger.info('Indexing specific video %s of %s.', uuid, host)
await indexVideos([ video ], true) await indexVideos([ video ], true)
} }
private prepareVideoForDB <T extends Video> (video: T, host: string): T & IndexableDoc { private async indexSpecificChannel (host: string, name: string) {
return Object.assign(video, { elasticSearchId: host + video.id, host }) const channel = await getChannel(host, name)
logger.info('Indexing specific channel %s@%s.', name, host)
await indexChannels([ channel ], true)
} }
static get Instance () { static get Instance () {

View File

@ -60,9 +60,22 @@ const videosSearchValidator = [
} }
] ]
const videoChannelsSearchValidator = [
query('search').not().isEmpty().withMessage('Should have a valid search'),
(req: express.Request, res: express.Response, next: express.NextFunction) => {
logger.debug('Checking video channels search query', { parameters: req.query })
if (areValidationErrors(req, res)) return
return next()
}
]
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
export { export {
videoChannelsSearchValidator,
commonVideosFiltersValidator, commonVideosFiltersValidator,
videosSearchValidator videosSearchValidator
} }

View File

@ -2,11 +2,14 @@ import { SORTABLE_COLUMNS } from '../../initializers/constants'
import { checkSort, createSortableColumns } from './utils' import { checkSort, createSortableColumns } from './utils'
const SORTABLE_VIDEOS_SEARCH_COLUMNS = createSortableColumns(SORTABLE_COLUMNS.VIDEOS_SEARCH) const SORTABLE_VIDEOS_SEARCH_COLUMNS = createSortableColumns(SORTABLE_COLUMNS.VIDEOS_SEARCH)
const SORTABLE_CHANNELS_SEARCH_COLUMNS = createSortableColumns(SORTABLE_COLUMNS.CHANNELS_SEARCH)
const videosSearchSortValidator = checkSort(SORTABLE_VIDEOS_SEARCH_COLUMNS) const videosSearchSortValidator = checkSort(SORTABLE_VIDEOS_SEARCH_COLUMNS)
const channelsSearchSortValidator = checkSort(SORTABLE_VIDEOS_SEARCH_COLUMNS)
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
export { export {
videosSearchSortValidator videosSearchSortValidator,
channelsSearchSortValidator
} }

View File

@ -0,0 +1,5 @@
import {
VideoChannelsSearchQuery as PeerTubeChannelsSearchQuery
} from '../../PeerTube/shared/models/search/video-channels-search-query.model'
export type ChannelsSearchQuery = PeerTubeChannelsSearchQuery

View File

@ -0,0 +1,16 @@
import { IndexableDoc } from './elastic-search.model'
import { VideoChannel, VideoChannelSummary } from '@shared/models'
export interface IndexableChannelSummary extends VideoChannelSummary, IndexableDoc {
}
export interface IndexableChannel extends VideoChannel, IndexableDoc {
}
export interface DBChannel extends Omit<VideoChannel, 'isLocal'> {
indexedAt: Date
}
export interface DBChannelSummary extends VideoChannelSummary {
indexedAt: Date
}

View File

@ -0,0 +1,4 @@
export interface IndexableDoc {
elasticSearchId: string
host: string
}

View File

@ -1,9 +1,5 @@
import { Video, VideoDetails } from '@shared/models/videos/video.model' import { Video, VideoDetails } from '@shared/models/videos/video.model'
import { IndexableDoc } from './elastic-search.model'
export interface IndexableDoc {
elasticSearchId: string
host: string
}
export interface IndexableVideo extends Video, IndexableDoc { export interface IndexableVideo extends Video, IndexableDoc {
} }