7074262550

Committed 03 Dec 2023 02:56AM UTC coverage: 91.541% (-0.1%) from 91.672%

Build # 7074262550

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #376 from cofacts/multi-msg-context

Detect multiple messages sent at once

Run Details

488 of 572 branches covered (0.0%)

Branch coverage included in aggregate %.

26 of 36 new or added lines in 4 files covered. (72.22%)

1 existing line in 1 file now uncovered.

1027 of 1083 relevant lines covered (94.83%)

12.2 hits per line

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

75.42

/src/webhook/handlers/singleUserHandler.ts

import { t } from 'ttag';
import { WebhookEvent } from '@line/bot-sdk';

import { CooccurredMessage, PostbackActionData } from 'src/types/chatbotState';
import ga from 'src/lib/ga';
import redis from 'src/lib/redisClient';
import { extractArticleId, sleep } from 'src/lib/sharedUtils';
import lineClient from 'src/webhook/lineClient';
import UserSettings from 'src/database/models/userSettings';
import { Result } from 'src/types/result';

import handlePostback from './handlePostback';
import {
  TUTORIAL_STEPS,
  createGreetingMessage,
  createTutorialMessage,
} from './tutorial';
import processMedia from './processMedia';
import initState from './initState';
import { createTextMessage } from './utils';

const userIdBlacklist = (process.env.USERID_BLACKLIST || '').split(',');

// Set 58s timeout.
// Reply tokens must be used within one minute after receiving the webhook.
// Ref: https://developers.line.biz/en/reference/messaging-api/#send-reply-message
//
const REPLY_TIMEOUT = 58000;

/**
 * The time of messages stays in the batch.
 * The messages sent within this timeout are in the same co-occurrence.
 */
const BATCH_TIMEOUT = 500; // ms

// A symbol that is used to prevent accidental return in singleUserHandler.
// It should only be used when timeout are correctly handled.
//
const PROCESSED = Symbol('Processed in singleUserHandler');

const singleUserHandler = async (
  userId: string,
  webhookEvent: WebhookEvent
): Promise<typeof PROCESSED> => {
  if (userIdBlacklist.indexOf(userId) !== -1) {
    // User blacklist
    console.log(
      `[LOG] Blocked user INPUT =\n${JSON.stringify({
        userId,
        ...webhookEvent,
      })}\n`
    );
    return PROCESSED;
  }

  let isRepliedDueToTimeout = false;

  // Tell the user the bot is busy if processing does not end within timeout
  //
  const timerId = setTimeout(async function () {
    console.log(
      `[LOG] Timeout ${JSON.stringify({
        userId,
        ...webhookEvent,
      })}\n`
    );
    await send({
      context,
      replies: [
        {
          type: 'text',
          text: t`Line bot is busy, or we cannot handle this message. Maybe you can try again a few minutes later.`,
        },
      ],
    });

    isRepliedDueToTimeout = true;
  }, REPLY_TIMEOUT);

  // Get user's context from redis or create a new one
  //
  const context = (await redis.get(userId)) || {
    data: { sessionId: Date.now() },
  };

  const REDIS_BATCH_KEY = getRedisBatchKey(userId);

  /**
   * @param msg
   * @returns if the specified CooccurredMsg is the last one in the current batch of CooccurredMsgs.
   */
  async function isLastInBatch(msg: CooccurredMessage) {
    const lastMsgInBatch: CooccurredMessage | undefined = (
      await redis.range(REDIS_BATCH_KEY, -1, -1)
    )[0];
    return !!lastMsgInBatch && msg.id === lastMsgInBatch.id;
  }

  // Helper functions in singleUserHandler that indicates the end of processing.
  // If `forMsg` is provided, also check if the message is the latest in batch.
  //
  async function send(
    result: Result,

    /**
     * The msg that this result is for.
     * If provided, exercise extra check ensure `result` is up-to-date before sending replies.
     * */
    forMsg?: CooccurredMessage
  ): Promise<typeof PROCESSED> {
    if (isRepliedDueToTimeout) {
      console.log('[LOG] reply & context setup aborted');
      return cancel();
    }

    // Check forMsg only when it is provided
    if (forMsg !== undefined && !(await isLastInBatch(forMsg))) {
      // The batch has new messages inside, thus the result is outdated and should be abandoned.
      // Leave the rest to the processor of the last msg in batch.
      //
      return cancel();
    }

    // We are sending reply, stop timer countdown
    clearTimeout(timerId);

    // The chatbot's reply cuts off the user's input streak, thus we end the current batch here.
    redis.del(REDIS_BATCH_KEY);

    console.log(
      JSON.stringify({
        CONTEXT: result.context,
        INPUT: { userId, ...webhookEvent },
        OUTPUT: result,
      })
    );

    // Send replies. Does not need to wait for lineClient's callbacks.
    // lineClient's callback does error handling by itself.
    //
    if ('replyToken' in webhookEvent) {
      lineClient.post('/message/reply', {
        replyToken: webhookEvent.replyToken,
        messages: result.replies,
      });
    }

    // Set context
    //
    await redis.set(userId, result.context);
    return PROCESSED;
  }

  // Does not reply and just exit processing.
  //
  function cancel(): typeof PROCESSED {
    clearTimeout(timerId); // Avoid timeout after we exit
    return PROCESSED;
  }

  /**
   * Adds cooccurred message to batch.
   * After BATCH_TIMEOUT since the last message has been added, initiate the processing of these
   * co-occurred messages.
   */
  async function addMsgToBatch(
    msg: CooccurredMessage
  ): Promise<typeof PROCESSED> {
    await redis.push(REDIS_BATCH_KEY, msg);

    await sleep(BATCH_TIMEOUT);

    if (!(await isLastInBatch(msg))) {
      // New message appears during we sleep,
      // abort processing and let the new message's callback do the work.
      return cancel();
    }

    // Try process the batch and calculate results
    const messages: CooccurredMessage[] = await redis.range(
      REDIS_BATCH_KEY,
      0,
      -1
    );

    if (messages.length !== 1) {
      // TODO: initiate multi-message processing here
      //
      await sleep(1000); // Simulate multi-message processing and see if more message in batch.
      return send(
        {
          context,
          replies: [
            createTextMessage({
              text: `目前我還沒辦法一次處理 ${messages.length} 則訊息，請一則一則傳進來唷！`,
            }),
          ],
        },
        msg
      );
    }

    const firstMsg = messages[0];
    if (firstMsg.type !== 'text') {
      return send(
        await processMedia(
          {
            message: {
              id: firstMsg.id,
              type: firstMsg.type,
            },
          },
          userId
        ),
        msg
      );
    }
    const result = await initState({
      data: {
        // Create a new "search session".
        // Used to determine button postbacks and GraphQL requests are from
        // previous sessions
        //
        sessionId: Date.now(),

        // Store user input into context
        searchedText: firstMsg.text,
      },
      userId,
    });

    return send(
      {
        context: { data: result.data },
        replies: result.replies,
      },
      msg
    );
  }

  switch (webhookEvent.type) {
    default: {
      // These events are not handled at all.
      return cancel();
    }

    case 'unfollow': {
      await UserSettings.setAllowNewReplyUpdate(userId, false);
      return cancel();
    }

    case 'follow': {
      await UserSettings.setAllowNewReplyUpdate(userId, true);

      // Create new context
      const data = { sessionId: Date.now() };

      const visitor = ga(userId, 'TUTORIAL');
      visitor.event({
        ec: 'Tutorial',
        ea: 'Step',
        el: 'ON_BOARDING',
      });
      visitor.send();

      return send({
        context: { data },
        replies: [
          createGreetingMessage(),
          createTutorialMessage(data.sessionId),
        ],
      });
    }

    case 'postback': {
      const postbackData = JSON.parse(
        webhookEvent.postback.data
      ) as PostbackActionData<unknown>;

      if (postbackData.sessionId === context.data.sessionId) {
        return send(await handlePostback(context.data, postbackData, userId));
      }

      // Postback data session ID != context session ID can happen when
      // (1) user context in redis is expired, or
      // (2) if other new messages have been sent before pressing buttons.
      //
      // Under these scenarios, tell the user about the expiry of buttons
      //
      console.log('Previous button pressed.');

      clearTimeout(timerId);
      return send({
        context, // Reuse existing context
        replies: [
          {
            type: 'text',
            text:
              '🚧 ' +
              t`You are currently searching for another message, buttons from previous search sessions do not work now.`,
          },
        ],
      });
    }

    case 'message': {
      break; // Handle message events later
    }
  }

  // We have message events left.
  //
  switch (webhookEvent.message.type) {
    default: {
      // Track other message type send by user
      ga(userId)
        .event({
          ec: 'UserInput',
          ea: 'MessageType',
          el: webhookEvent.message.type,
        })
        .send();
      return cancel();
    }

    case 'audio':
    case 'video':
    case 'image':
      return addMsgToBatch({
        type: webhookEvent.message.type,
        id: webhookEvent.message.id,
      });

    case 'text': {
      // Handle text events later
      break;
    }
  }

  // Handle text event messages
  //
  switch (webhookEvent.message.text) {
    // Debugging: type 'RESET' to reset user's context and start all over.
    //
    case 'RESET': {
      redis.del(userId);
      redis.del(REDIS_BATCH_KEY);
      return cancel();
    }

    case TUTORIAL_STEPS['RICH_MENU']: {
      // Start new session, reroute to TUTORIAL
      const sessionId = Date.now();
      return send(
        await handlePostback(
          { sessionId, searchedText: '' },
          {
            state: 'TUTORIAL',
            sessionId,
            input: TUTORIAL_STEPS['RICH_MENU'],
          },
          userId
        )
      );
    }

    default: {
      const trimmedInput = webhookEvent.message.text.trim();
      const articleId = extractArticleId(trimmedInput);

      if (articleId) {
        // It is a predefined text message wanting us to visit a article.
        // Start new session, reroute to CHOOSING_ARTILCE and simulate "choose article" postback event
        const sessionId = Date.now();
        return send(
          await handlePostback(
            // Start a new session
            {
              sessionId,
              searchedText: '',
            },
            {
              state: 'CHOOSING_ARTICLE',
              sessionId,
              input: articleId,
            },
            userId
          )
        );
      }

      // The user forwarded us an new message.
      //
      return addMsgToBatch({
        id: webhookEvent.message.id,
        type: 'text',
        text: trimmedInput,
      });
    }
  }
};

export function getRedisBatchKey(userId: string) {
  return `${userId}:batch`;
}

export default singleUserHandler;

1	import { t } from 'ttag';
2	import { WebhookEvent } from '@line/bot-sdk';
3
4	import { CooccurredMessage, PostbackActionData } from 'src/types/chatbotState';
5	import ga from 'src/lib/ga';
6	import redis from 'src/lib/redisClient';
7	import { extractArticleId, sleep } from 'src/lib/sharedUtils';
8	import lineClient from 'src/webhook/lineClient';
9	import UserSettings from 'src/database/models/userSettings';
10	import { Result } from 'src/types/result';
11
12	import handlePostback from './handlePostback';
13	import {
14	TUTORIAL_STEPS,
15	createGreetingMessage,
16	createTutorialMessage,
17	} from './tutorial';
18	import processMedia from './processMedia';
19	import initState from './initState';
20	import { createTextMessage } from './utils';
21
22	const userIdBlacklist = (process.env.USERID_BLACKLIST \|\| '').split(',');	2✔
23
24	// Set 58s timeout.
25	// Reply tokens must be used within one minute after receiving the webhook.
26	// Ref: https://developers.line.biz/en/reference/messaging-api/#send-reply-message
27	//
28	const REPLY_TIMEOUT = 58000;	2✔
29
30	/**
31	* The time of messages stays in the batch.
32	* The messages sent within this timeout are in the same co-occurrence.
33	*/
34	const BATCH_TIMEOUT = 500; // ms	2✔
35
36	// A symbol that is used to prevent accidental return in singleUserHandler.
37	// It should only be used when timeout are correctly handled.
38	//
39	const PROCESSED = Symbol('Processed in singleUserHandler');	2✔
40
41	const singleUserHandler = async (	2✔
42	userId: string,
43	webhookEvent: WebhookEvent
44	): Promise<typeof PROCESSED> => {
45	if (userIdBlacklist.indexOf(userId) !== -1) {	9!
46	// User blacklist
47	console.log(	×
48	`[LOG] Blocked user INPUT =\n${JSON.stringify({
49	userId,
50	...webhookEvent,
51	})}\n`
52	);
53	return PROCESSED;	×
54	}
55
56	let isRepliedDueToTimeout = false;	9✔
57
58	// Tell the user the bot is busy if processing does not end within timeout
59	//
60	const timerId = setTimeout(async function () {	9✔
61	console.log(	×
62	`[LOG] Timeout ${JSON.stringify({
63	userId,
64	...webhookEvent,
65	})}\n`
66	);
NEW 67	await send({	×
68	context,
69	replies: [
70	{
71	type: 'text',
72	text: t`Line bot is busy, or we cannot handle this message. Maybe you can try again a few minutes later.`,
73	},
74	],
75	});
76
NEW 77	isRepliedDueToTimeout = true;	×
78	}, REPLY_TIMEOUT);
79
80	// Get user's context from redis or create a new one
81	//
82	const context = (await redis.get(userId)) \|\| {	9✔
83	data: { sessionId: Date.now() },
84	};
85
86	const REDIS_BATCH_KEY = getRedisBatchKey(userId);	9✔
87
88	/**
89	* @param msg
90	* @returns if the specified CooccurredMsg is the last one in the current batch of CooccurredMsgs.
91	*/
92	async function isLastInBatch(msg: CooccurredMessage) {
93	const lastMsgInBatch: CooccurredMessage \| undefined = (	2✔
94	await redis.range(REDIS_BATCH_KEY, -1, -1)
95	)[0];
96	return !!lastMsgInBatch && msg.id === lastMsgInBatch.id;	2✔
97	}
98
99	// Helper functions in singleUserHandler that indicates the end of processing.
100	// If `forMsg` is provided, also check if the message is the latest in batch.
101	//
102	async function send(
103	result: Result,
104
105	/**
106	* The msg that this result is for.
107	* If provided, exercise extra check ensure `result` is up-to-date before sending replies.
108	* */
109	forMsg?: CooccurredMessage
110	): Promise<typeof PROCESSED> {
111	if (isRepliedDueToTimeout) {	7!
112	console.log('[LOG] reply & context setup aborted');	×
NEW 113	return cancel();	×
114	}
115
116	// Check forMsg only when it is provided
117	if (forMsg !== undefined && !(await isLastInBatch(forMsg))) {	7!
118	// The batch has new messages inside, thus the result is outdated and should be abandoned.
119	// Leave the rest to the processor of the last msg in batch.
120	//
NEW 121	return cancel();	×
122	}
123
124	// We are sending reply, stop timer countdown
125	clearTimeout(timerId);	7✔
126
127	// The chatbot's reply cuts off the user's input streak, thus we end the current batch here.
128	redis.del(REDIS_BATCH_KEY);	7✔
129
130	console.log(	7✔
131	JSON.stringify({
132	CONTEXT: result.context,
133	INPUT: { userId, ...webhookEvent },
134	OUTPUT: result,
135	})
136	);
137
138	// Send replies. Does not need to wait for lineClient's callbacks.
139	// lineClient's callback does error handling by itself.
140	//
141	if ('replyToken' in webhookEvent) {	7!
142	lineClient.post('/message/reply', {	7✔
143	replyToken: webhookEvent.replyToken,
144	messages: result.replies,
145	});
146	}
147
148	// Set context
149	//
150	await redis.set(userId, result.context);	7✔
151	return PROCESSED;	7✔
152	}
153
154	// Does not reply and just exit processing.
155	//
156	function cancel(): typeof PROCESSED {
157	clearTimeout(timerId); // Avoid timeout after we exit	2✔
158	return PROCESSED;	2✔
159	}
160
161	/**
162	* Adds cooccurred message to batch.
163	* After BATCH_TIMEOUT since the last message has been added, initiate the processing of these
164	* co-occurred messages.
165	*/
166	async function addMsgToBatch(
167	msg: CooccurredMessage
168	): Promise<typeof PROCESSED> {
169	await redis.push(REDIS_BATCH_KEY, msg);	1✔
170
171	await sleep(BATCH_TIMEOUT);	1✔
172
173	if (!(await isLastInBatch(msg))) {	1!
174	// New message appears during we sleep,
175	// abort processing and let the new message's callback do the work.
NEW 176	return cancel();	×
177	}
178
179	// Try process the batch and calculate results
180	const messages: CooccurredMessage[] = await redis.range(	1✔
181	REDIS_BATCH_KEY,
182	0,
183	-1
184	);
185
186	if (messages.length !== 1) {	1!
187	// TODO: initiate multi-message processing here
188	//
NEW 189	await sleep(1000); // Simulate multi-message processing and see if more message in batch.	×
NEW 190	return send(	×
191	{
192	context,
193	replies: [
194	createTextMessage({
195	text: `目前我還沒辦法一次處理 ${messages.length} 則訊息，請一則一則傳進來唷！`,
196	}),
197	],
198	},
199	msg
200	);
201	}
202
203	const firstMsg = messages[0];	1✔
204	if (firstMsg.type !== 'text') {	1!
NEW 205	return send(	×
206	await processMedia(
207	{
208	message: {
209	id: firstMsg.id,
210	type: firstMsg.type,
211	},
212	},
213	userId
214	),
215	msg
216	);
217	}
218	const result = await initState({	1✔
219	data: {
220	// Create a new "search session".
221	// Used to determine button postbacks and GraphQL requests are from
222	// previous sessions
223	//
224	sessionId: Date.now(),
225
226	// Store user input into context
227	searchedText: firstMsg.text,
228	},
229	userId,
230	});
231
232	return send(	1✔
233	{
234	context: { data: result.data },
235	replies: result.replies,
236	},
237	msg
238	);
239	}
240
241	switch (webhookEvent.type) {	9!
242	default: {
243	// These events are not handled at all.
244	return cancel();	×
245	}
246
247	case 'unfollow': {
248	await UserSettings.setAllowNewReplyUpdate(userId, false);	1✔
249	return cancel();	1✔
250	}
251
252	case 'follow': {
253	await UserSettings.setAllowNewReplyUpdate(userId, true);	1✔
254
255	// Create new context
256	const data = { sessionId: Date.now() };	1✔
257
258	const visitor = ga(userId, 'TUTORIAL');	1✔
259	visitor.event({	1✔
260	ec: 'Tutorial',
261	ea: 'Step',
262	el: 'ON_BOARDING',
263	});
264	visitor.send();	1✔
265
266	return send({	1✔
267	context: { data },
268	replies: [
269	createGreetingMessage(),
270	createTutorialMessage(data.sessionId),
271	],
272	});
273	}
274
275	case 'postback': {
276	const postbackData = JSON.parse(	2✔
277	webhookEvent.postback.data
278	) as PostbackActionData<unknown>;
279
280	if (postbackData.sessionId === context.data.sessionId) {	2✔
281	return send(await handlePostback(context.data, postbackData, userId));	1✔
282	}
283
284	// Postback data session ID != context session ID can happen when
285	// (1) user context in redis is expired, or
286	// (2) if other new messages have been sent before pressing buttons.
287	//
288	// Under these scenarios, tell the user about the expiry of buttons
289	//
290	console.log('Previous button pressed.');	1✔
291
292	clearTimeout(timerId);	1✔
293	return send({	1✔
294	context, // Reuse existing context
295	replies: [
296	{
297	type: 'text',
298	text:
299	'🚧 ' +
300	t`You are currently searching for another message, buttons from previous search sessions do not work now.`,
301	},
302	],
303	});
304	}
305
306	case 'message': {
307	break; // Handle message events later	5✔
308	}
309	}
310
311	// We have message events left.
312	//
313	switch (webhookEvent.message.type) {	5!
314	default: {
315	// Track other message type send by user
316	ga(userId)	1✔
317	.event({
318	ec: 'UserInput',
319	ea: 'MessageType',
320	el: webhookEvent.message.type,
321	})
322	.send();
323	return cancel();	1✔
324	}
325
326	case 'audio':
327	case 'video':
328	case 'image':
NEW 329	return addMsgToBatch({	×
330	type: webhookEvent.message.type,
331	id: webhookEvent.message.id,
332	});
333
334	case 'text': {
335	// Handle text events later
336	break;	4✔
337	}
338	}
339
340	// Handle text event messages
341	//
342	switch (webhookEvent.message.text) {	4!
343	// Debugging: type 'RESET' to reset user's context and start all over.
344	//
345	case 'RESET': {
346	redis.del(userId);	×
NEW 347	redis.del(REDIS_BATCH_KEY);	×
UNCOV 348	return cancel();	×
349	}
350
351	case TUTORIAL_STEPS['RICH_MENU']: {
352	// Start new session, reroute to TUTORIAL
353	const sessionId = Date.now();	1✔
354	return send(	1✔
355	await handlePostback(
356	{ sessionId, searchedText: '' },
357	{
358	state: 'TUTORIAL',
359	sessionId,
360	input: TUTORIAL_STEPS['RICH_MENU'],
361	},
362	userId
363	)
364	);
365	}
366
367	default: {
368	const trimmedInput = webhookEvent.message.text.trim();	3✔
369	const articleId = extractArticleId(trimmedInput);	3✔
370
371	if (articleId) {	3✔
372	// It is a predefined text message wanting us to visit a article.
373	// Start new session, reroute to CHOOSING_ARTILCE and simulate "choose article" postback event
374	const sessionId = Date.now();	2✔
375	return send(	2✔
376	await handlePostback(
377	// Start a new session
378	{
379	sessionId,
380	searchedText: '',
381	},
382	{
383	state: 'CHOOSING_ARTICLE',
384	sessionId,
385	input: articleId,
386	},
387	userId
388	)
389	);
390	}
391
392	// The user forwarded us an new message.
393	//
394	return addMsgToBatch({	1✔
395	id: webhookEvent.message.id,
396	type: 'text',
397	text: trimmedInput,
398	});
399	}
400	}
401	};
402
403	export function getRedisBatchKey(userId: string) {
404	return `${userId}:batch`;	10✔
405	}
406
407	export default singleUserHandler;

cofacts / rumors-line-bot / 7074262550

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