Handle tempban change stream fallback

This commit is contained in:
2026-01-27 09:12:13 +01:00
parent 6dc3f2632a
commit 6a1c81fa3c
+44 -3
View File
@@ -7,9 +7,18 @@ module.exports = {
const dbClient = await connectToDatabase(); const dbClient = await connectToDatabase();
const db = dbClient.db('discord'); const db = dbClient.db('discord');
const tempBanCollection = db.collection('tempban'); const tempBanCollection = db.collection('tempban');
const scheduledBanIds = new Set();
let pollingInterval = null;
// 2) Define a function to schedule unbans // 2) Define a function to schedule unbans
async function scheduleUnban(banData) { async function scheduleUnban(banData) {
const banId = banData?._id?.toString();
if (banId && scheduledBanIds.has(banId)) {
return;
}
if (banId) {
scheduledBanIds.add(banId);
}
const delay = banData.banTime - Date.now(); const delay = banData.banTime - Date.now();
if (delay <= 0) { if (delay <= 0) {
@@ -23,22 +32,54 @@ module.exports = {
await tempBanCollection.deleteOne({ _id: banData._id }); await tempBanCollection.deleteOne({ _id: banData._id });
} catch (err) { } catch (err) {
console.error('Error unbanning member:', err); console.error('Error unbanning member:', err);
} finally {
if (banId) {
scheduledBanIds.delete(banId);
}
} }
}, delay); }, delay);
} }
try { async function loadAndScheduleBans() {
const allBans = await tempBanCollection.find({}).toArray(); const allBans = await tempBanCollection.find({}).toArray();
allBans.forEach(scheduleUnban); allBans.forEach(scheduleUnban);
}
tempBanCollection.watch().on('change', (change) => { function startPollingFallback() {
if (pollingInterval) {
return;
}
console.warn('Change streams unavailable; falling back to polling for temp bans.');
pollingInterval = setInterval(() => {
loadAndScheduleBans().catch((err) => {
console.error('Error polling temp bans:', err);
});
}, 60_000);
}
try {
await loadAndScheduleBans();
const changeStream = tempBanCollection.watch();
changeStream.on('change', (change) => {
if (change.operationType === 'insert') { if (change.operationType === 'insert') {
const newBan = change.fullDocument; const newBan = change.fullDocument;
scheduleUnban(newBan); scheduleUnban(newBan);
} }
}); });
changeStream.on('error', (err) => {
if (err?.code === 40573) {
startPollingFallback();
} else {
console.error('Change stream error:', err);
}
});
} catch (err) { } catch (err) {
console.error('Error setting up unban system:', err); if (err?.code === 40573) {
startPollingFallback();
} else {
console.error('Error setting up unban system:', err);
}
} }
}, },
}; };