]> git.immae.eu Git - github/Chocobozzz/PeerTube.git/blob - server/helpers/database-utils.ts
Feature/filter already watched videos (#5739)
[github/Chocobozzz/PeerTube.git] / server / helpers / database-utils.ts
1 import retry from 'async/retry'
2 import Bluebird from 'bluebird'
3 import { Transaction } from 'sequelize'
4 import { Model } from 'sequelize-typescript'
5 import { sequelizeTypescript } from '@server/initializers/database'
6 import { logger } from './logger'
7
8 function retryTransactionWrapper <T, A, B, C, D> (
9 functionToRetry: (arg1: A, arg2: B, arg3: C, arg4: D) => Promise<T>,
10 arg1: A,
11 arg2: B,
12 arg3: C,
13 arg4: D,
14 ): Promise<T>
15
16 function retryTransactionWrapper <T, A, B, C> (
17 functionToRetry: (arg1: A, arg2: B, arg3: C) => Promise<T>,
18 arg1: A,
19 arg2: B,
20 arg3: C
21 ): Promise<T>
22
23 function retryTransactionWrapper <T, A, B> (
24 functionToRetry: (arg1: A, arg2: B) => Promise<T>,
25 arg1: A,
26 arg2: B
27 ): Promise<T>
28
29 function retryTransactionWrapper <T, A> (
30 functionToRetry: (arg1: A) => Promise<T>,
31 arg1: A
32 ): Promise<T>
33
34 function retryTransactionWrapper <T> (
35 functionToRetry: () => Promise<T> | Bluebird<T>
36 ): Promise<T>
37
38 function retryTransactionWrapper <T> (
39 functionToRetry: (...args: any[]) => Promise<T>,
40 ...args: any[]
41 ): Promise<T> {
42 return transactionRetryer<T>(callback => {
43 functionToRetry.apply(null, args)
44 .then((result: T) => callback(null, result))
45 .catch(err => callback(err))
46 })
47 .catch(err => {
48 logger.warn(`Cannot execute ${functionToRetry.name} with many retries.`, { err })
49 throw err
50 })
51 }
52
53 function transactionRetryer <T> (func: (err: any, data: T) => any) {
54 return new Promise<T>((res, rej) => {
55 retry(
56 {
57 times: 5,
58
59 errorFilter: err => {
60 const willRetry = (err.name === 'SequelizeDatabaseError')
61 logger.debug('Maybe retrying the transaction function.', { willRetry, err, tags: [ 'sql', 'retry' ] })
62 return willRetry
63 }
64 },
65 func,
66 (err, data) => err ? rej(err) : res(data)
67 )
68 })
69 }
70
71 // ---------------------------------------------------------------------------
72
73 function updateInstanceWithAnother <M, T extends U, U extends Model<M>> (instanceToUpdate: T, baseInstance: U) {
74 const obj = baseInstance.toJSON()
75
76 for (const key of Object.keys(obj)) {
77 instanceToUpdate[key] = obj[key]
78 }
79 }
80
81 function resetSequelizeInstance <T> (instance: Model<T>) {
82 instance.set(instance.previous())
83 }
84
85 function filterNonExistingModels <T extends { hasSameUniqueKeysThan (other: T): boolean }> (
86 fromDatabase: T[],
87 newModels: T[]
88 ) {
89 return fromDatabase.filter(f => !newModels.find(newModel => newModel.hasSameUniqueKeysThan(f)))
90 }
91
92 function deleteAllModels <T extends Pick<Model, 'destroy'>> (models: T[], transaction: Transaction) {
93 return Promise.all(models.map(f => f.destroy({ transaction })))
94 }
95
96 // ---------------------------------------------------------------------------
97
98 function runInReadCommittedTransaction <T> (fn: (t: Transaction) => Promise<T>) {
99 const options = { isolationLevel: Transaction.ISOLATION_LEVELS.READ_COMMITTED }
100
101 return sequelizeTypescript.transaction(options, t => fn(t))
102 }
103
104 function afterCommitIfTransaction (t: Transaction, fn: Function) {
105 if (t) return t.afterCommit(() => fn())
106
107 return fn()
108 }
109
110 // ---------------------------------------------------------------------------
111
112 export {
113 resetSequelizeInstance,
114 retryTransactionWrapper,
115 transactionRetryer,
116 updateInstanceWithAnother,
117 afterCommitIfTransaction,
118 filterNonExistingModels,
119 deleteAllModels,
120 runInReadCommittedTransaction
121 }