add collect hourly aqi

backend/package.json

@@ -39,6 +39,7 @@
 		"mysql2": "^3.6.1",
 		"nodemon": "^3.0.1",
 		"puppeteer": "^21.3.8",
+		"redis": "^4.6.12",
 		"socket.io": "^4.7.2",
 		"uuid": "^9.0.1",
 		"web-push": "^3.6.6",

backend/src/app.js

@@ -4,9 +4,7 @@ const cors = require("cors");
 const logger = require("./config/logger");
 const webpush = require("web-push");
 const config = require("./config/config");
-// const { captureScreenshot } = require("./puppeteer/puppeteer");
 const { calcAQI } = require("./helper/calculateTotalAQI");
-const Agendash = require("agendash");
 const v1Route = require("./routes/v1");
 const { errors } = require("celebrate");
 const { errorHandler } = require("./middlewares/errors");
@@ -15,6 +13,8 @@ const httpStatus = require("http-status");
 
 const { producer } = require("./rabbitmq/producer");
 
+const { redisClient } = require("./redis/redis");
+
 const start = async (agenda) => {
 	const app = express();
 
@@ -45,6 +45,11 @@ const start = async (agenda) => {
 const apiRoutes = (app, io, mqtt) => {
 	app.use("/v1", v1Route);
 
+	app.get("/test-cache", async (req, res, next) => {
+		const value = await redisClient.get("value");
+		res.send(value);
+	});
+
 	app.post("/amqp", async (req, res, next) => {
 		await producer.publishEmailMessage(req.body.mailList);
 		res.send("Sent");

backend/src/config/config.js

@@ -29,6 +29,9 @@ const envVarsSchema = Joi.object()
 		RABBITMQ: Joi.string(),
 		RABBITMQ_MAILSERVICE_EXCHANGE_NAME: Joi.string(),
 		RABBITMQ_MAILSERVICE_QUEUE_NAME: Joi.string(),
+		REDIS_HOST: Joi.string(),
+		REDIS_PORT: Joi.string(),
+		REDIS_PASSWORD: Joi.string(),
 	})
 	.unknown();
 
@@ -70,4 +73,9 @@ module.exports = {
 		mailServiceExchangeName: envVars.RABBITMQ_MAILSERVICE_EXCHANGE_NAME,
 		mailServiceQueueName: envVars.RABBITMQ_MAILSERVICE_QUEUE_NAME,
 	},
+	redis: {
+		host: envVars.REDIS_HOST,
+		port: envVars.REDIS_PORT,
+		password: envVars.REDIS_PASSWORD,
+	},
 };

backend/src/index.js

@@ -22,7 +22,8 @@ const startAgenda = async () => {
 	// Local
 	// await agenda.every(CronTime.everyDayAt(23), "retrieveDailyAqi");
 
-	await agenda.every(CronTime.everyDayAt(16), "retrieveDailyAqi");
+	await agenda.every(CronTime.everyDayAt(16, 5), "retrieveDailyAqi");
+	await agenda.every(CronTime.everyHour(), "getHourlyAqi");
 };
 
 mongoose

backend/src/jobs/definitions/getHourlyAqi.js

@@ -0,0 +1,54 @@
+const { aqiApiToken, predictionServiceUrl } = require("../../config/config");
+const axios = require("axios");
+const logger = require("../../config/logger");
+const HourlyAQIModel = require("../../models/HourlyAQI");
+
+const AQI_URL = "http://api.waqi.info";
+
+const getHourlyAqi = async (agenda) => {
+	try {
+		// define(jobName, fn, [options])
+		agenda.define(
+			"getHourlyAqi",
+			async function (job, done) {
+				try {
+					var response = await axios({
+						method: "get",
+						url: `${AQI_URL}/feed/@1583/?token=${aqiApiToken}`,
+					});
+
+					response = response.data;
+
+					const aqi = await HourlyAQIModel.findOne({
+						dateTime: response.data.time.iso,
+					});
+
+					logger.info(
+						`New AQI retrieved for date ${response.data.time.iso} from ${AQI_URL}: ${response.data.aqi}`
+					);
+
+					if (!aqi) {
+						await HourlyAQIModel.create({
+							aqi: response.data.aqi,
+							dateTime: response.data.time.iso,
+						});
+					}
+
+					done();
+				} catch (error) {
+					console.error(error);
+
+					logger.error(
+						"Failed to update new daily aqi in job [agenda.retrieveDailyAqi] : " +
+							error.message
+					);
+				}
+			},
+			{ priority: "highest", concurrency: 20 }
+		);
+	} catch (error) {
+		logger.error("Error in job [agenda.retrieveDailyAqi] : " + error);
+	}
+};
+
+module.exports = { getHourlyAqi };

backend/src/jobs/definitions/index.js

@@ -1,11 +1,12 @@
 const { retrieveDailyAqi } = require("./retrieveDailyAqi");
+const { getHourlyAqi } = require("./getHourlyAqi");
 const logger = require("../../config/logger");
 
-let definitions = [retrieveDailyAqi];
+let definitions = [retrieveDailyAqi, getHourlyAqi];
 
 const loadDefinitions = async (agenda) => {
 	try {
-		console.log()
+		console.log();
 		for (let definition of definitions) {
 			await definition(agenda);
 		}