Verified Commit 22df69fd authored by Chocobozzz's avatar Chocobozzz
Browse files

Add job queue hooks

parent 7a9e420a
......@@ -112,8 +112,14 @@ class PluginsManager {
for (const hook of this.hooks[hookName]) {
logger.info(`Running hook ${hookName} of plugin ${hook.plugin.name}`)
result = await internalRunHook(hook.handler, hookType, result, params, err => {
logger.error(`Cannot run hook ${hookName} of script ${hook.clientScript.script} of plugin ${hook.plugin.name}`, err)
result = await internalRunHook({
handler: hook.handler,
hookType,
result,
params,
onError: err => {
logger.error(`Cannot run hook ${hookName} of script ${hook.clientScript.script} of plugin ${hook.plugin.name}`, err)
}
})
}
......
......@@ -24,6 +24,7 @@ import {
} from '../../../shared/models'
import { logger } from '../../helpers/logger'
import { JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_TTL, REPEAT_JOBS, WEBSERVER } from '../../initializers/constants'
import { Hooks } from '../plugins/hooks'
import { processActivityPubCleaner } from './handlers/activitypub-cleaner'
import { processActivityPubFollow } from './handlers/activitypub-follow'
import { processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast'
......@@ -157,8 +158,11 @@ class JobQueue {
const handler = handlers[handlerName]
queue.process(this.getJobConcurrency(handlerName), handler)
.catch(err => logger.error('Error in job queue processor %s.', handlerName, { err }))
queue.process(this.getJobConcurrency(handlerName), async (jobArg: Job<any>) => {
const job = await Hooks.wrapObject(jobArg, 'filter:job-queue.process.params', { type: handlerName })
return Hooks.wrapPromiseFun(handler, job, 'filter:job-queue.process.result')
}).catch(err => logger.error('Error in job queue processor %s.', handlerName, { err }))
queue.on('failed', (job, err) => {
const logLevel = silentFailure.has(handlerName)
......
......@@ -8,8 +8,8 @@ type RawFunction <U, T> = (params: U) => T
// Helpers to run hooks
const Hooks = {
wrapObject: <T, U extends ServerFilterHookName>(result: T, hookName: U) => {
return PluginManager.Instance.runHook(hookName, result)
wrapObject: <T, U extends ServerFilterHookName>(result: T, hookName: U, context?: any) => {
return PluginManager.Instance.runHook(hookName, result, context)
},
wrapPromiseFun: async <U, T, V extends ServerFilterHookName>(fun: PromiseFunction<U, T>, params: U, hookName: V) => {
......
......@@ -220,6 +220,10 @@ function buildPluginRelatedHelpers (plugin: MPlugin, npmName: string) {
function buildUserHelpers () {
return {
loadById: (id: number) => {
return UserModel.loadByIdFull(id)
},
getAuthUser: (res: express.Response) => {
const user = res.locals.oauth?.token?.User
if (!user) return undefined
......
......@@ -215,8 +215,12 @@ export class PluginManager implements ServerHook {
for (const hook of this.hooks[hookName]) {
logger.debug('Running hook %s of plugin %s.', hookName, hook.npmName)
result = await internalRunHook(hook.handler, hookType, result, params, err => {
logger.error('Cannot run hook %s of plugin %s.', hookName, hook.pluginName, { err })
result = await internalRunHook({
handler: hook.handler,
hookType,
result,
params,
onError: err => { logger.error('Cannot run hook %s of plugin %s.', hookName, hook.pluginName, { err }) }
})
}
......
......@@ -88,6 +88,15 @@ async function register ({
return res.json({ routerRoute })
})
router.get('/user/:id', async (req, res) => {
const user = await peertubeHelpers.user.loadById(req.params.id)
if (!user) return res.status(404).end()
return res.json({
username: user.username
})
})
router.get('/user', async (req, res) => {
const user = await peertubeHelpers.user.getAuthUser(res)
if (!user) return res.sendStatus(404)
......@@ -97,6 +106,7 @@ async function register ({
const isUser = user.role === 2
return res.json({
id: user.id,
username: user.username,
displayName: user.Account.name,
isAdmin,
......
......@@ -253,6 +253,27 @@ async function register ({ registerHook, registerSetting, settingsManager, stora
}
})
registerHook({
target: 'filter:job-queue.process.params',
handler: (object, context) => {
peertubeHelpers.logger.debug('TOTO.', { object, context })
if (context.type !== 'video-studio-edition') return object
object.data.tasks = [
{
name: 'cut',
options: {
start: 0,
end: 1
}
}
]
return object
}
})
// Upload/import/live attributes
for (const target of [
'filter:api.video.upload.video-attribute.result',
......@@ -284,7 +305,10 @@ async function register ({ registerHook, registerSetting, settingsManager, stora
'filter:api.search.video-playlists.index.list.result',
'filter:api.overviews.videos.list.params',
'filter:api.overviews.videos.list.result'
'filter:api.overviews.videos.list.result',
'filter:job-queue.process.params',
'filter:job-queue.process.result'
]
for (const h of filterHooks) {
......
......@@ -632,6 +632,51 @@ describe('Test plugin filter hooks', function () {
})
describe('Job queue filters', function () {
let videoUUID: string
before(async function () {
this.timeout(120_000)
const { uuid } = await servers[0].videos.quickUpload({ name: 'studio' })
const video = await servers[0].videos.get({ id: uuid })
expect(video.duration).at.least(2)
videoUUID = video.uuid
await waitJobs(servers)
await servers[0].config.enableStudio()
})
it('Should run filter:job-queue.process.params', async function () {
this.timeout(120_000)
await servers[0].videoStudio.createEditionTasks({
videoId: videoUUID,
tasks: [
{
name: 'add-intro',
options: {
file: 'video_very_short_240p.mp4'
}
}
]
})
await waitJobs(servers)
await servers[0].servers.waitUntilLog('Run hook filter:job-queue.process.params', 1, false)
const video = await servers[0].videos.get({ id: videoUUID })
expect(video.duration).at.most(2)
})
it('Should run filter:job-queue.process.result', async function () {
await servers[0].servers.waitUntilLog('Run hook filter:job-queue.process.result', 1, false)
})
})
after(async function () {
await cleanupTests(servers)
})
......
......@@ -110,6 +110,7 @@ describe('Test plugin helpers', function () {
})
describe('User', function () {
let rootId: number
it('Should not get a user if not authenticated', async function () {
await makeGetRequest({
......@@ -132,6 +133,28 @@ describe('Test plugin helpers', function () {
expect(res.body.isAdmin).to.be.true
expect(res.body.isModerator).to.be.false
expect(res.body.isUser).to.be.false
rootId = res.body.id
})
it('Should load a user by id', async function () {
{
const res = await makeGetRequest({
url: servers[0].url,
path: '/plugins/test-four/router/user/' + rootId,
expectedStatus: HttpStatusCode.OK_200
})
expect(res.body.username).to.equal('root')
}
{
await makeGetRequest({
url: servers[0].url,
path: '/plugins/test-four/router/user/42',
expectedStatus: HttpStatusCode.NOT_FOUND_404
})
}
})
})
......
......@@ -14,10 +14,9 @@ import {
RegisterServerSettingOptions,
ServerConfig,
ThumbnailType,
UserRole,
VideoBlacklistCreate
} from '@shared/models'
import { MVideoThumbnail } from '../models'
import { MUserDefault, MVideoThumbnail } from '../models'
import {
RegisterServerAuthExternalOptions,
RegisterServerAuthExternalResult,
......@@ -100,16 +99,10 @@ export type PeerTubeHelpers = {
user: {
// PeerTube >= 3.2
getAuthUser: (response: Response) => Promise<{
id?: string
username: string
email: string
blocked: boolean
role: UserRole
Account: {
name: string
}
} | undefined>
getAuthUser: (response: Response) => Promise<MUserDefault>
// PeerTube >= 4.3
loadById: (id: number) => Promise<MUserDefault>
}
}
......
......@@ -8,15 +8,24 @@ function getHookType (hookName: string) {
return HookType.STATIC
}
async function internalRunHook <T> (handler: Function, hookType: HookType, result: T, params: any, onError: (err: Error) => void) {
async function internalRunHook <T> (options: {
handler: Function
hookType: HookType
result: T
params: any
onError: (err: Error) => void
}) {
const { handler, hookType, result, params, onError } = options
try {
if (hookType === HookType.FILTER) {
const p = handler(result, params)
if (isPromise(p)) result = await p
else result = p
const newResult = isPromise(p)
? await p
: p
return result
return newResult
}
// Action/static hooks do not have result value
......
......@@ -90,7 +90,10 @@ export const serverFilterHookObject = {
// Filter result to check if the embed is allowed for a particular request
'filter:html.embed.video.allowed.result': true,
'filter:html.embed.video-playlist.allowed.result': true
'filter:html.embed.video-playlist.allowed.result': true,
'filter:job-queue.process.params': true,
'filter:job-queue.process.result': true
}
export type ServerFilterHookName = keyof typeof serverFilterHookObject
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment