forks/peertube
1
0
Fork 0
Code Releases Activity

Refractor retry transaction function

Browse source
This commit is contained in:
Chocobozzz 2018-06-13 14:27:40 +02:00
parent 3cd0734fd9
commit 90d4bb8125
No known key found for this signature in database
GPG key ID: 583A612D890159BE
19 changed files with 162 additions and 415 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
server/controllers/api/users.ts
View file

@ -4,7 +4,6 @@ import { extname, join } from 'path'
import * as uuidv4 from 'uuid/v4'
import * as RateLimit from 'express-rate-limit'
import { UserCreate, UserRight, UserRole, UserUpdate, UserUpdateMe, UserVideoRate as FormattedUserVideoRate } from '../../../shared'
import { retryTransactionWrapper } from '../../helpers/database-utils'
import { processImage } from '../../helpers/image-utils'
import { logger } from '../../helpers/logger'
import { getFormattedObjects } from '../../helpers/utils'
@ -16,6 +15,7 @@ import { Redis } from '../../lib/redis'
import { createUserAccountAndChannel } from '../../lib/user'
import {
asyncMiddleware,
asyncRetryTransactionMiddleware,
authenticate,
ensureUserHasRight,
ensureUserRegistrationAllowed,
@ -102,14 +102,14 @@ usersRouter.post('/',
authenticate,
ensureUserHasRight(UserRight.MANAGE_USERS),
asyncMiddleware(usersAddValidator),
asyncMiddleware(createUserRetryWrapper)
asyncRetryTransactionMiddleware(createUser)
)
usersRouter.post('/register',
asyncMiddleware(ensureUserRegistrationAllowed),
ensureUserRegistrationAllowedForIP,
asyncMiddleware(usersRegisterValidator),
asyncMiddleware(registerUserRetryWrapper)
asyncRetryTransactionMiddleware(registerUser)
)
usersRouter.put('/me',
@ -178,26 +178,7 @@ async function getUserVideos (req: express.Request, res: express.Response, next:
return res.json(getFormattedObjects(resultList.data, resultList.total, { additionalAttributes }))
}
async function createUserRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req ],
errorMessage: 'Cannot insert the user with many retries.'
}
const { user, account } = await retryTransactionWrapper(createUser, options)
return res.json({
user: {
id: user.id,
account: {
id: account.id,
uuid: account.Actor.uuid
}
}
}).end()
}
async function createUser (req: express.Request) {
async function createUser (req: express.Request, res: express.Response) {
const body: UserCreate = req.body
const userToCreate = new UserModel({
username: body.username,
@ -213,21 +194,18 @@ async function createUser (req: express.Request) {
logger.info('User %s with its channel and account created.', body.username)
return { user, account }
}
async function registerUserRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req ],
errorMessage: 'Cannot insert the user with many retries.'
return res.json({
user: {
id: user.id,
account: {
id: account.id,
uuid: account.Actor.uuid
}
await retryTransactionWrapper(registerUser, options)
return res.type('json').status(204).end()
}
}).end()
}
async function registerUser (req: express.Request) {
async function registerUser (req: express.Request, res: express.Response) {
const body: UserCreate = req.body
const user = new UserModel({
@ -243,6 +221,8 @@ async function registerUser (req: express.Request) {
await createUserAccountAndChannel(user)
logger.info('User %s with its channel and account registered.', body.username)
return res.type('json').status(204).end()
}
async function getUserInformation (req: express.Request, res: express.Response, next: express.NextFunction) {

53
server/controllers/api/video-channel.ts
View file

@ -2,6 +2,7 @@ import * as express from 'express'
import { getFormattedObjects, resetSequelizeInstance } from '../../helpers/utils'
import {
asyncMiddleware,
asyncRetryTransactionMiddleware,
authenticate,
optionalAuthenticate,
paginationValidator,
@ -20,7 +21,6 @@ import { VideoChannelCreate, VideoChannelUpdate } from '../../../shared'
import { createVideoChannel } from '../../lib/video-channel'
import { isNSFWHidden } from '../../helpers/express-utils'
import { setAsyncActorKeys } from '../../lib/activitypub'
import { retryTransactionWrapper } from '../../helpers/database-utils'
import { AccountModel } from '../../models/account/account'
import { sequelizeTypescript } from '../../initializers'
import { logger } from '../../helpers/logger'
@ -39,19 +39,19 @@ videoChannelRouter.get('/',
videoChannelRouter.post('/',
authenticate,
videoChannelsAddValidator,
asyncMiddleware(addVideoChannelRetryWrapper)
asyncRetryTransactionMiddleware(addVideoChannel)
)
videoChannelRouter.put('/:id',
authenticate,
asyncMiddleware(videoChannelsUpdateValidator),
updateVideoChannelRetryWrapper
asyncRetryTransactionMiddleware(updateVideoChannel)
)
videoChannelRouter.delete('/:id',
authenticate,
asyncMiddleware(videoChannelsRemoveValidator),
asyncMiddleware(removeVideoChannelRetryWrapper)
asyncRetryTransactionMiddleware(removeVideoChannel)
)
videoChannelRouter.get('/:id',
@ -83,23 +83,6 @@ async function listVideoChannels (req: express.Request, res: express.Response, n
return res.json(getFormattedObjects(resultList.data, resultList.total))
}
// Wrapper to video channel add that retry the async function if there is a database error
// We need this because we run the transaction in SERIALIZABLE isolation that can fail
async function addVideoChannelRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res ],
errorMessage: 'Cannot insert the video video channel with many retries.'
}
const videoChannel = await retryTransactionWrapper(addVideoChannel, options)
return res.json({
videoChannel: {
id: videoChannel.id,
uuid: videoChannel.Actor.uuid
}
}).end()
}
async function addVideoChannel (req: express.Request, res: express.Response) {
const videoChannelInfo: VideoChannelCreate = req.body
const account: AccountModel = res.locals.oauth.token.User.Account
@ -113,18 +96,12 @@ async function addVideoChannel (req: express.Request, res: express.Response) {
logger.info('Video channel with uuid %s created.', videoChannelCreated.Actor.uuid)
return videoChannelCreated
}
async function updateVideoChannelRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res ],
errorMessage: 'Cannot update the video with many retries.'
return res.json({
videoChannel: {
id: videoChannelCreated.id,
uuid: videoChannelCreated.Actor.uuid
}
await retryTransactionWrapper(updateVideoChannel, options)
return res.type('json').status(204).end()
}).end()
}
async function updateVideoChannel (req: express.Request, res: express.Response) {
@ -157,15 +134,6 @@ async function updateVideoChannel (req: express.Request, res: express.Response)
throw err
}
}
async function removeVideoChannelRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res ],
errorMessage: 'Cannot remove the video channel with many retries.'
}
await retryTransactionWrapper(removeVideoChannel, options)
return res.type('json').status(204).end()
}
@ -173,12 +141,13 @@ async function removeVideoChannelRetryWrapper (req: express.Request, res: expres
async function removeVideoChannel (req: express.Request, res: express.Response) {
const videoChannelInstance: VideoChannelModel = res.locals.videoChannel
return sequelizeTypescript.transaction(async t => {
await sequelizeTypescript.transaction(async t => {
await videoChannelInstance.destroy({ transaction: t })
logger.info('Video channel with name %s and uuid %s deleted.', videoChannelInstance.name, videoChannelInstance.Actor.uuid)
})
return res.type('json').status(204).end()
}
async function getVideoChannel (req: express.Request, res: express.Response, next: express.NextFunction) {

25
server/controllers/api/videos/abuse.ts
View file

@ -1,12 +1,18 @@
import * as express from 'express'
import { UserRight, VideoAbuseCreate } from '../../../../shared'
import { retryTransactionWrapper } from '../../../helpers/database-utils'
import { logger } from '../../../helpers/logger'
import { getFormattedObjects } from '../../../helpers/utils'
import { sequelizeTypescript } from '../../../initializers'
import { sendVideoAbuse } from '../../../lib/activitypub/send'
import {
asyncMiddleware, authenticate, ensureUserHasRight, paginationValidator, setDefaultSort, setDefaultPagination, videoAbuseReportValidator,
asyncMiddleware,
asyncRetryTransactionMiddleware,
authenticate,
ensureUserHasRight,
paginationValidator,
setDefaultPagination,
setDefaultSort,
videoAbuseReportValidator,
videoAbusesSortValidator
} from '../../../middlewares'
import { AccountModel } from '../../../models/account/account'
@ -27,7 +33,7 @@ abuseVideoRouter.get('/abuse',
abuseVideoRouter.post('/:id/abuse',
authenticate,
asyncMiddleware(videoAbuseReportValidator),
asyncMiddleware(reportVideoAbuseRetryWrapper)
asyncRetryTransactionMiddleware(reportVideoAbuse)
)
// ---------------------------------------------------------------------------
@ -44,17 +50,6 @@ async function listVideoAbuses (req: express.Request, res: express.Response, nex
return res.json(getFormattedObjects(resultList.data, resultList.total))
}
async function reportVideoAbuseRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res ],
errorMessage: 'Cannot report abuse to the video with many retries.'
}
await retryTransactionWrapper(reportVideoAbuse, options)
return res.type('json').status(204).end()
}
async function reportVideoAbuse (req: express.Request, res: express.Response) {
const videoInstance = res.locals.video as VideoModel
const reporterAccount = res.locals.oauth.token.User.Account as AccountModel
@ -77,4 +72,6 @@ async function reportVideoAbuse (req: express.Request, res: express.Response) {
})
logger.info('Abuse report for video %s created.', videoInstance.name)
return res.type('json').status(204).end()
}

68
server/controllers/api/videos/comment.ts
View file

@ -1,15 +1,24 @@
import * as express from 'express'
import { ResultList } from '../../../../shared/models'
import { VideoCommentCreate } from '../../../../shared/models/videos/video-comment.model'
import { retryTransactionWrapper } from '../../../helpers/database-utils'
import { logger } from '../../../helpers/logger'
import { getFormattedObjects } from '../../../helpers/utils'
import { sequelizeTypescript } from '../../../initializers'
import { buildFormattedCommentTree, createVideoComment } from '../../../lib/video-comment'
import { asyncMiddleware, authenticate, paginationValidator, setDefaultSort, setDefaultPagination } from '../../../middlewares'
import {
asyncMiddleware,
asyncRetryTransactionMiddleware,
authenticate,
paginationValidator,
setDefaultPagination,
setDefaultSort
} from '../../../middlewares'
import { videoCommentThreadsSortValidator } from '../../../middlewares/validators'
import {
addVideoCommentReplyValidator, addVideoCommentThreadValidator, listVideoCommentThreadsValidator, listVideoThreadCommentsValidator,
addVideoCommentReplyValidator,
addVideoCommentThreadValidator,
listVideoCommentThreadsValidator,
listVideoThreadCommentsValidator,
removeVideoCommentValidator
} from '../../../middlewares/validators/video-comments'
import { VideoModel } from '../../../models/video/video'
@ -33,17 +42,17 @@ videoCommentRouter.get('/:videoId/comment-threads/:threadId',
videoCommentRouter.post('/:videoId/comment-threads',
authenticate,
asyncMiddleware(addVideoCommentThreadValidator),
asyncMiddleware(addVideoCommentThreadRetryWrapper)
asyncRetryTransactionMiddleware(addVideoCommentThread)
)
videoCommentRouter.post('/:videoId/comments/:commentId',
authenticate,
asyncMiddleware(addVideoCommentReplyValidator),
asyncMiddleware(addVideoCommentReplyRetryWrapper)
asyncRetryTransactionMiddleware(addVideoCommentReply)
)
videoCommentRouter.delete('/:videoId/comments/:commentId',
authenticate,
asyncMiddleware(removeVideoCommentValidator),
asyncMiddleware(removeVideoCommentRetryWrapper)
asyncRetryTransactionMiddleware(removeVideoComment)
)
// ---------------------------------------------------------------------------
@ -86,23 +95,10 @@ async function listVideoThreadComments (req: express.Request, res: express.Respo
return res.json(buildFormattedCommentTree(resultList))
}
async function addVideoCommentThreadRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res ],
errorMessage: 'Cannot insert the video comment thread with many retries.'
}
const comment = await retryTransactionWrapper(addVideoCommentThread, options)
res.json({
comment: comment.toFormattedJSON()
}).end()
}
function addVideoCommentThread (req: express.Request, res: express.Response) {
async function addVideoCommentThread (req: express.Request, res: express.Response) {
const videoCommentInfo: VideoCommentCreate = req.body
return sequelizeTypescript.transaction(async t => {
const comment = await sequelizeTypescript.transaction(async t => {
return createVideoComment({
text: videoCommentInfo.text,
inReplyToComment: null,
@ -110,25 +106,16 @@ function addVideoCommentThread (req: express.Request, res: express.Response) {
account: res.locals.oauth.token.User.Account
}, t)
})
}
async function addVideoCommentReplyRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res ],
errorMessage: 'Cannot insert the video comment reply with many retries.'
}
const comment = await retryTransactionWrapper(addVideoCommentReply, options)
res.json({
return res.json({
comment: comment.toFormattedJSON()
}).end()
}
function addVideoCommentReply (req: express.Request, res: express.Response, next: express.NextFunction) {
async function addVideoCommentReply (req: express.Request, res: express.Response) {
const videoCommentInfo: VideoCommentCreate = req.body
return sequelizeTypescript.transaction(async t => {
const comment = await sequelizeTypescript.transaction(async t => {
return createVideoComment({
text: videoCommentInfo.text,
inReplyToComment: res.locals.videoComment,
@ -136,17 +123,10 @@ function addVideoCommentReply (req: express.Request, res: express.Response, next
account: res.locals.oauth.token.User.Account
}, t)
})
}
async function removeVideoCommentRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res ],
errorMessage: 'Cannot remove the video comment with many retries.'
}
await retryTransactionWrapper(removeVideoComment, options)
return res.type('json').status(204).end()
return res.json({
comment: comment.toFormattedJSON()
}).end()
}
async function removeVideoComment (req: express.Request, res: express.Response) {
@ -157,4 +137,6 @@ async function removeVideoComment (req: express.Request, res: express.Response)
})
logger.info('Video comment %d deleted.', videoCommentInstance.id)
return res.type('json').status(204).end()
}

60
server/controllers/api/videos/index.ts
View file

@ -2,7 +2,6 @@ import * as express from 'express'
import { extname, join } from 'path'
import { VideoCreate, VideoPrivacy, VideoState, VideoUpdate } from '../../../../shared'
import { renamePromise } from '../../../helpers/core-utils'
import { retryTransactionWrapper } from '../../../helpers/database-utils'
import { getVideoFileResolution } from '../../../helpers/ffmpeg-utils'
import { processImage } from '../../../helpers/image-utils'
import { logger } from '../../../helpers/logger'
@ -30,6 +29,7 @@ import { JobQueue } from '../../../lib/job-queue'
import { Redis } from '../../../lib/redis'
import {
asyncMiddleware,
asyncRetryTransactionMiddleware,
authenticate,
optionalAuthenticate,
paginationValidator,
@ -104,13 +104,13 @@ videosRouter.put('/:id',
authenticate,
reqVideoFileUpdate,
asyncMiddleware(videosUpdateValidator),
asyncMiddleware(updateVideoRetryWrapper)
asyncRetryTransactionMiddleware(updateVideo)
)
videosRouter.post('/upload',
authenticate,
reqVideoFileAdd,
asyncMiddleware(videosAddValidator),
asyncMiddleware(addVideoRetryWrapper)
asyncRetryTransactionMiddleware(addVideo)
)
videosRouter.get('/:id/description',
@ -129,7 +129,7 @@ videosRouter.post('/:id/views',
videosRouter.delete('/:id',
authenticate,
asyncMiddleware(videosRemoveValidator),
asyncMiddleware(removeVideoRetryWrapper)
asyncRetryTransactionMiddleware(removeVideo)
)
// ---------------------------------------------------------------------------
@ -156,25 +156,8 @@ function listVideoPrivacies (req: express.Request, res: express.Response) {
res.json(VIDEO_PRIVACIES)
}
// Wrapper to video add that retry the function if there is a database error
// We need this because we run the transaction in SERIALIZABLE isolation that can fail
async function addVideoRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res, req.files['videofile'][0] ],
errorMessage: 'Cannot insert the video with many retries.'
}
const video = await retryTransactionWrapper(addVideo, options)
res.json({
video: {
id: video.id,
uuid: video.uuid
}
}).end()
}
async function addVideo (req: express.Request, res: express.Response, videoPhysicalFile: Express.Multer.File) {
async function addVideo (req: express.Request, res: express.Response) {
const videoPhysicalFile = req.files['videofile'][0]
const videoInfo: VideoCreate = req.body
// Prepare data so we don't block the transaction
@ -272,18 +255,12 @@ async function addVideo (req: express.Request, res: express.Response, videoPhysi
await JobQueue.Instance.createJob({ type: 'video-file', payload: dataInput })
}
return videoCreated
}
async function updateVideoRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res ],
errorMessage: 'Cannot update the video with many retries.'
return res.json({
video: {
id: videoCreated.id,
uuid: videoCreated.uuid
}
await retryTransactionWrapper(updateVideo, options)
return res.type('json').status(204).end()
}).end()
}
async function updateVideo (req: express.Request, res: express.Response) {
@ -360,6 +337,8 @@ async function updateVideo (req: express.Request, res: express.Response) {
throw err
}
return res.type('json').status(204).end()
}
function getVideo (req: express.Request, res: express.Response) {
@ -414,17 +393,6 @@ async function listVideos (req: express.Request, res: express.Response, next: ex
return res.json(getFormattedObjects(resultList.data, resultList.total))
}
async function removeVideoRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res ],
errorMessage: 'Cannot remove the video with many retries.'
}
await retryTransactionWrapper(removeVideo, options)
return res.type('json').status(204).end()
}
async function removeVideo (req: express.Request, res: express.Response) {
const videoInstance: VideoModel = res.locals.video
@ -433,6 +401,8 @@ async function removeVideo (req: express.Request, res: express.Response) {
})
logger.info('Video with name %s and uuid %s deleted.', videoInstance.name, videoInstance.uuid)
return res.type('json').status(204).end()
}
async function searchVideos (req: express.Request, res: express.Response, next: express.NextFunction) {

18
server/controllers/api/videos/rate.ts
View file

@ -1,10 +1,9 @@
import * as express from 'express'
import { UserVideoRateUpdate } from '../../../../shared'
import { retryTransactionWrapper } from '../../../helpers/database-utils'
import { logger } from '../../../helpers/logger'
import { sequelizeTypescript, VIDEO_RATE_TYPES } from '../../../initializers'
import { sendVideoRateChange } from '../../../lib/activitypub'
import { asyncMiddleware, authenticate, videoRateValidator } from '../../../middlewares'
import { asyncMiddleware, asyncRetryTransactionMiddleware, authenticate, videoRateValidator } from '../../../middlewares'
import { AccountModel } from '../../../models/account/account'
import { AccountVideoRateModel } from '../../../models/account/account-video-rate'
import { VideoModel } from '../../../models/video/video'
@ -14,7 +13,7 @@ const rateVideoRouter = express.Router()
rateVideoRouter.put('/:id/rate',
authenticate,
asyncMiddleware(videoRateValidator),
asyncMiddleware(rateVideoRetryWrapper)
asyncRetryTransactionMiddleware(rateVideo)
)
// ---------------------------------------------------------------------------
@ -25,17 +24,6 @@ export {
// ---------------------------------------------------------------------------
async function rateVideoRetryWrapper (req: express.Request, res: express.Response, next: express.NextFunction) {
const options = {
arguments: [ req, res ],
errorMessage: 'Cannot update the user video rate.'
}
await retryTransactionWrapper(rateVideo, options)
return res.type('json').status(204).end()
}
async function rateVideo (req: express.Request, res: express.Response) {
const body: UserVideoRateUpdate = req.body
const rateType = body.rating
@ -87,4 +75,6 @@ async function rateVideo (req: express.Request, res: express.Response) {
})
logger.info('Account video rate for video %s of account %s updated.', videoInstance.name, accountInstance.name)
return res.type('json').status(204).end()
}

37
server/helpers/database-utils.ts
View file

@ -3,27 +3,43 @@ import * as Bluebird from 'bluebird'
import { Model } from 'sequelize-typescript'
import { logger } from './logger'
type RetryTransactionWrapperOptions = { errorMessage: string, arguments?: any[] }
function retryTransactionWrapper <T> (
functionToRetry: (...args) => Promise<T> | Bluebird<T>,
options: RetryTransactionWrapperOptions
): Promise<T> {
const args = options.arguments ? options.arguments : []
function retryTransactionWrapper <T, A, B, C> (
functionToRetry: (arg1: A, arg2: B, arg3: C) => Promise<T> | Bluebird<T>,
arg1: A,
arg2: B,
arg3: C
): Promise<T>
function retryTransactionWrapper <T, A, B> (
functionToRetry: (arg1: A, arg2: B) => Promise<T> | Bluebird<T>,
arg1: A,
arg2: B
): Promise<T>
function retryTransactionWrapper <T, A> (
functionToRetry: (arg1: A) => Promise<T> | Bluebird<T>,
arg1: A
): Promise<T>
function retryTransactionWrapper <T> (
functionToRetry: (...args: any[]) => Promise<T> | Bluebird<T>,
...args: any[]
): Promise<T> {
return transactionRetryer<T>(callback => {
functionToRetry.apply(this, args)
.then((result: T) => callback(null, result))
.catch(err => callback(err))
})
.catch(err => {
logger.error(options.errorMessage, { err })
logger.error('Cannot execute %s with many retries.', functionToRetry.toString(), { err })
throw err
})
}
function transactionRetryer <T> (func: (err: any, data: T) => any) {
return new Promise<T>((res, rej) => {
retry({
retry(
{
times: 5,
errorFilter: err => {
@ -31,7 +47,10 @@ function transactionRetryer <T> (func: (err: any, data: T) => any) {
logger.debug('Maybe retrying the transaction function.', { willRetry, err })
return willRetry
}
}, func, (err, data) => err ? rej(err) : res(data))
},
func,
(err, data) => err ? rej(err) : res(data)
)
})
}

12
server/lib/activitypub/actor.ts
View file

@ -62,18 +62,10 @@ async function getOrCreateActorAndServerAndModel (activityActor: string | Activi
}
}
const options = {
arguments: [ result, ownerActor ],
errorMessage: 'Cannot save actor and server with many retries.'
}
actor = await retryTransactionWrapper(saveActorAndServerAndModelIfNotExist, options)
actor = await retryTransactionWrapper(saveActorAndServerAndModelIfNotExist, result, ownerActor)
}
const options = {
arguments: [ actor ],
errorMessage: 'Cannot refresh actor if needed with many retries.'
}
return retryTransactionWrapper(refreshActorIfNeeded, options)
return retryTransactionWrapper(refreshActorIfNeeded, actor)
}
function buildActorInstance (type: ActivityPubActorType, url: string, preferredUsername: string, uuid?: string) {

13
server/lib/activitypub/process/process-announce.ts
View file

@ -11,7 +11,7 @@ import { getOrCreateAccountAndVideoAndChannel } from '../videos'
async function processAnnounceActivity (activity: ActivityAnnounce) {
const actorAnnouncer = await getOrCreateActorAndServerAndModel(activity.actor)
return processVideoShare(actorAnnouncer, activity)
return retryTransactionWrapper(processVideoShare, actorAnnouncer, activity)
}
// ---------------------------------------------------------------------------
@ -22,16 +22,7 @@ export {
// ---------------------------------------------------------------------------
function processVideoShare (actorAnnouncer: ActorModel, activity: ActivityAnnounce) {
const options = {
arguments: [ actorAnnouncer, activity ],
errorMessage: 'Cannot share the video activity with many retries.'
}
return retryTransactionWrapper(shareVideo, options)
}
async function shareVideo (actorAnnouncer: ActorModel, activity: ActivityAnnounce) {
async function processVideoShare (actorAnnouncer: ActorModel, activity: ActivityAnnounce) {
const objectUri = typeof activity.object === 'string' ? activity.object : activity.object.id
let video: VideoModel

37
server/lib/activitypub/process/process-create.ts
View file

@ -21,13 +21,13 @@ async function processCreateActivity (activity: ActivityCreate) {
if (activityType === 'View') {
return processCreateView(actor, activity)
} else if (activityType === 'Dislike') {
return processCreateDislike(actor, activity)
return retryTransactionWrapper(processCreateDislike, actor, activity)
} else if (activityType === 'Video') {
return processCreateVideo(actor, activity)
} else if (activityType === 'Flag') {
return processCreateVideoAbuse(actor, activityObject as VideoAbuseObject)
return retryTransactionWrapper(processCreateVideoAbuse, actor, activityObject as VideoAbuseObject)
} else if (activityType === 'Note') {
return processCreateVideoComment(actor, activity)
return retryTransactionWrapper(processCreateVideoComment, actor, activity)
}
logger.warn('Unknown activity object type %s when creating activity.', activityType, { activity: activity.id })
@ -54,15 +54,6 @@ async function processCreateVideo (
}
async function processCreateDislike (byActor: ActorModel, activity: ActivityCreate) {
const options = {
arguments: [ byActor, activity ],
errorMessage: 'Cannot dislike the video with many retries.'
}
return retryTransactionWrapper(createVideoDislike, options)
}
async function createVideoDislike (byActor: ActorModel, activity: ActivityCreate) {
const dislike = activity.object as DislikeObject
const byAccount = byActor.Account
@ -109,16 +100,7 @@ async function processCreateView (byActor: ActorModel, activity: ActivityCreate)
}
}
function processCreateVideoAbuse (actor: ActorModel, videoAbuseToCreateData: VideoAbuseObject) {
const options = {
arguments: [ actor, videoAbuseToCreateData ],
errorMessage: 'Cannot insert the remote video abuse with many retries.'
}
return retryTransactionWrapper(addRemoteVideoAbuse, options)
}
async function addRemoteVideoAbuse (actor: ActorModel, videoAbuseToCreateData: VideoAbuseObject) {
async function processCreateVideoAbuse (actor: ActorModel, videoAbuseToCreateData: VideoAbuseObject) {
logger.debug('Reporting remote abuse for video %s.', videoAbuseToCreateData.object)
const account = actor.Account
@ -139,16 +121,7 @@ async function addRemoteVideoAbuse (actor: ActorModel, videoAbuseToCreateData: V
})
}
function processCreateVideoComment (byActor: ActorModel, activity: ActivityCreate) {
const options = {
arguments: [ byActor, activity ],
errorMessage: 'Cannot create video comment with many retries.'
}
return retryTransactionWrapper(createVideoComment, options)
}
async function createVideoComment (byActor: ActorModel, activity: ActivityCreate) {
async function processCreateVideoComment (byActor: ActorModel, activity: ActivityCreate) {
const comment = activity.object as VideoCommentObject
const byAccount = byActor.Account

46
server/lib/activitypub/process/process-delete.ts
View file

@ -21,12 +21,12 @@ async function processDeleteActivity (activity: ActivityDelete) {
if (!actor.Account) throw new Error('Actor ' + actor.url + ' is a person but we cannot find it in database.')
actor.Account.Actor = await actor.Account.$get('Actor') as ActorModel
return processDeleteAccount(actor.Account)
return retryTransactionWrapper(processDeleteAccount, actor.Account)
} else if (actor.type === 'Group') {
if (!actor.VideoChannel) throw new Error('Actor ' + actor.url + ' is a group but we cannot find it in database.')
actor.VideoChannel.Actor = await actor.VideoChannel.$get('Actor') as ActorModel
return processDeleteVideoChannel(actor.VideoChannel)
return retryTransactionWrapper(processDeleteVideoChannel, actor.VideoChannel)
}
}
@ -34,14 +34,14 @@ async function processDeleteActivity (activity: ActivityDelete) {
{
const videoCommentInstance = await VideoCommentModel.loadByUrlAndPopulateAccount(objectUrl)
if (videoCommentInstance) {
return processDeleteVideoComment(actor, videoCommentInstance, activity)
return retryTransactionWrapper(processDeleteVideoComment, actor, videoCommentInstance, activity)
}
}
{
const videoInstance = await VideoModel.loadByUrlAndPopulateAccount(objectUrl)
if (videoInstance) {
return processDeleteVideo(actor, videoInstance)
return retryTransactionWrapper(processDeleteVideo, actor, videoInstance)
}
}
@ -57,15 +57,6 @@ export {
// ---------------------------------------------------------------------------
async function processDeleteVideo (actor: ActorModel, videoToDelete: VideoModel) {
const options = {
arguments: [ actor, videoToDelete ],
errorMessage: 'Cannot remove the remote video with many retries.'
}
await retryTransactionWrapper(deleteRemoteVideo, options)
}
async function deleteRemoteVideo (actor: ActorModel, videoToDelete: VideoModel) {
logger.debug('Removing remote video "%s".', videoToDelete.uuid)
await sequelizeTypescript.transaction(async t => {
@ -80,15 +71,6 @@ async function deleteRemoteVideo (actor: ActorModel, videoToDelete: VideoModel)
}
async function processDeleteAccount (accountToRemove: AccountModel) {
const options = {
arguments: [ accountToRemove ],
errorMessage: 'Cannot remove the remote account with many retries.'
}
await retryTransactionWrapper(deleteRemoteAccount, options)
}
async function deleteRemoteAccount (accountToRemove: AccountModel) {
logger.debug('Removing remote account "%s".', accountToRemove.Actor.uuid)
await sequelizeTypescript.transaction(async t => {
@ -99,15 +81,6 @@ async function deleteRemoteAccount (accountToRemove: AccountModel) {
}
async function processDeleteVideoChannel (videoChannelToRemove: VideoChannelModel) {
const options = {
arguments: [ videoChannelToRemove ],
errorMessage: 'Cannot remove the remote video channel with many retries.'
}
await retryTransactionWrapper(deleteRemoteVideoChannel, options)
}
async function deleteRemoteVideoChannel (videoChannelToRemove: VideoChannelModel) {
logger.debug('Removing remote video channel "%s".', videoChannelToRemove.Actor.uuid)
await sequelizeTypescript.transaction(async t => {
@ -117,16 +90,7 @@ async function deleteRemoteVideoChannel (videoChannelToRemove: VideoChannelModel
logger.info('Remote video channel with uuid %s removed.', videoChannelToRemove.Actor.uuid)
}
async function processDeleteVideoComment (byActor: ActorModel, videoComment: VideoCommentModel, activity: ActivityDelete) {
const options = {
arguments: [ byActor, videoComment, activity ],
errorMessage: 'Cannot remove the remote video comment with many retries.'
}
await retryTransactionWrapper(deleteRemoteVideoComment, options)
}
function deleteRemoteVideoComment (byActor: ActorModel, videoComment: VideoCommentModel, activity: ActivityDelete) {
function processDeleteVideoComment (byActor: ActorModel, videoComment: VideoCommentModel, activity: ActivityDelete) {
logger.debug('Removing remote video comment "%s".', videoComment.url)
return sequelizeTypescript.transaction(async t => {

13
server/lib/activitypub/process/process-follow.ts
View file

@ -11,7 +11,7 @@ async function processFollowActivity (activity: ActivityFollow) {
const activityObject = activity.object
const actor = await getOrCreateActorAndServerAndModel(activity.actor)
return processFollow(actor, activityObject)
return retryTransactionWrapper(processFollow, actor, activityObject)
}
// ---------------------------------------------------------------------------
@ -22,16 +22,7 @@ export {
// ---------------------------------------------------------------------------
function processFollow (actor: ActorModel, targetActorURL: string) {
const options = {
arguments: [ actor, targetActorURL ],
errorMessage: 'Cannot follow with many retries.'
}
return retryTransactionWrapper(follow, options)
}
async function follow (actor: ActorModel, targetActorURL: string) {
async function processFollow (actor: ActorModel, targetActorURL: string) {
await sequelizeTypescript.transaction(async t => {
const targetActor = await ActorModel.loadByUrl(targetActorURL, t)

16
server/lib/activitypub/process/process-like.ts
View file

@ -4,14 +4,13 @@ import { sequelizeTypescript } from '../../../initializers'
import { AccountVideoRateModel } from '../../../models/account/account-video-rate'
import { ActorModel } from '../../../models/activitypub/actor'
import { getOrCreateActorAndServerAndModel } from '../actor'
import { forwardActivity, forwardVideoRelatedActivity } from '../send/utils'
import { forwardVideoRelatedActivity } from '../send/utils'
import { getOrCreateAccountAndVideoAndChannel } from '../videos'
import { getActorsInvolvedInVideo } from '../audience'
async function processLikeActivity (activity: ActivityLike) {
const actor = await getOrCreateActorAndServerAndModel(activity.actor)
return processLikeVideo(actor, activity)
return retryTransactionWrapper(processLikeVideo, actor, activity)
}
// ---------------------------------------------------------------------------
@ -22,16 +21,7 @@ export {
// ---------------------------------------------------------------------------
async function processLikeVideo (actor: ActorModel, activity: ActivityLike) {
const options = {
arguments: [ actor, activity ],
errorMessage: 'Cannot like the video with many retries.'
}
return retryTransactionWrapper(createVideoLike, options)
}
async function createVideoLike (byActor: ActorModel, activity: ActivityLike) {
async function processLikeVideo (byActor: ActorModel, activity: ActivityLike) {
const videoUrl = activity.object
const byAccount = byActor.Account

48
server/lib/activitypub/process/process-undo.ts
View file

@ -18,13 +18,13 @@ async function processUndoActivity (activity: ActivityUndo) {
const actorUrl = getActorUrl(activity.actor)
if (activityToUndo.type === 'Like') {
return processUndoLike(actorUrl, activity)
return retryTransactionWrapper(processUndoLike, actorUrl, activity)
} else if (activityToUndo.type === 'Create' && activityToUndo.object.type === 'Dislike') {
return processUndoDislike(actorUrl, activity)
return retryTransactionWrapper(processUndoDislike, actorUrl, activity)
} else if (activityToUndo.type === 'Follow') {
return processUndoFollow(actorUrl, activityToUndo)
return retryTransactionWrapper(processUndoFollow, actorUrl, activityToUndo)
} else if (activityToUndo.type === 'Announce') {
return processUndoAnnounce(actorUrl, activityToUndo)
return retryTransactionWrapper(processUndoAnnounce, actorUrl, activityToUndo)
}
logger.warn('Unknown activity object type %s -> %s when undo activity.', activityToUndo.type, { activity: activity.id })
@ -40,16 +40,7 @@ export {
// ---------------------------------------------------------------------------
function processUndoLike (actorUrl: string, activity: ActivityUndo) {
const options = {
arguments: [ actorUrl, activity ],
errorMessage: 'Cannot undo like with many retries.'
}
return retryTransactionWrapper(undoLike, options)
}
async function undoLike (actorUrl: string, activity: ActivityUndo) {
async function processUndoLike (actorUrl: string, activity: ActivityUndo) {
const likeActivity = activity.object as ActivityLike
const { video } = await getOrCreateAccountAndVideoAndChannel(likeActivity.object)
@ -73,16 +64,7 @@ async function undoLike (actorUrl: string, activity: ActivityUndo) {
})
}
function processUndoDislike (actorUrl: string, activity: ActivityUndo) {
const options = {
arguments: [ actorUrl, activity ],
errorMessage: 'Cannot undo dislike with many retries.'
}
return retryTransactionWrapper(undoDislike, options)
}
async function undoDislike (actorUrl: string, activity: ActivityUndo) {
async function processUndoDislike (actorUrl: string, activity: ActivityUndo) {
const dislike = activity.object.object as DislikeObject
const { video } = await getOrCreateAccountAndVideoAndChannel(dislike.object)
@ -107,15 +89,6 @@ async function undoDislike (actorUrl: string, activity: ActivityUndo) {
}
function processUndoFollow (actorUrl: string, followActivity: ActivityFollow) {
const options = {
arguments: [ actorUrl, followActivity ],
errorMessage: 'Cannot undo follow with many retries.'
}
return retryTransactionWrapper(undoFollow, options)
}
function undoFollow (actorUrl: string, followActivity: ActivityFollow) {
return sequelizeTypescript.transaction(async t => {
const follower = await ActorModel.loadByUrl(actorUrl, t)
const following = await ActorModel.loadByUrl(followActivity.object, t)
@ -130,15 +103,6 @@ function undoFollow (actorUrl: string, followActivity: ActivityFollow) {
}
function processUndoAnnounce (actorUrl: string, announceActivity: ActivityAnnounce) {
const options = {
arguments: [ actorUrl, announceActivity ],
errorMessage: 'Cannot undo announce with many retries.'
}
return retryTransactionWrapper(undoAnnounce, options)
}
function undoAnnounce (actorUrl: string, announceActivity: ActivityAnnounce) {
return sequelizeTypescript.transaction(async t => {
const byAccount = await AccountModel.loadByUrl(actorUrl, t)
if (!byAccount) throw new Error('Unknown account ' + actorUrl)

26
server/lib/activitypub/process/process-update.ts
View file

@ -25,9 +25,9 @@ async function processUpdateActivity (activity: ActivityUpdate) {
const objectType = activity.object.type
if (objectType === 'Video') {
return processUpdateVideo(actor, activity)
return retryTransactionWrapper(processUpdateVideo, actor, activity)
} else if (objectType === 'Person' || objectType === 'Application' || objectType === 'Group') {
return processUpdateActor(actor, activity)
return retryTransactionWrapper(processUpdateActor, actor, activity)
}
return undefined
@ -41,16 +41,7 @@ export {
// ---------------------------------------------------------------------------
function processUpdateVideo (actor: ActorModel, activity: ActivityUpdate) {
const options = {
arguments: [ actor, activity ],
errorMessage: 'Cannot update the remote video with many retries'
}
return retryTransactionWrapper(updateRemoteVideo, options)
}
async function updateRemoteVideo (actor: ActorModel, activity: ActivityUpdate) {
async function processUpdateVideo (actor: ActorModel, activity: ActivityUpdate) {
const videoObject = activity.object as VideoTorrentObject
if (sanitizeAndCheckVideoTorrentObject(videoObject) === false) {
@ -136,16 +127,7 @@ async function updateRemoteVideo (actor: ActorModel, activity: ActivityUpdate) {
}
}
function processUpdateActor (actor: ActorModel, activity: ActivityUpdate) {
const options = {
arguments: [ actor, activity ],
errorMessage: 'Cannot update the remote actor with many retries'
}
return retryTransactionWrapper(updateRemoteActor, options)
}
async function updateRemoteActor (actor: ActorModel, activity: ActivityUpdate) {
async function processUpdateActor (actor: ActorModel, activity: ActivityUpdate) {
const actorAttributesToUpdate = activity.object as ActivityPubActor
logger.debug('Updating remote account "%s".', actorAttributesToUpdate.uuid)

7
server/lib/activitypub/videos.ts
View file

@ -228,12 +228,7 @@ async function getOrCreateAccountAndVideoAndChannel (videoObject: VideoTorrentOb
const channelActor = await getOrCreateVideoChannel(videoObject)
const options = {
arguments: [ videoObject, channelActor ],
errorMessage: 'Cannot insert the remote video with many retries.'
}
const video = await retryTransactionWrapper(getOrCreateVideo, options)
const video = await retryTransactionWrapper(getOrCreateVideo, videoObject, channelActor)
// Process outside the transaction because we could fetch remote data
logger.info('Adding likes of video %s.', video.uuid)

6
server/lib/job-queue/handlers/activitypub-follow.ts
View file

@ -26,12 +26,8 @@ async function processActivityPubFollow (job: kue.Job) {
const targetActor = await getOrCreateActorAndServerAndModel(actorUrl)
const fromActor = await getServerActor()
const options = {
arguments: [ fromActor, targetActor ],
errorMessage: 'Cannot follow with many retries.'
}
return retryTransactionWrapper(follow, options)
return retryTransactionWrapper(follow, fromActor, targetActor)
}
// ---------------------------------------------------------------------------

12
server/lib/job-queue/handlers/video-file.ts
View file

@ -52,19 +52,11 @@ async function processVideoFile (job: kue.Job) {
if (payload.resolution) {
await video.transcodeOriginalVideofile(payload.resolution, payload.isPortraitMode)
const options = {
arguments: [ video ],
errorMessage: 'Cannot execute onVideoFileTranscoderOrImportSuccess with many retries.'
}
await retryTransactionWrapper(onVideoFileTranscoderOrImportSuccess, options)
await retryTransactionWrapper(onVideoFileTranscoderOrImportSuccess, video)
} else {
await video.optimizeOriginalVideofile()
const options = {
arguments: [ video, payload.isNewVideo ],
errorMessage: 'Cannot execute onVideoFileOptimizerSuccess with many retries.'
}
await retryTransactionWrapper(onVideoFileOptimizerSuccess, options)
await retryTransactionWrapper(onVideoFileOptimizerSuccess, video, payload.isNewVideo)
}
return video

12
server/middlewares/async.ts
View file

@ -1,5 +1,6 @@
import { eachSeries } from 'async'
import { NextFunction, Request, RequestHandler, Response } from 'express'
import { retryTransactionWrapper } from '../helpers/database-utils'
// Syntactic sugar to avoid try/catch in express controllers
// Thanks: https://medium.com/@Abazhenov/using-async-await-in-express-with-node-8-b8af872c0016
@ -20,8 +21,17 @@ function asyncMiddleware (fun: RequestPromiseHandler | RequestPromiseHandler[])
}
}
function asyncRetryTransactionMiddleware (fun: RequestPromiseHandler) {
return (req: Request, res: Response, next: NextFunction) => {
return Promise.resolve(
retryTransactionWrapper(fun, req, res, next)
).catch(err => next(err))
}
}
// ---------------------------------------------------------------------------
export {
asyncMiddleware
asyncMiddleware,
asyncRetryTransactionMiddleware
}