From a8e1454d45f6c0b26a461ecc00c2a2ccd8843167 Mon Sep 17 00:00:00 2001 From: Behzad Rabiei <53224485+Behzad-rabiei@users.noreply.github.com> Date: Tue, 17 Sep 2024 18:53:56 +0400 Subject: [PATCH 1/5] feat: add discourse platform; --- src/services/platform.service.ts | 2 ++ src/validations/module.validation.ts | 45 +++++++++++++++++++++++++- src/validations/platform.validation.ts | 13 ++++++++ 3 files changed, 59 insertions(+), 1 deletion(-) diff --git a/src/services/platform.service.ts b/src/services/platform.service.ts index 0935141..054fb61 100644 --- a/src/services/platform.service.ts +++ b/src/services/platform.service.ts @@ -150,6 +150,8 @@ function getMetadataKey(platformName: string): string { return 'workspace_id'; case PlatformNames.MediaWiki: return 'baseURL'; + case PlatformNames.Discourse: + return 'id'; default: throw new Error('Unsupported platform'); } diff --git a/src/validations/module.validation.ts b/src/validations/module.validation.ts index 1bc7759..1503476 100644 --- a/src/validations/module.validation.ts +++ b/src/validations/module.validation.ts @@ -1,6 +1,11 @@ import Joi from 'joi'; import { objectId } from './custom.validation'; -import { PlatformNames, ModuleNames, HivemindPlatformNames } from '@togethercrew.dev/db'; +import { + PlatformNames, + ModuleNames, + HivemindPlatformNames, + ViolationDetectionPlatformNames, +} from '@togethercrew.dev/db'; const createModule = { body: Joi.object().keys({ @@ -110,6 +115,37 @@ const hivemindOptions = () => { }); }; +const violationDetectionMetadata = () => { + return Joi.object().keys({ + selectedEmails: Joi.array().items(Joi.string().email()), + fromDate: Joi.date(), + toDate: Joi.date().valid(null), + selectedResources: Joi.array().items(Joi.number().empty()), + }); +}; + +const violationDetectionOptions = () => { + return Joi.object().keys({ + platforms: Joi.array().items( + Joi.object().keys({ + name: Joi.string() + .required() + .valid(...Object.values(ViolationDetectionPlatformNames)), + platform: Joi.string().custom(objectId).required(), + metadata: Joi.when('name', { + switch: [ + { + is: PlatformNames.Discourse, + then: violationDetectionMetadata(), + }, + ], + otherwise: Joi.any().forbidden(), + }).required(), + }), + ), + }); +}; + const dynamicModuleUpdate = (req: any) => { const moduleName = req.module?.name; const paramsOption = { @@ -127,6 +163,13 @@ const dynamicModuleUpdate = (req: any) => { }), }; break; + case ModuleNames.ViolationDetection: + bodyOption = { + body: Joi.object().required().keys({ + options: violationDetectionOptions(), + }), + }; + break; default: req.allowInput = false; return {}; diff --git a/src/validations/platform.validation.ts b/src/validations/platform.validation.ts index e248321..44d95d9 100644 --- a/src/validations/platform.validation.ts +++ b/src/validations/platform.validation.ts @@ -74,6 +74,15 @@ const mediaWikiMetadata = () => { }); }; +const discourseMetadata = () => { + return Joi.object().keys({ + id: Joi.string().required(), + period: Joi.date().required(), + analyzerStartedAt: Joi.date().required(), + resources: Joi.array().items(Joi.number().empty()).required(), + }); +}; + const createPlatform = { body: Joi.object().keys({ name: Joi.string() @@ -106,6 +115,10 @@ const createPlatform = { is: PlatformNames.MediaWiki, then: mediaWikiMetadata(), }, + { + is: PlatformNames.Discourse, + then: discourseMetadata(), + }, ], }).required(), }), From 5555862679738529cd17cd832f42a177c157431f Mon Sep 17 00:00:00 2001 From: Behzad Rabiei <53224485+Behzad-rabiei@users.noreply.github.com> Date: Tue, 17 Sep 2024 18:54:20 +0400 Subject: [PATCH 2/5] docs: update doc --- src/docs/module.doc.yml | 83 ++++++++++++++++++++++++--------------- src/docs/platform.doc.yml | 16 ++++++++ 2 files changed, 67 insertions(+), 32 deletions(-) diff --git a/src/docs/module.doc.yml b/src/docs/module.doc.yml index 16f7075..4e54e2b 100644 --- a/src/docs/module.doc.yml +++ b/src/docs/module.doc.yml @@ -18,23 +18,24 @@ paths: type: string enum: - hivemind + - violationDetection community: type: string format: objectId responses: - "201": + '201': description: Module created successfully. content: application/json: schema: - $ref: "#/components/schemas/Module" - "400": + $ref: '#/components/schemas/Module' + '400': description: Bad Request - $ref: "#/components/responses/BadRequest" - "401": + $ref: '#/components/responses/BadRequest' + '401': description: Unauthorized - $ref: "#/components/responses/Unauthorized" - + $ref: '#/components/responses/Unauthorized' + get: tags: - Module @@ -48,12 +49,13 @@ paths: type: string enum: - hivemind + - violationDetection - in: query name: community schema: type: string format: objectId - description: + description: required: true - in: query name: sortBy @@ -75,7 +77,7 @@ paths: default: 1 description: Page number responses: - "200": + '200': description: OK content: application/json: @@ -85,7 +87,7 @@ paths: results: type: array items: - $ref: "#/components/schemas/Module" + $ref: '#/components/schemas/Module' page: type: integer example: 1 @@ -98,9 +100,9 @@ paths: totalResults: type: integer example: 1 - "401": + '401': description: Unauthorized - $ref: "#/components/responses/Unauthorized" + $ref: '#/components/responses/Unauthorized' /api/v1/modules/{moduleId}: get: @@ -117,18 +119,18 @@ paths: schema: type: string responses: - "200": + '200': description: Module details retrieved successfully. content: application/json: schema: - $ref: "#/components/schemas/Module" - "401": + $ref: '#/components/schemas/Module' + '401': description: Unauthorized - $ref: "#/components/responses/Unauthorized" - "404": + $ref: '#/components/responses/Unauthorized' + '404': description: NotFound - $ref: "#/components/responses/NotFound" + $ref: '#/components/responses/NotFound' patch: tags: @@ -165,7 +167,7 @@ paths: name: type: string description: Name of the platform. - enum: ['discord', 'google','github','notion','mediaWiki'] + enum: ['discord', 'google', 'github', 'notion', 'mediaWiki', 'discourse'] metadata: type: object description: Metadata specific to the module and platform. Varies depending on the platform name and module name. @@ -174,7 +176,7 @@ paths: properties: answering: type: object - required: ["selectedChannels"] + required: ['selectedChannels'] properties: selectedChannels: type: array @@ -183,7 +185,7 @@ paths: description: Channels selected for answering queries. learning: type: object - required: ["selectedChannels", "fromDate"] + required: ['selectedChannels', 'fromDate'] properties: selectedChannels: type: array @@ -232,20 +234,37 @@ paths: items: type: string description: Metadata for the hivemind module on MediaWiki. + - type: object + properties: + selectedEmails: + type: array + items: + type: string + fromDate: + type: string + format: date. + toDate: + type: string + format: date. + selectedResources: + type: array + items: + type: integer + description: Metadata for the violation detection module on Discourse. minProperties: 1 responses: - "200": + '200': description: Module updated successfully. content: application/json: schema: - $ref: "#/components/schemas/Module" - "401": + $ref: '#/components/schemas/Module' + '401': description: Unauthorized - $ref: "#/components/responses/Unauthorized" - "404": + $ref: '#/components/responses/Unauthorized' + '404': description: NotFound - $ref: "#/components/responses/NotFound" + $ref: '#/components/responses/NotFound' delete: tags: @@ -261,11 +280,11 @@ paths: schema: type: string responses: - "204": + '204': description: Module deleted successfully. - "401": + '401': description: Unauthorized - $ref: "#/components/responses/Unauthorized" - "404": + $ref: '#/components/responses/Unauthorized' + '404': description: NotFound - $ref: "#/components/responses/NotFound" + $ref: '#/components/responses/NotFound' diff --git a/src/docs/platform.doc.yml b/src/docs/platform.doc.yml index 012e064..72bc71b 100644 --- a/src/docs/platform.doc.yml +++ b/src/docs/platform.doc.yml @@ -67,6 +67,7 @@ paths: - github - notion - mediaWiki + - discourse description: Name of the platform to create. Must be one of the supported platforms. community: type: string @@ -170,6 +171,21 @@ paths: type: string default: /w/api.php description: Metadata for MediaWiki. + - type: object + required: [id, period, analyzerStartedAt, resources] + properties: + id: + type: string + example: gov.optimism.io + period: + type: date + analyzerStartedAt: + type: date + resources: + type: array + items: + type: integer + description: Metadata for Discourse. responses: '201': description: Platform created successfully. From 06d93dc3940bcb01e24a2319c4c2fcfa053bb4c7 Mon Sep 17 00:00:00 2001 From: Behzad Rabiei Date: Wed, 18 Sep 2024 15:23:20 +0400 Subject: [PATCH 3/5] feat: worked on the airflow api call --- src/config/index.ts | 5 ++ src/constants/airflow.constant.ts | 8 +++ src/controllers/platform.controller.ts | 5 +- src/interfaces/Airflow.interface.ts | 9 +++ src/interfaces/index.ts | 1 + src/services/airflow.service.ts | 84 ++++++++++++++++++++++++++ src/services/index.ts | 3 +- src/types/airflow.type.ts | 3 + 8 files changed, 115 insertions(+), 3 deletions(-) create mode 100644 src/constants/airflow.constant.ts create mode 100644 src/interfaces/Airflow.interface.ts create mode 100644 src/services/airflow.service.ts create mode 100644 src/types/airflow.type.ts diff --git a/src/config/index.ts b/src/config/index.ts index 6bd75a2..699bae2 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -151,4 +151,9 @@ export default { session: { secret: envVars.SESSION_SECRET, }, + airflow: { + baseURL: envVars.AIRFLOW_BASE_URL, + username: envVars.AIRFLOW_USERNAME, + password: envVars.AIRFLOW_PASSWORD, + }, }; diff --git a/src/constants/airflow.constant.ts b/src/constants/airflow.constant.ts new file mode 100644 index 0000000..d184a43 --- /dev/null +++ b/src/constants/airflow.constant.ts @@ -0,0 +1,8 @@ +import { PlatformNames } from '@togethercrew.dev/db'; + +export const SUPPORTED_AIRFLOW_PLATFORMS = [PlatformNames.Discourse] as const; +export const AIRFLOW_PLATFORM_INFO = { + [PlatformNames.Discourse]: { + etl: 'discourse_analyzer_etl', + }, +}; diff --git a/src/controllers/platform.controller.ts b/src/controllers/platform.controller.ts index 80220ec..93cf928 100644 --- a/src/controllers/platform.controller.ts +++ b/src/controllers/platform.controller.ts @@ -9,6 +9,7 @@ import { githubService, notionService, discourseService, + airflowService, } from '../services'; import { IAuthRequest } from '../interfaces/Request.interface'; import { catchAsync, pick, ApiError } from '../utils'; @@ -18,14 +19,14 @@ import config from '../config'; import httpStatus from 'http-status'; import querystring from 'querystring'; import parentLogger from '../config/logger'; -import { PlatformNames } from '@togethercrew.dev/db'; -import { DatabaseManager } from '@togethercrew.dev/db'; +import { PlatformNames, DatabaseManager } from '@togethercrew.dev/db'; const logger = parentLogger.child({ module: 'PlatformController' }); const createPlatform = catchAsync(async function (req: IAuthRequest, res: Response) { const community = req.community; const platform = await platformService.managePlatformConnection(community?.id, req.body); + await airflowService.triggerDag(platform); res.status(httpStatus.CREATED).send(platform); }); diff --git a/src/interfaces/Airflow.interface.ts b/src/interfaces/Airflow.interface.ts new file mode 100644 index 0000000..d7f27b9 --- /dev/null +++ b/src/interfaces/Airflow.interface.ts @@ -0,0 +1,9 @@ +import { type ObjectId } from 'mongodb'; +import { type Snowflake } from 'discord.js'; + +export interface IDiscourseDagConfig { + platform_id: ObjectId; + id: Snowflake; + period: Date; + recompute: boolean; +} diff --git a/src/interfaces/index.ts b/src/interfaces/index.ts index ed53011..5f276a5 100644 --- a/src/interfaces/index.ts +++ b/src/interfaces/index.ts @@ -5,3 +5,4 @@ export * from './Channel.interface'; export * from './Guild.interface'; export * from './Twitter.interface'; export * from './Role.interface'; +export * from './Airflow.interface'; diff --git a/src/services/airflow.service.ts b/src/services/airflow.service.ts new file mode 100644 index 0000000..baf3603 --- /dev/null +++ b/src/services/airflow.service.ts @@ -0,0 +1,84 @@ +import fetch from 'node-fetch'; +import parentLogger from '../config/logger'; +import moment from 'moment'; +import { v4 as uuidv4 } from 'uuid'; +import config from '../config'; +import { IDiscourseDagConfig } from '../interfaces'; +import { ApiError } from '../utils'; +import httpStatus from 'http-status'; +import { AIRFLOW_PLATFORM_INFO, SUPPORTED_AIRFLOW_PLATFORMS } from '../constants/airflow.constant'; +import { SupportedAirflowPlatforms } from '../types/airflow.type'; +import { HydratedDocument } from 'mongoose'; +import { IPlatform, PlatformNames } from '@togethercrew.dev/db'; +const logger = parentLogger.child({ module: 'airflowService' }); + +function getParamsByPlatform(platform: HydratedDocument) { + switch (platform.name) { + case PlatformNames.Discourse: { + return { + platform_id: platform.id, + id: platform.metadata?.id, + period: platform.metadata?.period, + recompute: false, + }; + } + } +} +/** + * Triggers the DAG run in Apache Airflow. + * @param {IDiscourseDagConfig} params - The parameters object. + * @returns {Promise} - The response from the Airflow DAG run trigger. + */ +async function triggerDag(platform: HydratedDocument): Promise { + if (SUPPORTED_AIRFLOW_PLATFORMS.includes(platform.name as SupportedAirflowPlatforms)) { + const params = getParamsByPlatform(platform); + const dagRunId = uuidv4(); + const logicalDate = moment().add(1, 'minute').toISOString(); + const body = { + dag_run_id: dagRunId, + logical_date: logicalDate, + conf: { + ...params, + }, + note: 'compute', + }; + logger.info({ body }, 'Triggering DAG run'); + + try { + const auth = Buffer.from(`${config.airflow.username}:${config.airflow.password}`).toString('base64'); + const response = await fetch( + `${config.airflow.baseURL}/api/v1/dags/${AIRFLOW_PLATFORM_INFO[platform.name as SupportedAirflowPlatforms].etl}/dagRuns`, + { + method: 'POST', + body: JSON.stringify(body), + headers: { + 'Content-Type': 'application/json', + Authorization: `Basic ${auth}`, + }, + }, + ); + + if (!response.ok) { + const error = await response.json(); + logger.error({ error, body }, 'Failed to trigger DAG run'); + throw new Error(`Airflow API call failed with status ${response.status}: ${error.message}`); + } else { + const data = await response.json(); + logger.info({ data }, 'Successfully triggered DAG run'); + return data; + } + } catch (error) { + logger.error( + error, + `Failed to trigger DAG run for the ${AIRFLOW_PLATFORM_INFO[platform.name as SupportedAirflowPlatforms].etl}`, + ); + throw new ApiError(httpStatus.INTERNAL_SERVER_ERROR, 'Failed to trigger DAG run for the'); + } + } else { + return; + } +} + +export default { + triggerDag, +}; diff --git a/src/services/index.ts b/src/services/index.ts index 6f2c278..80ae5b0 100644 --- a/src/services/index.ts +++ b/src/services/index.ts @@ -14,7 +14,7 @@ import googleService from './google'; import githubService from './github'; import notionService from './notion'; import discourseService from './discourse'; - +import airflowService from './airflow.service'; export { userService, authService, @@ -32,4 +32,5 @@ export { githubService, notionService, discourseService, + airflowService, }; diff --git a/src/types/airflow.type.ts b/src/types/airflow.type.ts new file mode 100644 index 0000000..b99077d --- /dev/null +++ b/src/types/airflow.type.ts @@ -0,0 +1,3 @@ +import { SUPPORTED_AIRFLOW_PLATFORMS } from '../constants/airflow.constant'; + +export type SupportedAirflowPlatforms = (typeof SUPPORTED_AIRFLOW_PLATFORMS)[number]; From c84748692f77b459fe6adaa489f093a43c7aa1ab Mon Sep 17 00:00:00 2001 From: Behzad Rabiei Date: Wed, 18 Sep 2024 18:54:28 +0400 Subject: [PATCH 4/5] chore: disable the airflow api call --- src/controllers/platform.controller.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/controllers/platform.controller.ts b/src/controllers/platform.controller.ts index 93cf928..3d2e61c 100644 --- a/src/controllers/platform.controller.ts +++ b/src/controllers/platform.controller.ts @@ -26,7 +26,7 @@ const logger = parentLogger.child({ module: 'PlatformController' }); const createPlatform = catchAsync(async function (req: IAuthRequest, res: Response) { const community = req.community; const platform = await platformService.managePlatformConnection(community?.id, req.body); - await airflowService.triggerDag(platform); + // await airflowService.triggerDag(platform); res.status(httpStatus.CREATED).send(platform); }); From 47e105695aaabfe43b1bcaa37581de4acee4460e Mon Sep 17 00:00:00 2001 From: Behzad Rabiei Date: Wed, 18 Sep 2024 18:56:22 +0400 Subject: [PATCH 5/5] chore: disable the airflow api call --- src/config/index.ts | 5 -- src/controllers/platform.controller.ts | 1 - src/services/airflow.service.ts | 84 -------------------------- src/services/index.ts | 2 - 4 files changed, 92 deletions(-) delete mode 100644 src/services/airflow.service.ts diff --git a/src/config/index.ts b/src/config/index.ts index 5b5093a..f0706c7 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -154,9 +154,4 @@ export default { session: { secret: envVars.SESSION_SECRET, }, - airflow: { - baseURL: envVars.AIRFLOW_BASE_URL, - username: envVars.AIRFLOW_USERNAME, - password: envVars.AIRFLOW_PASSWORD, - }, }; diff --git a/src/controllers/platform.controller.ts b/src/controllers/platform.controller.ts index 3d2e61c..4b6d384 100644 --- a/src/controllers/platform.controller.ts +++ b/src/controllers/platform.controller.ts @@ -9,7 +9,6 @@ import { githubService, notionService, discourseService, - airflowService, } from '../services'; import { IAuthRequest } from '../interfaces/Request.interface'; import { catchAsync, pick, ApiError } from '../utils'; diff --git a/src/services/airflow.service.ts b/src/services/airflow.service.ts deleted file mode 100644 index baf3603..0000000 --- a/src/services/airflow.service.ts +++ /dev/null @@ -1,84 +0,0 @@ -import fetch from 'node-fetch'; -import parentLogger from '../config/logger'; -import moment from 'moment'; -import { v4 as uuidv4 } from 'uuid'; -import config from '../config'; -import { IDiscourseDagConfig } from '../interfaces'; -import { ApiError } from '../utils'; -import httpStatus from 'http-status'; -import { AIRFLOW_PLATFORM_INFO, SUPPORTED_AIRFLOW_PLATFORMS } from '../constants/airflow.constant'; -import { SupportedAirflowPlatforms } from '../types/airflow.type'; -import { HydratedDocument } from 'mongoose'; -import { IPlatform, PlatformNames } from '@togethercrew.dev/db'; -const logger = parentLogger.child({ module: 'airflowService' }); - -function getParamsByPlatform(platform: HydratedDocument) { - switch (platform.name) { - case PlatformNames.Discourse: { - return { - platform_id: platform.id, - id: platform.metadata?.id, - period: platform.metadata?.period, - recompute: false, - }; - } - } -} -/** - * Triggers the DAG run in Apache Airflow. - * @param {IDiscourseDagConfig} params - The parameters object. - * @returns {Promise} - The response from the Airflow DAG run trigger. - */ -async function triggerDag(platform: HydratedDocument): Promise { - if (SUPPORTED_AIRFLOW_PLATFORMS.includes(platform.name as SupportedAirflowPlatforms)) { - const params = getParamsByPlatform(platform); - const dagRunId = uuidv4(); - const logicalDate = moment().add(1, 'minute').toISOString(); - const body = { - dag_run_id: dagRunId, - logical_date: logicalDate, - conf: { - ...params, - }, - note: 'compute', - }; - logger.info({ body }, 'Triggering DAG run'); - - try { - const auth = Buffer.from(`${config.airflow.username}:${config.airflow.password}`).toString('base64'); - const response = await fetch( - `${config.airflow.baseURL}/api/v1/dags/${AIRFLOW_PLATFORM_INFO[platform.name as SupportedAirflowPlatforms].etl}/dagRuns`, - { - method: 'POST', - body: JSON.stringify(body), - headers: { - 'Content-Type': 'application/json', - Authorization: `Basic ${auth}`, - }, - }, - ); - - if (!response.ok) { - const error = await response.json(); - logger.error({ error, body }, 'Failed to trigger DAG run'); - throw new Error(`Airflow API call failed with status ${response.status}: ${error.message}`); - } else { - const data = await response.json(); - logger.info({ data }, 'Successfully triggered DAG run'); - return data; - } - } catch (error) { - logger.error( - error, - `Failed to trigger DAG run for the ${AIRFLOW_PLATFORM_INFO[platform.name as SupportedAirflowPlatforms].etl}`, - ); - throw new ApiError(httpStatus.INTERNAL_SERVER_ERROR, 'Failed to trigger DAG run for the'); - } - } else { - return; - } -} - -export default { - triggerDag, -}; diff --git a/src/services/index.ts b/src/services/index.ts index 80ae5b0..cfb20ce 100644 --- a/src/services/index.ts +++ b/src/services/index.ts @@ -14,7 +14,6 @@ import googleService from './google'; import githubService from './github'; import notionService from './notion'; import discourseService from './discourse'; -import airflowService from './airflow.service'; export { userService, authService, @@ -32,5 +31,4 @@ export { githubService, notionService, discourseService, - airflowService, };