mirror of
https://github.com/overleaf/overleaf.git
synced 2024-11-21 20:47:08 -05:00
Merge pull request #18850 from overleaf/ac-promisify-resync-subscriptions
Promisify resync_subscriptions GitOrigin-RevId: c23a99683a22104815525b179d4e46d3ad568f94
This commit is contained in:
parent
86e2abc90b
commit
50b19c7793
1 changed files with 87 additions and 79 deletions
|
@ -1,11 +1,12 @@
|
|||
const { Subscription } = require('../../app/src/models/Subscription')
|
||||
const RecurlyWrapper = require('../../app/src/Features/Subscription/RecurlyWrapper')
|
||||
const SubscriptionUpdater = require('../../app/src/Features/Subscription/SubscriptionUpdater')
|
||||
const async = require('async')
|
||||
const minimist = require('minimist')
|
||||
const { setTimeout } = require('node:timers/promises')
|
||||
|
||||
// make sure all `allMismatchReasons` are displayed in the output
|
||||
const util = require('util')
|
||||
const pLimit = require('p-limit')
|
||||
util.inspect.defaultOptions.maxArrayLength = null
|
||||
|
||||
const ScriptLogger = {
|
||||
|
@ -59,104 +60,109 @@ const ScriptLogger = {
|
|||
},
|
||||
}
|
||||
|
||||
const slowCallback = callback => setTimeout(callback, 80)
|
||||
|
||||
const handleSyncSubscriptionError = (subscription, error, callback) => {
|
||||
const handleSyncSubscriptionError = async (subscription, error) => {
|
||||
console.warn(`Errors with subscription id=${subscription._id}:`, error)
|
||||
if (typeof error === 'string' && error.match(/429$/)) {
|
||||
return setTimeout(callback, 1000 * 60 * 5)
|
||||
await setTimeout(1000 * 60 * 5)
|
||||
return
|
||||
}
|
||||
if (typeof error === 'string' && error.match(/5\d\d$/)) {
|
||||
return setTimeout(() => {
|
||||
syncSubscription(subscription, callback)
|
||||
}, 1000 * 60)
|
||||
await setTimeout(1000 * 60)
|
||||
await syncSubscription(subscription)
|
||||
return
|
||||
}
|
||||
slowCallback(callback)
|
||||
await setTimeout(80)
|
||||
}
|
||||
|
||||
const syncSubscription = (subscription, callback) => {
|
||||
RecurlyWrapper.getSubscription(
|
||||
subscription.recurlySubscription_id,
|
||||
(error, recurlySubscription) => {
|
||||
if (error) {
|
||||
return handleSyncSubscriptionError(subscription, error, callback)
|
||||
const syncSubscription = async subscription => {
|
||||
let recurlySubscription
|
||||
try {
|
||||
recurlySubscription = await RecurlyWrapper.promises.getSubscription(
|
||||
subscription.recurlySubscription_id
|
||||
)
|
||||
} catch (error) {
|
||||
await handleSyncSubscriptionError(subscription, error)
|
||||
return
|
||||
}
|
||||
|
||||
ScriptLogger.recordMismatch(subscription, recurlySubscription)
|
||||
|
||||
if (!COMMIT) {
|
||||
return callback()
|
||||
return
|
||||
}
|
||||
|
||||
SubscriptionUpdater.updateSubscriptionFromRecurly(
|
||||
try {
|
||||
await SubscriptionUpdater.promises.updateSubscriptionFromRecurly(
|
||||
recurlySubscription,
|
||||
subscription,
|
||||
{},
|
||||
error => {
|
||||
if (error) {
|
||||
return handleSyncSubscriptionError(subscription, error, callback)
|
||||
}
|
||||
slowCallback(callback)
|
||||
}
|
||||
{}
|
||||
)
|
||||
} catch (error) {
|
||||
await handleSyncSubscriptionError(subscription, error)
|
||||
return
|
||||
}
|
||||
await setTimeout(80)
|
||||
}
|
||||
|
||||
const syncSubscriptions = async subscriptions => {
|
||||
const limit = pLimit(ASYNC_LIMIT)
|
||||
return await Promise.all(
|
||||
subscriptions.map(subscription =>
|
||||
limit(() => syncSubscription(subscription))
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
const syncSubscriptions = (subscriptions, callback) => {
|
||||
async.eachLimit(subscriptions, ASYNC_LIMIT, syncSubscription, callback)
|
||||
}
|
||||
const loopForSubscriptions = async skipInitial => {
|
||||
let skip = skipInitial
|
||||
|
||||
const loopForSubscriptions = (skip, callback) => {
|
||||
Subscription.find({
|
||||
// iterate while there are more subscriptions to fetch
|
||||
while (true) {
|
||||
const subscriptions = await Subscription.find({
|
||||
recurlySubscription_id: { $exists: true, $ne: '' },
|
||||
})
|
||||
.sort('_id')
|
||||
.skip(skip)
|
||||
.limit(FETCH_LIMIT)
|
||||
.exec((error, subscriptions) => {
|
||||
if (error) {
|
||||
return callback(error)
|
||||
}
|
||||
.exec()
|
||||
|
||||
if (subscriptions.length === 0) {
|
||||
console.warn('DONE')
|
||||
return callback()
|
||||
return
|
||||
}
|
||||
|
||||
syncSubscriptions(subscriptions, error => {
|
||||
if (error) {
|
||||
return callback(error)
|
||||
}
|
||||
await syncSubscriptions(subscriptions)
|
||||
|
||||
ScriptLogger.checkedSubscriptionsCount += subscriptions.length
|
||||
retryCounter = 0
|
||||
ScriptLogger.printProgress()
|
||||
ScriptLogger.printSummary()
|
||||
loopForSubscriptions(
|
||||
MONGO_SKIP + ScriptLogger.checkedSubscriptionsCount,
|
||||
callback
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
skip += FETCH_LIMIT
|
||||
}
|
||||
}
|
||||
|
||||
let retryCounter = 0
|
||||
const run = () =>
|
||||
loopForSubscriptions(
|
||||
MONGO_SKIP + ScriptLogger.checkedSubscriptionsCount,
|
||||
error => {
|
||||
if (error) {
|
||||
const run = async () => {
|
||||
while (true) {
|
||||
try {
|
||||
await loopForSubscriptions(
|
||||
MONGO_SKIP + ScriptLogger.checkedSubscriptionsCount
|
||||
)
|
||||
break
|
||||
} catch (error) {
|
||||
if (retryCounter < 3) {
|
||||
console.error(error)
|
||||
retryCounter += 1
|
||||
console.warn(`RETRYING IN 60 SECONDS. (${retryCounter}/3)`)
|
||||
return setTimeout(run, 60000)
|
||||
}
|
||||
await setTimeout(60000)
|
||||
} else {
|
||||
console.error('Failed after 3 retries')
|
||||
throw error
|
||||
}
|
||||
process.exit()
|
||||
}
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
let FETCH_LIMIT, ASYNC_LIMIT, COMMIT, MONGO_SKIP
|
||||
const setup = () => {
|
||||
|
@ -174,4 +180,6 @@ const setup = () => {
|
|||
}
|
||||
|
||||
setup()
|
||||
run()
|
||||
run().then(() => {
|
||||
process.exit()
|
||||
})
|
||||
|
|
Loading…
Reference in a new issue