]>
Commit | Line | Data |
---|---|---|
4d4e5cd4 | 1 | import * as retry from 'async/retry' |
20494f12 | 2 | import * as Bluebird from 'bluebird' |
e024fd6a | 3 | import { QueryTypes, Transaction } from 'sequelize' |
06215f15 | 4 | import { Model } from 'sequelize-typescript' |
e024fd6a | 5 | import { sequelizeTypescript } from '@server/initializers/database' |
65fcc311 | 6 | import { logger } from './logger' |
4145c1c6 | 7 | |
77d7e851 C |
8 | function retryTransactionWrapper <T, A, B, C, D> ( |
9 | functionToRetry: (arg1: A, arg2: B, arg3: C, arg4: D) => Promise<T> | Bluebird<T>, | |
10 | arg1: A, | |
11 | arg2: B, | |
12 | arg3: C, | |
13 | arg4: D, | |
14 | ): Promise<T> | |
15 | ||
90d4bb81 C |
16 | function retryTransactionWrapper <T, A, B, C> ( |
17 | functionToRetry: (arg1: A, arg2: B, arg3: C) => Promise<T> | Bluebird<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> | Bluebird<T>, | |
25 | arg1: A, | |
26 | arg2: B | |
27 | ): Promise<T> | |
28 | ||
29 | function retryTransactionWrapper <T, A> ( | |
30 | functionToRetry: (arg1: A) => Promise<T> | Bluebird<T>, | |
31 | arg1: A | |
32 | ): Promise<T> | |
33 | ||
2baea0c7 C |
34 | function retryTransactionWrapper <T> ( |
35 | functionToRetry: () => Promise<T> | Bluebird<T> | |
36 | ): Promise<T> | |
37 | ||
0f91ae62 | 38 | function retryTransactionWrapper <T> ( |
90d4bb81 C |
39 | functionToRetry: (...args: any[]) => Promise<T> | Bluebird<T>, |
40 | ...args: any[] | |
0f91ae62 | 41 | ): Promise<T> { |
0f91ae62 | 42 | return transactionRetryer<T>(callback => { |
2baea0c7 | 43 | functionToRetry.apply(null, args) |
0f91ae62 | 44 | .then((result: T) => callback(null, result)) |
6fcd19ba | 45 | .catch(err => callback(err)) |
075f16ca | 46 | }) |
6fcd19ba | 47 | .catch(err => { |
1e11f67b | 48 | logger.error(`Cannot execute ${functionToRetry.name} with many retries.`, { err }) |
20494f12 | 49 | throw err |
6fcd19ba | 50 | }) |
4df023f2 C |
51 | } |
52 | ||
0f91ae62 C |
53 | function transactionRetryer <T> (func: (err: any, data: T) => any) { |
54 | return new Promise<T>((res, rej) => { | |
90d4bb81 C |
55 | retry( |
56 | { | |
57 | times: 5, | |
58 | ||
59 | errorFilter: err => { | |
60 | const willRetry = (err.name === 'SequelizeDatabaseError') | |
28dfb44b | 61 | logger.debug('Maybe retrying the transaction function.', { willRetry, err, tags: [ 'sql', 'retry' ] }) |
90d4bb81 C |
62 | return willRetry |
63 | } | |
64 | }, | |
65 | func, | |
66 | (err, data) => err ? rej(err) : res(data) | |
67 | ) | |
4df023f2 C |
68 | }) |
69 | } | |
70 | ||
28dfb44b C |
71 | // --------------------------------------------------------------------------- |
72 | ||
16c016e8 | 73 | function updateInstanceWithAnother <M, T extends U, U extends Model<M>> (instanceToUpdate: T, baseInstance: U) { |
a5625b41 C |
74 | const obj = baseInstance.toJSON() |
75 | ||
76 | for (const key of Object.keys(obj)) { | |
1735c825 | 77 | instanceToUpdate[key] = obj[key] |
a5625b41 C |
78 | } |
79 | } | |
80 | ||
06215f15 C |
81 | function resetSequelizeInstance (instance: Model<any>, savedFields: object) { |
82 | Object.keys(savedFields).forEach(key => { | |
1735c825 | 83 | instance[key] = savedFields[key] |
06215f15 C |
84 | }) |
85 | } | |
86 | ||
16c016e8 | 87 | function deleteNonExistingModels <T extends { hasSameUniqueKeysThan (other: T): boolean } & Pick<Model, 'destroy'>> ( |
d7a25329 C |
88 | fromDatabase: T[], |
89 | newModels: T[], | |
90 | t: Transaction | |
91 | ) { | |
92 | return fromDatabase.filter(f => !newModels.find(newModel => newModel.hasSameUniqueKeysThan(f))) | |
93 | .map(f => f.destroy({ transaction: t })) | |
94 | } | |
95 | ||
e024fd6a C |
96 | // Sequelize always skip the update if we only update updatedAt field |
97 | function setAsUpdated (table: string, id: number, transaction?: Transaction) { | |
98 | return sequelizeTypescript.query( | |
99 | `UPDATE "${table}" SET "updatedAt" = :updatedAt WHERE id = :id`, | |
100 | { | |
101 | replacements: { table, id, updatedAt: new Date() }, | |
102 | type: QueryTypes.UPDATE, | |
103 | transaction | |
104 | } | |
105 | ) | |
106 | } | |
107 | ||
4df023f2 C |
108 | // --------------------------------------------------------------------------- |
109 | ||
28dfb44b | 110 | function runInReadCommittedTransaction <T> (fn: (t: Transaction) => Promise<T>) { |
a6a12dae C |
111 | const options = { isolationLevel: Transaction.ISOLATION_LEVELS.READ_COMMITTED } |
112 | ||
113 | return sequelizeTypescript.transaction(options, t => fn(t)) | |
28dfb44b C |
114 | } |
115 | ||
116 | function afterCommitIfTransaction (t: Transaction, fn: Function) { | |
117 | if (t) return t.afterCommit(() => fn()) | |
118 | ||
119 | return fn() | |
120 | } | |
121 | ||
122 | // --------------------------------------------------------------------------- | |
123 | ||
65fcc311 | 124 | export { |
06215f15 | 125 | resetSequelizeInstance, |
65fcc311 | 126 | retryTransactionWrapper, |
a5625b41 | 127 | transactionRetryer, |
2284f202 | 128 | updateInstanceWithAnother, |
d7a25329 | 129 | afterCommitIfTransaction, |
e024fd6a | 130 | deleteNonExistingModels, |
28dfb44b C |
131 | setAsUpdated, |
132 | runInReadCommittedTransaction | |
65fcc311 | 133 | } |