refactor: use object to pass kafka message (#1230)

Author: trim21

bin/mq.ts

@@ -15,7 +15,7 @@ import {
   handleTopic as handleSubjectTopicEvent,
 } from '@app/event/subject';
 import { handle as handleTimelineEvent } from '@app/event/timeline';
-import type { Payload } from '@app/event/type';
+import type { KafkaMessage, Payload } from '@app/event/type';
 import { handle as handleUserEvent, handleFriend as handleFriendEvent } from '@app/event/user';
 import { newConsumer } from '@app/lib/kafka.ts';
 import { logger } from '@app/lib/logger';
@@ -43,7 +43,7 @@ const TOPICS = [
   'debezium.chii.bangumi.chii_subject_revisions',
 ];
 
-type Handler = (topic: string, key: string, value: string) => Promise<void>;
+type Handler = (msg: KafkaMessage) => Promise<void>;
 
 const binlogHandlers: Record<string, Handler | Handler[]> = {
   chii_blog_entry: handleBlogEvent,
@@ -63,8 +63,8 @@ const binlogHandlers: Record<string, Handler | Handler[]> = {
   chii_subject_revisions: handleSubjectDate,
 };
 
-async function onBinlogMessage(topic: string, key: string, value: string) {
-  const payload = JSON.parse(value) as Payload;
+async function onBinlogMessage(msg: KafkaMessage) {
+  const payload = JSON.parse(msg.value) as Payload;
   const handler = binlogHandlers[payload.source.table];
   if (!handler) {
     return;
@@ -74,14 +74,14 @@ async function onBinlogMessage(topic: string, key: string, value: string) {
     for (const h of handler) {
       // catch on each handler to it doesn't get swallowed by Promise.all.
       ts.push(
-        h(topic, key, value).catch((error) => {
+        h(msg).catch((error) => {
           logger.error('failed to handle event from %s: %o', payload.source.table, error);
         }),
       );
     }
     await Promise.all(ts);
   } else {
-    await handler(topic, key, value).catch((error) => {
+    await handler(msg).catch((error) => {
       logger.error('failed to handle event from %s: %o', payload.source.table, error);
     });
   }
@@ -91,13 +91,13 @@ const serviceHandlers: Record<string, Handler> = {
   timeline: handleTimelineMessage,
 };
 
-async function onServiceMessage(topic: string, key: string, value: string) {
-  const handler = serviceHandlers[topic];
+async function onServiceMessage(msg: KafkaMessage) {
+  const handler = serviceHandlers[msg.topic];
   if (!handler) {
     return;
   }
-  await handler(topic, key, value).catch((error) => {
-    logger.error('failed to handle event from %s: %o', topic, error);
+  await handler(msg).catch((error) => {
+    logger.error('failed to handle event from %s: %o', msg.topic, error);
   });
 }
 
@@ -119,9 +119,17 @@ async function main() {
       }
       try {
         if (topic.startsWith('debezium.')) {
-          await onBinlogMessage(topic, message.key.toString(), message.value.toString());
+          await onBinlogMessage({
+            topic: topic,
+            key: message.key.toString(),
+            value: message.value.toString(),
+          });
         } else {
-          await onServiceMessage(topic, message.key.toString(), message.value.toString());
+          await onServiceMessage({
+            topic: topic,
+            key: message.key.toString(),
+            value: message.value.toString(),
+          });
         }
       } catch (error) {
         logger.error(error, `error processing message ${message.key.toString()}`);

event/blog.ts

@@ -1,7 +1,7 @@
 import { getSlimCacheKey } from '@app/lib/blog/cache';
 import redis from '@app/lib/redis.ts';
 
-import { EventOp } from './type';
+import { EventOp, type KafkaMessage } from './type';
 
 interface BlogKey {
   entry_id: number;
@@ -11,7 +11,7 @@ interface Payload {
   op: EventOp;
 }
 
-export async function handle(topic: string, key: string, value: string) {
+export async function handle({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as BlogKey;
   const payload = JSON.parse(value) as Payload;
   switch (payload.op) {

event/character.ts

@@ -1,7 +1,7 @@
 import { getSlimCacheKey } from '@app/lib/character/cache';
 import redis from '@app/lib/redis.ts';
 
-import { EventOp } from './type';
+import { EventOp, type KafkaMessage } from './type';
 
 interface CharacterKey {
   crt_id: number;
@@ -11,7 +11,7 @@ interface Payload {
   op: EventOp;
 }
 
-export async function handle(topic: string, key: string, value: string) {
+export async function handle({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as CharacterKey;
   const payload = JSON.parse(value) as Payload;
   switch (payload.op) {

event/group.ts

@@ -2,7 +2,7 @@ import { getSlimCacheKey, getTopicCacheKey } from '@app/lib/group/cache.ts';
 import redis from '@app/lib/redis.ts';
 import { getJoinedGroupsCacheKey } from '@app/lib/user/cache.ts';
 
-import { EventOp } from './type';
+import { EventOp, type KafkaMessage } from './type';
 
 interface GroupKey {
   grp_id: number;
@@ -12,7 +12,7 @@ interface GroupPayload {
   op: EventOp;
 }
 
-export async function handle(topic: string, key: string, value: string) {
+export async function handle({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as GroupKey;
   const payload = JSON.parse(value) as GroupPayload;
   switch (payload.op) {
@@ -42,7 +42,7 @@ interface GroupMemberPayload {
   };
 }
 
-export async function handleMember(topic: string, _: string, value: string) {
+export async function handleMember({ value }: KafkaMessage) {
   const payload = JSON.parse(value) as GroupMemberPayload;
   const uid = payload.before?.gmb_uid ?? payload.after?.gmb_uid;
   if (!uid) {
@@ -70,7 +70,7 @@ interface GroupTopicPayload {
   op: EventOp;
 }
 
-export async function handleTopic(topic: string, key: string, value: string) {
+export async function handleTopic({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as GroupTopicKey;
   const payload = JSON.parse(value) as GroupTopicPayload;
   switch (payload.op) {

event/index.ts

@@ -1,7 +1,7 @@
 import { getSlimCacheKey } from '@app/lib/index/cache';
 import redis from '@app/lib/redis.ts';
 
-import { EventOp } from './type';
+import { EventOp, type KafkaMessage } from './type';
 
 interface IndexKey {
   idx_id: number;
@@ -11,7 +11,7 @@ interface Payload {
   op: EventOp;
 }
 
-export async function handle(topic: string, key: string, value: string) {
+export async function handle({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as IndexKey;
   const payload = JSON.parse(value) as Payload;
   switch (payload.op) {

event/person.ts

@@ -1,7 +1,7 @@
 import { getSlimCacheKey } from '@app/lib/person/cache';
 import redis from '@app/lib/redis.ts';
 
-import { EventOp } from './type';
+import { EventOp, type KafkaMessage } from './type';
 
 interface PersonKey {
   prsn_id: number;
@@ -11,7 +11,7 @@ interface Payload {
   op: EventOp;
 }
 
-export async function handle(topic: string, key: string, value: string) {
+export async function handle({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as PersonKey;
   const payload = JSON.parse(value) as Payload;
   switch (payload.op) {

event/subject.ts

@@ -11,7 +11,7 @@ import {
 import { extractDate } from '@app/lib/subject/date';
 import { sleep } from '@app/lib/utils';
 
-import { EventOp } from './type';
+import { EventOp, type KafkaMessage } from './type';
 
 interface SubjectKey {
   subject_id: number;
@@ -24,7 +24,7 @@ interface Payload {
   };
 }
 
-export async function handle(topic: string, key: string, value: string) {
+export async function handle({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as SubjectKey;
   const payload = JSON.parse(value) as Payload;
   switch (payload.op) {
@@ -46,7 +46,7 @@ interface FieldsKey {
   field_sid: number;
 }
 
-export async function handleFields(topic: string, key: string, value: string) {
+export async function handleFields({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as FieldsKey;
   const payload = JSON.parse(value) as Payload;
   switch (payload.op) {
@@ -68,7 +68,7 @@ interface TopicKey {
   sbj_tpc_id: number;
 }
 
-export async function handleTopic(topic: string, key: string, value: string) {
+export async function handleTopic({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as TopicKey;
   const payload = JSON.parse(value) as Payload;
   switch (payload.op) {
@@ -90,7 +90,7 @@ interface EpisodeKey {
   ep_id: number;
 }
 
-export async function handleEpisode(topic: string, key: string, value: string) {
+export async function handleEpisode({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as EpisodeKey;
   const payload = JSON.parse(value) as Payload;
   switch (payload.op) {
@@ -112,7 +112,7 @@ interface SubjectRevKey {
   rev_subject_id: number;
 }
 
-export async function handleSubjectDate(topic: string, key: string, value: string) {
+export async function handleSubjectDate({ topic, key, value }: KafkaMessage) {
   let subjectID: number;
   if (topic.endsWith('.chii_subject_revisions')) {
     const idx = JSON.parse(key) as SubjectRevKey;

event/timeline.ts

@@ -11,7 +11,7 @@ import {
 } from '@app/lib/timeline/cache';
 import { fetchFollowers } from '@app/lib/user/utils';
 
-import { EventOp } from './type';
+import { EventOp, type KafkaMessage } from './type';
 
 interface Key {
   tml_id: number;
@@ -29,7 +29,7 @@ interface Payload {
   after: TimelineItem | null;
 }
 
-export async function handle(topic: string, key: string, value: string) {
+export async function handle({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as Key;
   const payload = JSON.parse(value) as Payload;
 

event/type.ts

@@ -15,3 +15,9 @@ export interface Payload {
   };
   op: EventOp;
 }
+
+export interface KafkaMessage {
+  topic: string;
+  key: string;
+  value: string;
+}

event/user.ts

@@ -6,7 +6,7 @@ import {
   getSlimCacheKey,
 } from '@app/lib/user/cache';
 
-import { EventOp } from './type';
+import { EventOp, type KafkaMessage } from './type';
 
 interface UserPayload {
   op: EventOp;
@@ -16,7 +16,7 @@ interface UserKey {
   uid: number;
 }
 
-export async function handle(topic: string, key: string, value: string) {
+export async function handle({ key, value }: KafkaMessage) {
   const idx = JSON.parse(key) as UserKey;
   const payload = JSON.parse(value) as UserPayload;
   switch (payload.op) {
@@ -46,7 +46,7 @@ interface FriendPayload {
   };
 }
 
-export async function handleFriend(topic: string, _: string, value: string) {
+export async function handleFriend({ value }: KafkaMessage) {
   const payload = JSON.parse(value) as FriendPayload;
   const uid = payload.before?.frd_uid ?? payload.after?.frd_uid;
   const fid = payload.before?.frd_fid ?? payload.after?.frd_fid;

lib/timeline/kafka.ts

@@ -1,3 +1,4 @@
+import type { KafkaMessage } from '@app/event/type';
 import { logger } from '@app/lib/logger';
 
 import { type TimelineMessage, TimelineWriter } from './writer';
@@ -7,7 +8,7 @@ interface Payload {
   message: TimelineMessage[keyof TimelineMessage];
 }
 
-export async function handleTimelineMessage(topic: string, key: string, value: string) {
+export async function handleTimelineMessage({ value }: KafkaMessage) {
   const payload = JSON.parse(value) as Payload;
 
   switch (payload.op) {