feat: trigger another worker (#547)

worker/src/admin_api/worker_config.ts

@@ -1,7 +1,7 @@
 import { Context } from 'hono';
 
 import { HonoCustomType } from '../types';
-import { getAdminPasswords, getBooleanValue, getDefaultDomains, getDomains, getIntValue, getPasswords, getStringArray, getStringValue, getUserRoles } from '../utils';
+import { getAdminPasswords, getBooleanValue, getDefaultDomains, getDomains, getIntValue, getPasswords, getStringArray, getStringValue, getUserRoles, getAnotherWorkerList } from '../utils';
 import { CONSTANTS } from '../constants';
 import { isS3Enabled } from '../mails_api/s3_attachment';
 
@@ -44,6 +44,9 @@ export default {
             "DISABLE_ADMIN_PASSWORD_CHECK": getBooleanValue(c.env.DISABLE_ADMIN_PASSWORD_CHECK),
             "ENABLE_CHECK_JUNK_MAIL": getBooleanValue(c.env.ENABLE_CHECK_JUNK_MAIL),
             "JUNK_MAIL_FORCE_PASS_LIST": getStringArray(c.env.JUNK_MAIL_FORCE_PASS_LIST),
+
+            "ENABLE_ANOTHER_WORKER": getBooleanValue(c.env.ENABLE_ANOTHER_WORKER),
+            "ANOTHER_WORKER_LIST": getAnotherWorkerList(c),
         })
     }
 }

worker/src/common.ts

@@ -1,8 +1,8 @@
 import { Context } from 'hono';
 import { Jwt } from 'hono/utils/jwt'
 
-import { getBooleanValue, getDomains, getStringValue, getIntValue, getUserRoles, getDefaultDomains, getJsonSetting } from './utils';
-import { HonoCustomType, UserRole } from './types';
+import { getBooleanValue, getDomains, getStringValue, getIntValue, getUserRoles, getDefaultDomains, getJsonSetting, getAnotherWorkerList } from './utils';
+import { HonoCustomType, UserRole, AnotherWorker, RPCEmailMessage } from './types';
 import { unbindTelegramByAddress } from './telegram_api/common';
 import { CONSTANTS } from './constants';
 import { AdminWebhookSettings, WebhookMail, WebhookSettings } from './models';
@@ -360,7 +360,7 @@ export async function triggerWebhook(
     address: string,
     raw_mail: string,
     message_id: string | null
-): Promise<void> {
+): Promise<string | undefined | null> {
     if (!c.env.KV || !getBooleanValue(c.env.ENABLE_WEBHOOK)) {
         return
     }
@@ -408,4 +408,49 @@ export async function triggerWebhook(
             console.error(res.message);
         }
     }
+    return webhookMail.parsedText
+}
+
+export async function triggerAnotherWorker(
+    c: Context<HonoCustomType>,
+    rpcEmailMessage: RPCEmailMessage,
+    parsedText: string | undefined | null
+): Promise<void> {
+    if (!parsedText) {
+        return;
+    }
+
+    const anotherWorkerList: AnotherWorker[] = getAnotherWorkerList(c);
+    if (!getBooleanValue(c.env.ENABLE_ANOTHER_WORKER) || anotherWorkerList.length === 0) {
+        console.log(`another worker disabled or anotherWorkerList is empty`);
+        return;
+    }
+
+    const parsedTextLowercase: string = parsedText.toLowerCase();
+    for (const worker of anotherWorkerList) {
+
+        const keywords = worker?.keywords ?? [];
+        const bindingName = worker?.binding ?? "";
+        const methodName = worker.method ?? "rpcEmail";
+
+        const serviceBinding = (c.env as any)[bindingName] ?? {};
+        const method = serviceBinding[methodName];
+
+        if (!method || typeof method !== "function") {
+            console.log(`method = ${methodName} not found or not function`);
+            continue;
+        }
+
+        if (!keywords.some(keyword => keyword && parsedTextLowercase.includes(keyword.toLowerCase()))) {
+            console.log(`worker.binding = ${bindingName} not match keywords, parsedText = ${parsedText}`);
+            continue;
+        }
+
+        try {
+            const requestBody = JSON.stringify(rpcEmailMessage);
+            await method(requestBody);
+        } catch (e1) {
+            console.error(`execute method = ${methodName} error`, e1);
+        }
+    }
 }

worker/src/email/index.ts

@@ -2,10 +2,10 @@ import { Context } from "hono";
 
 import { getEnvStringList } from "../utils";
 import { sendMailToTelegram } from "../telegram_api";
-import { Bindings, HonoCustomType } from "../types";
+import { Bindings, HonoCustomType, RPCEmailMessage } from "../types";
 import { auto_reply } from "./auto_reply";
 import { isBlocked } from "./black_list";
-import { triggerWebhook } from "../common";
+import { triggerWebhook, triggerAnotherWorker, commonParseMail} from "../common";
 import { check_if_junk_mail } from "./check_junk";
 
 
@@ -61,8 +61,9 @@ async function email(message: ForwardableEmailMessage, env: Bindings, ctx: Execu
     }
 
     // send webhook
+    let parsedText;
     try {
-        await triggerWebhook(
+        parsedText = await triggerWebhook(
             { env: env } as Context<HonoCustomType>,
             message.to, rawEmail, message_id
         );
@@ -70,6 +71,26 @@ async function email(message: ForwardableEmailMessage, env: Bindings, ctx: Execu
         console.log("send webhook error", error);
     }
 
+    // trigger another worker
+    try {
+        const headersMap = new Map<string, string>();
+        if(message.headers) {
+            message.headers.forEach((value, key) => {headersMap.set(key, value);});
+        }
+        if (!parsedText){
+          parsedText = (await commonParseMail(rawEmail))?.text ?? ""
+        }
+        const rpcEmail: RPCEmailMessage = {
+            from: message.from,
+            to: message.to,
+            rawEmail: rawEmail,
+            headers: headersMap
+        }
+        await triggerAnotherWorker({ env: env } as Context<HonoCustomType>, rpcEmail, parsedText);
+    } catch (error) {
+        console.error("trigger another worker error", error);
+    }
+
     // auto reply email
     await auto_reply(message, env);
 }

worker/src/types.d.ts

@@ -46,6 +46,10 @@ export type Bindings = {
     ENABLE_CHECK_JUNK_MAIL: string | boolean | undefined
     JUNK_MAIL_FORCE_PASS_LIST: string | string[] | undefined
 
+    ENABLE_ANOTHER_WORKER: string | boolean | undefined
+    ANOTHER_WORKER_LIST: string | AnotherWorker[] | undefined
+
+
     // s3 config
     S3_ENDPOINT: string | undefined
     S3_ACCESS_KEY_ID: string | undefined
@@ -92,3 +96,16 @@ type HonoCustomType = {
     "Bindings": Bindings;
     "Variables": Variables;
 }
+
+type AnotherWorker = {
+    binding: string | undefined | null,
+    method: string | undefined | null,
+    keywords: string[] | undefined | null
+}
+
+type RPCEmailMessage = {
+    from: string | undefined | null,
+    to: string | undefined | null,
+    rawEmail: string | undefined | null,
+    headers: Map<string, string>,
+}

worker/src/utils.ts

@@ -1,6 +1,6 @@
 import { Context } from "hono";
 import { createMimeMessage } from "mimetext";
-import { HonoCustomType, UserRole } from "./types";
+import { HonoCustomType, UserRole,AnotherWorker } from "./types";
 
 export const getJsonObjectValue = <T = any>(
     value: string | any
@@ -156,6 +156,22 @@ export const getUserRoles = (c: Context<HonoCustomType>): UserRole[] => {
     return c.env.USER_ROLES;
 }
 
+export const getAnotherWorkerList = (c: Context<HonoCustomType>): AnotherWorker[] => {
+    if (!c.env.ANOTHER_WORKER_LIST) {
+        return [];
+    }
+    // check if ANOTHER_WORKER_LIST is an array, if not use json.parse
+    if (!Array.isArray(c.env.ANOTHER_WORKER_LIST)) {
+        try {
+            return JSON.parse(c.env.ANOTHER_WORKER_LIST);
+        } catch (e) {
+            console.error("Failed to parse ANOTHER_WORKER_LIST", e);
+            return [];
+        }
+    }
+    return c.env.ANOTHER_WORKER_LIST;
+}
+
 export const getPasswords = (c: Context<HonoCustomType>): string[] => {
     if (!c.env.PASSWORDS) {
         return [];