10921662451

Committed 18 Sep 2024 12:01PM UTC coverage: 13.799% (-0.03%) from 13.825%

Build # 10921662451

Build Type

Pull #9226

github

Committed by

web-flow

Commit Message

Merge b6e66e882 into 910cd3883

Pull Request Pull Request #9226: Multi-Adapter Ports

Run Details

3042 of 20466 branches covered (14.86%)

Branch coverage included in aggregate %.

12 of 79 new or added lines in 12 files covered. (15.19%)

73 existing lines in 11 files now uncovered.

3977 of 30401 relevant lines covered (13.08%)

3.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/packages/discord-bot/src/discord-consumer/discordConsumer.ts

import {
  HotShotsStats,
  RabbitMQAdapter,
  RascalConfigServices,
  ServiceKey,
  getRabbitMQConfig,
  startHealthCheckLoop,
} from '@hicommonwealth/adapters';
import {
  Broker,
  BrokerSubscriptions,
  EventHandler,
  Policy,
  broker,
  events,
  logger,
  stats,
} from '@hicommonwealth/core';
import {
  CommentDiscordActions,
  IDiscordMessage,
  ThreadDiscordActions,
} from '@hicommonwealth/model';
import { DISCORD_BOT_ADDRESS } from '@hicommonwealth/shared';
import v8 from 'v8';
import { ZodUndefined } from 'zod';
import { config } from '../config';

const log = logger(import.meta);
stats({
  adapter: HotShotsStats(),
});

let isServiceHealthy = false;

startHealthCheckLoop({
  service: ServiceKey.DiscordBotConsumer,
  checkFn: async () => {
    if (!isServiceHealthy) {
      throw new Error('service not healthy');
    }
  },
});

log.info(
  `Node Option max-old-space-size set to: ${JSON.stringify(
    v8.getHeapStatistics().heap_size_limit / 1000000000,
  )} GB`,
);

const processDiscordMessageCreated: EventHandler<
  'DiscordMessageCreated',
  ZodUndefined
> = async ({ payload }) => {
  // async imports to delay calling logger
  const { handleCommentMessages, handleThreadMessages } = await import(
    '../discord-consumer/handlers'
  );
  const { getForumLinkedTopic } = await import('../util');

  try {
    const parsedMessage = payload as IDiscordMessage;
    const topic = await getForumLinkedTopic(parsedMessage.parent_channel_id);
    const action = parsedMessage.action;
    const sharedReqData = {
      auth: config.DISCORD.CW_BOT_KEY,
      discord_meta: {
        message_id: parsedMessage.message_id,
        channel_id: parsedMessage.parent_channel_id,
        user: parsedMessage.user,
      },
      author_chain: topic.community_id,
      address: DISCORD_BOT_ADDRESS,
      chain: topic.community_id,
    };

    if (
      [
        'thread-delete',
        'thread-title-update',
        'thread-body-update',
        'thread-create',
      ].includes(action)
    ) {
      await handleThreadMessages(
        action as ThreadDiscordActions,
        parsedMessage,
        topic,
        sharedReqData,
      );
    } else {
      await handleCommentMessages(
        action as CommentDiscordActions,
        parsedMessage,
        sharedReqData,
      );
    }

    stats().increment('cw.discobot_message_processed', {
      chain: topic.community_id,
      action: action,
    });
  } catch (error) {
    // non 2XX response
    if (error.response) {
      const msg =
        'Axios Error - Failed to process message:' +
        `\n\tStatus: ${error.response.status}` +
        `\n\tData: ${JSON.stringify(error.response.data)}`;
      log.error(msg, new Error(error.response.data.error));
    } else {
      log.error(`Failed to process Message:`, error);
    }
  }
};

async function main() {
  config.APP_ENV === 'local' && console.log(config);

  let brokerInstance: Broker;
  try {
    const rmqAdapter = new RabbitMQAdapter(
      getRabbitMQConfig(
        config.BROKER.RABBITMQ_URI,
        RascalConfigServices.DiscobotService,
      ),
    );
    await rmqAdapter.init();
    broker({
      adapter: rmqAdapter,
    });
    brokerInstance = rmqAdapter;
  } catch (e) {
    log.error('Broker setup failed', e);
    throw e;
  }

  const inputs = {
    DiscordMessageCreated: events.DiscordMessageCreated,
  };

  function Discord(): Policy<typeof inputs> {
    return {
      inputs,
      body: {
        DiscordMessageCreated: processDiscordMessageCreated,
      },
    };
  }

  const result = await brokerInstance.subscribe(
    BrokerSubscriptions.DiscordListener,
    Discord(),
  );

  if (!result) {
    throw new Error(
      `Failed to subscribe to ${BrokerSubscriptions.DiscordListener}`,
    );
  }

  isServiceHealthy = true;
}

main().catch((err) => {
  log.fatal(err);
});

1	import {
2	HotShotsStats,
3	RabbitMQAdapter,
4	RascalConfigServices,
5	ServiceKey,
6	getRabbitMQConfig,
7	startHealthCheckLoop,
8	} from '@hicommonwealth/adapters';
9	import {
10	Broker,
11	BrokerSubscriptions,
12	EventHandler,
13	Policy,
14	broker,
15	events,
16	logger,
17	stats,
18	} from '@hicommonwealth/core';
19	import {
20	CommentDiscordActions,
21	IDiscordMessage,
22	ThreadDiscordActions,
23	} from '@hicommonwealth/model';
24	import { DISCORD_BOT_ADDRESS } from '@hicommonwealth/shared';
25	import v8 from 'v8';
26	import { ZodUndefined } from 'zod';
27	import { config } from '../config';
28
29	const log = logger(import.meta);	×
NEW 30	stats({	×
31	adapter: HotShotsStats(),
32	});
33
34	let isServiceHealthy = false;	×
35
36	startHealthCheckLoop({	×
37	service: ServiceKey.DiscordBotConsumer,
38	checkFn: async () => {
39	if (!isServiceHealthy) {	×
40	throw new Error('service not healthy');	×
41	}
42	},
43	});
44
45	log.info(	×
46	`Node Option max-old-space-size set to: ${JSON.stringify(
47	v8.getHeapStatistics().heap_size_limit / 1000000000,
48	)} GB`,
49	);
50
51	const processDiscordMessageCreated: EventHandler<
52	'DiscordMessageCreated',
53	ZodUndefined
54	> = async ({ payload }) => {	×
55	// async imports to delay calling logger
56	const { handleCommentMessages, handleThreadMessages } = await import(	×
57	'../discord-consumer/handlers'
58	);
59	const { getForumLinkedTopic } = await import('../util');	×
60
61	try {	×
62	const parsedMessage = payload as IDiscordMessage;	×
63	const topic = await getForumLinkedTopic(parsedMessage.parent_channel_id);	×
64	const action = parsedMessage.action;	×
65	const sharedReqData = {	×
66	auth: config.DISCORD.CW_BOT_KEY,
67	discord_meta: {
68	message_id: parsedMessage.message_id,
69	channel_id: parsedMessage.parent_channel_id,
70	user: parsedMessage.user,
71	},
72	author_chain: topic.community_id,
73	address: DISCORD_BOT_ADDRESS,
74	chain: topic.community_id,
75	};
76
77	if (	×
78	[
79	'thread-delete',
80	'thread-title-update',
81	'thread-body-update',
82	'thread-create',
83	].includes(action)
84	) {
85	await handleThreadMessages(	×
86	action as ThreadDiscordActions,
87	parsedMessage,
88	topic,
89	sharedReqData,
90	);
91	} else {
92	await handleCommentMessages(	×
93	action as CommentDiscordActions,
94	parsedMessage,
95	sharedReqData,
96	);
97	}
98
99	stats().increment('cw.discobot_message_processed', {	×
100	chain: topic.community_id,
101	action: action,
102	});
103	} catch (error) {
104	// non 2XX response
105	if (error.response) {	×
106	const msg =
107	'Axios Error - Failed to process message:' +	×
108	`\n\tStatus: ${error.response.status}` +
109	`\n\tData: ${JSON.stringify(error.response.data)}`;
110	log.error(msg, new Error(error.response.data.error));	×
111	} else {
112	log.error(`Failed to process Message:`, error);	×
113	}
114	}
115	};
116
117	async function main() {
118	config.APP_ENV === 'local' && console.log(config);	×
119
120	let brokerInstance: Broker;
121	try {	×
122	const rmqAdapter = new RabbitMQAdapter(	×
123	getRabbitMQConfig(
124	config.BROKER.RABBITMQ_URI,
125	RascalConfigServices.DiscobotService,
126	),
127	);
128	await rmqAdapter.init();	×
NEW 129	broker({	×
130	adapter: rmqAdapter,
131	});
UNCOV 132	brokerInstance = rmqAdapter;	×
133	} catch (e) {
134	log.error('Broker setup failed', e);	×
135	throw e;	×
136	}
137
138	const inputs = {	×
139	DiscordMessageCreated: events.DiscordMessageCreated,
140	};
141
142	function Discord(): Policy<typeof inputs> {
143	return {	×
144	inputs,
145	body: {
146	DiscordMessageCreated: processDiscordMessageCreated,
147	},
148	};
149	}
150
151	const result = await brokerInstance.subscribe(	×
152	BrokerSubscriptions.DiscordListener,
153	Discord(),
154	);
155
156	if (!result) {	×
157	throw new Error(	×
158	`Failed to subscribe to ${BrokerSubscriptions.DiscordListener}`,
159	);
160	}
161
162	isServiceHealthy = true;	×
163	}
164
165	main().catch((err) => {	×
166	log.fatal(err);	×
167	});

hicommonwealth / commonwealth / 10921662451

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous