Skip to content

Commit

Permalink
fix: 🐛 exit job correctly
Browse files Browse the repository at this point in the history 
Signed-off-by: Manuel Ruck <[email protected]>
  • Loading branch information
Manuel Ruck committed Oct 22, 2023
1 parent 3d2c7c7 commit 46481f3
Show file tree
Hide file tree
Showing 38 changed files with 197 additions and 665 deletions.
20 changes: 19 additions & 1 deletion common/bundestagio/src/models/mongoose.ts
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,21 @@
import Mongoose from "mongoose";
import Mongoose from 'mongoose';
const mongoose = Mongoose;
export { mongoose };

export const mongoConnect = async () =>
new Promise(async (resolve) => {
// Mongo Debug
mongoose.set('debug', false);

mongoose.connect(process.env.DB_URL!);

mongoose.connection.once('connected', () => {
console.info('MongoDB is running');
resolve(true);
});
mongoose.connection.on('error', (e: Error) => {
// Unknown if this ends up in main - therefore we log here
console.error(e.stack);
throw e;
});
});
26 changes: 13 additions & 13 deletions common/democracy/src/models/index.ts
View file
Original file line number Diff line number Diff line change
@@ -1,13 +1,13 @@
export * from "./mongoose";
export * from "./activity";
export * from "./cronjob";
export * from "./deputy";
export * from "./device";
export * from "./document";
export * from "./phone";
export * from "./procedure";
export * from "./pushNotification";
export * from "./search-term";
export * from "./user";
export * from "./verification";
export * from "./vote";
export * from './mongoose';
export * from './activity';
export * from './cronjob';
export * from './deputy';
export * from './device';
export * from './document';
export * from './phone';
export * from './procedure';
export * from './pushNotification';
export * from './search-term';
export * from './user';
export * from './verification';
export * from './vote';
20 changes: 19 additions & 1 deletion common/democracy/src/models/mongoose.ts
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,21 @@
import Mongoose from "mongoose";
import Mongoose from 'mongoose';
const mongoose = Mongoose;
export { mongoose };

export const mongoConnect = async () =>
new Promise(async (resolve) => {
// Mongo Debug
mongoose.set('debug', false);

mongoose.connect(process.env.DB_URL!);

mongoose.connection.once('connected', () => {
console.info('MongoDB is running');
resolve(true);
});
mongoose.connection.on('error', (e: Error) => {
// Unknown if this ends up in main - therefore we log here
console.error(e.stack);
throw e;
});
});
83 changes: 36 additions & 47 deletions services/cron-jobs/cleanup-push-queue/src/index.ts
View file
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
import { mongoConnect, mongoDisconnect } from "./mongoose";
import {
DeviceModel,
PushNotificationModel,
PUSH_CATEGORY,
setCronStart,
setCronSuccess,
} from "@democracy-deutschland/democracy-common";
import { DB_URL, ENTRY_PERSIST_MILLISECONDS } from "./config";
} from '@democracy-deutschland/democracy-common';
import { DB_URL, ENTRY_PERSIST_MILLISECONDS } from './config';
import { mongoConnect } from '@democracy-deutschland/democracy-common';

/*
This service cleanes up the democrac.pushnotifications collection:
Expand All @@ -18,18 +18,12 @@ import { DB_URL, ENTRY_PERSIST_MILLISECONDS } from "./config";

const cleanupPushQueue = async () => {
// if ENTRY_PERSIST_MILLISECONDS is set, entries will persist
const updatedAt = ENTRY_PERSIST_MILLISECONDS
? { $lt: new Date(Date.now() - ENTRY_PERSIST_MILLISECONDS) }
: undefined;
const updatedAt = ENTRY_PERSIST_MILLISECONDS ? { $lt: new Date(Date.now() - ENTRY_PERSIST_MILLISECONDS) } : undefined;

const oldPushs = await PushNotificationModel.deleteMany({
sent: true,
category: {
$in: [
PUSH_CATEGORY.CONFERENCE_WEEK,
PUSH_CATEGORY.CONFERENCE_WEEK_VOTE,
PUSH_CATEGORY.OUTCOME,
],
$in: [PUSH_CATEGORY.CONFERENCE_WEEK, PUSH_CATEGORY.CONFERENCE_WEEK_VOTE, PUSH_CATEGORY.OUTCOME],
},
...(updatedAt && { updatedAt }),
});
Expand All @@ -38,15 +32,15 @@ const cleanupPushQueue = async () => {
};

const cleanupDuplicateTop100 = async () => {
console.log("#start cleanupDuplicateTop100");
console.log('#start cleanupDuplicateTop100');

const procedureIds = await PushNotificationModel.distinct("procedureIds");
const procedureIds = await PushNotificationModel.distinct('procedureIds');

let totalDups = 0;
let counter = 0;
for (let procedureId of procedureIds) {
for (const procedureId of procedureIds) {
counter++;
console.log(" # ## # #", procedureId, `${counter}/${procedureIds.length}`);
console.log(' # ## # #', procedureId, `${counter}/${procedureIds.length}`);
const duplicates = await PushNotificationModel.aggregate<{
type: string;
category: string;
Expand All @@ -56,7 +50,7 @@ const cleanupDuplicateTop100 = async () => {
time: Date;
token: string;
os: string;
_id: import("mongoose").Types.ObjectId;
_id: import('mongoose').Types.ObjectId;
createdAt: Date;
updatedAt: Date;
__v: number;
Expand All @@ -65,24 +59,24 @@ const cleanupDuplicateTop100 = async () => {
}>([
{
$match: {
category: "top100",
category: 'top100',
procedureIds: procedureId,
},
},
{
$unwind: "$procedureIds",
$unwind: '$procedureIds',
},
{
$group: {
_id: {
token: "$token",
procedureId: "$procedureIds",
token: '$token',
procedureId: '$procedureIds',
},
count: {
$sum: 1,
},
doc: {
$first: "$$ROOT",
$first: '$$ROOT',
},
},
},
Expand All @@ -91,9 +85,9 @@ const cleanupDuplicateTop100 = async () => {
newRoot: {
$mergeObjects: [
{
count: "$count",
count: '$count',
},
"$doc",
'$doc',
],
},
},
Expand All @@ -112,20 +106,20 @@ const cleanupDuplicateTop100 = async () => {
},
]);

for (let push of duplicates) {
for (const push of duplicates) {
const dups = await PushNotificationModel.deleteMany({
category: "top100",
category: 'top100',
procedureIds: push.procedureIds,
token: push.token,
_id: { $ne: push._id },
});
process.stdout.write(".");
process.stdout.write('.');
totalDups += dups?.deletedCount || 0;
}
process.stdout.write("\n");
process.stdout.write('\n');
}
console.log(`removed duplicate top100 pushs: ${totalDups}`);
console.log("#finish cleanupDuplicateTop100");
console.log('#finish cleanupDuplicateTop100');
};

const removeDuplicateTokens = async () => {
Expand All @@ -135,15 +129,15 @@ const removeDuplicateTokens = async () => {
}>([
{
$project: {
"pushTokens.token": 1,
'pushTokens.token': 1,
},
},
{
$unwind: "$pushTokens",
$unwind: '$pushTokens',
},
{
$group: {
_id: "$pushTokens.token",
_id: '$pushTokens.token',
count: {
$sum: 1,
},
Expand All @@ -160,45 +154,43 @@ const removeDuplicateTokens = async () => {

console.log(`found ${duplicatedTokens.length} devices with duplicate tokens`);

for (let duplicateToken of duplicatedTokens) {
for (const duplicateToken of duplicatedTokens) {
const devices = await DeviceModel.find(
{
pushTokens: { $elemMatch: { token: duplicateToken._id } },
},
{},
{ sort: { createdAt: 1 } }
{ sort: { createdAt: 1 } },
);
/** Remove duplicate tokens from device */
for (let device of devices) {
for (const device of devices) {
if (device) {
device.pushTokens = device.pushTokens.filter(
(elem, index) =>
device.pushTokens.findIndex((obj) => obj.token === elem.token) ===
index
(elem, index) => device.pushTokens.findIndex((obj) => obj.token === elem.token) === index,
);
await device.save();
}
process.stdout.write(".");
process.stdout.write('.');
}

/** Remove tokens from old devices */
devices.pop();
for (let device of devices) {
for (const device of devices) {
device.pushTokens.filter(({ token }) => token !== duplicateToken._id);
await device.save();
}
}

process.stdout.write("\n");
process.stdout.write('\n');
};

(async () => {
if (!DB_URL) {
throw new Error("you have to set environment variable: DB_URL");
throw new Error('you have to set environment variable: DB_URL');
}
await mongoConnect();

const CRON_NAME = "cleanup-push-queue";
const CRON_NAME = 'cleanup-push-queue';
const startDate = new Date();
await setCronStart({ name: CRON_NAME, startDate });

Expand All @@ -212,8 +204,5 @@ const removeDuplicateTokens = async () => {
await removeDuplicateTokens();

await setCronSuccess({ name: CRON_NAME, successStartDate: startDate });
await mongoDisconnect();
})().catch(async (e) => {
await mongoDisconnect();
throw e;
});
process.exit(0);
})();
27 changes: 0 additions & 27 deletions services/cron-jobs/cleanup-push-queue/src/mongoose.ts
View file

This file was deleted.

46 changes: 12 additions & 34 deletions services/cron-jobs/crawler/src/import-procedures/index.ts
View file
Original file line number Diff line number Diff line change
@@ -1,39 +1,17 @@
import { mongoConnect, mongoDisconnect } from '../mongoose';
import { getCron, setCronError, setCronStart, setCronSuccess } from '@democracy-deutschland/bundestagio-common';
import { getCron, mongoConnect, setCronStart, setCronSuccess } from '@democracy-deutschland/bundestagio-common';
import config from '../config';
import importProcedures from './import-procedures';
import debug from 'debug';
const error = debug('bundestag-io:import-procedures:error');

(async () => {
let withError = false;
try {
await mongoConnect();
const cronjob = await getCron({ name: 'import-procedures' });
await setCronStart({ name: 'import-procedures' });
await importProcedures({
...config,
IMPORT_PROCEDURES_FILTER_AFTER:
// cronjob?.lastSuccessDate?.toISOString().slice(0, 10) ||
config.IMPORT_PROCEDURES_FILTER_AFTER,
});
await setCronSuccess({ name: 'import-procedures', successStartDate: cronjob.lastStartDate || new Date() });
} catch (err) {
withError = true;
error(err);
try {
if (typeof err === 'string') {
await setCronError({ name: 'import-procedures', error: err });
} else {
console.error(err);
}
} catch (error) {
console.error(error);
}
} finally {
mongoDisconnect();
}
if (withError) {
process.exit(1);
}
await mongoConnect();
const cronjob = await getCron({ name: 'import-procedures' });
await setCronStart({ name: 'import-procedures' });
await importProcedures({
...config,
IMPORT_PROCEDURES_FILTER_AFTER:
// cronjob?.lastSuccessDate?.toISOString().slice(0, 10) ||
config.IMPORT_PROCEDURES_FILTER_AFTER,
});
await setCronSuccess({ name: 'import-procedures', successStartDate: cronjob.lastStartDate || new Date() });
process.exit(0);
})();
24 changes: 0 additions & 24 deletions services/cron-jobs/crawler/src/mongoose.ts
View file

This file was deleted.

Loading

0 comments on commit 46481f3

Please sign in to comment.