]>
Commit | Line | Data |
---|---|---|
1 | import * as retry from 'async/retry' | |
2 | import * as Bluebird from 'bluebird' | |
3 | import { QueryTypes, 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> | Bluebird<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> | 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 | ||
34 | function retryTransactionWrapper <T> ( | |
35 | functionToRetry: () => Promise<T> | Bluebird<T> | |
36 | ): Promise<T> | |
37 | ||
38 | function retryTransactionWrapper <T> ( | |
39 | functionToRetry: (...args: any[]) => Promise<T> | Bluebird<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.error(`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 (instance: Model<any>, savedFields: object) { | |
82 | Object.keys(savedFields).forEach(key => { | |
83 | instance[key] = savedFields[key] | |
84 | }) | |
85 | } | |
86 | ||
87 | function deleteNonExistingModels <T extends { hasSameUniqueKeysThan (other: T): boolean } & Pick<Model, 'destroy'>> ( | |
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 | ||
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 | ||
108 | // --------------------------------------------------------------------------- | |
109 | ||
110 | function runInReadCommittedTransaction <T> (fn: (t: Transaction) => Promise<T>) { | |
111 | const options = { isolationLevel: Transaction.ISOLATION_LEVELS.READ_COMMITTED } | |
112 | ||
113 | return sequelizeTypescript.transaction(options, t => fn(t)) | |
114 | } | |
115 | ||
116 | function afterCommitIfTransaction (t: Transaction, fn: Function) { | |
117 | if (t) return t.afterCommit(() => fn()) | |
118 | ||
119 | return fn() | |
120 | } | |
121 | ||
122 | // --------------------------------------------------------------------------- | |
123 | ||
124 | export { | |
125 | resetSequelizeInstance, | |
126 | retryTransactionWrapper, | |
127 | transactionRetryer, | |
128 | updateInstanceWithAnother, | |
129 | afterCommitIfTransaction, | |
130 | deleteNonExistingModels, | |
131 | setAsUpdated, | |
132 | runInReadCommittedTransaction | |
133 | } |