Inbound: IMAP → ticket

TLDR

Một cron mỗi phút quét từng hd_connections có inbound_provider=imap, dùng ImapFlow kéo mail mới hơn watermark UID, buffer toàn bộ vào RAM trước rồi mới xử lý (tránh lỗi “Connection not available” vì imapflow serialise command). Mỗi email được đẩy qua processInboundEmail — pipeline 8 bước biến nó thành ticket mới hoặc nối vào hội thoại cũ.

Vị trí mã nguồn

• Thư mụcbackend/extensions/helpdesk/src/endpoints/services/inbound/

  • imap-source.ts cron poll, ImapFlow, watermark, buffer-then-process
  • pipeline.ts processInboundEmail — 8 bước
  • normalized.ts NormalizedInboundEmail + loop-guard + sender filter + plus-routing
  • sanitize.ts detectThreats + sanitizeInboundHtml
  • crypto.ts AES-256-GCM cho secret hd_connections
  • security-alert.ts gắn label “Security Risk” + Power Notify
  • ../attachments.ts storeAttachments — upload qua FilesService

Cron poll: pollImapConnections

hooks/index.ts đăng ký cron * * * * * (per-minute, đa-instance-safe qua context.schedule) gọi pollImapConnections(ctx):

backend/extensions/helpdesk/src/endpoints/services/inbound/imap-source.ts

const conns: HdConnection[] = await ctx.database('hd_connections')
    .where({ inbound_provider: 'imap' })
    .whereNotIn('status', ['error', 'disconnected', 'disabled'])   // poll unless explicitly down
    .where((qb: any) => qb.whereNull('needs_reauth').orWhere('needs_reauth', false));
for (const conn of conns) {
    try {
        await pollOne(ctx, conn);
    } catch (err: any) {
        await onConnError(ctx, conn, err);
    }
}

Một guard pollInFlight (module-level boolean) ngăn 2 tick chồng lên nhau trong cùng một instance khi mailbox chậm. Cross-instance overlap do context.schedule của ODP lo (leader election).

pollOne: watermark + first-sync

Mỗi connection dùng imap_last_uid + imap_uidvalidity làm watermark. Lần đầu bật connection (hoặc UIDVALIDITY đổi — mailbox bị rebuild), pipeline không kéo mail cũ: nó set watermark = UID mới nhất hiện có rồi return, chỉ mail đến sau đó mới được ingest.

imap-source.ts — first sync

let lastUid = Number(conn.imap_last_uid ?? 0);
const validityChanged = uidValidity !== Number(conn.imap_uidvalidity ?? 0);
if (!lastUid || validityChanged) {
    lastUid = Math.max(0, uidNext - 1);
    await updateWatermark(ctx, conn.id, { imap_uidvalidity: uidValidity, imap_last_uid: lastUid });
    await touchSynced(ctx, conn.id);
    return;
}

Buffer-then-process

imapflow serialise command trên cùng một connection — nếu bạn chạy lệnh IMAP khác (như messageFlagsAdd/messageMove để đánh dấu đã xử lý) hoặc một query DB chậm trong lúc fetch stream còn mở, nó ném Connection not available. Giải pháp: tách 2 pha.

1. Phase 1 — drain fetch stream vào RAM. Đọc tối đa MAX_PER_TICK = 50 message (uid + source Buffer) rồi thoát vòng lặp, không xử lý gì trong lúc stream còn mở.
2. Phase 2 — xử lý sau khi stream đã đóng. Với mỗi message đã buffer: simpleParser → processInboundEmail → markProcessed (an toàn gọi IMAP command ở đây vì không còn fetch stream).

imap-source.ts — 2 phase

const buffered: Array<{ uid: number; source: Buffer }> = [];
let count = 0;
for await (const msg of client.fetch({ uid: `${lastUid + 1}:*` }, { uid: true, source: true })) {
    if (count++ >= MAX_PER_TICK) break;         // continue next tick (watermark advances)
    if (msg.source) buffered.push({ uid: Number(msg.uid), source: msg.source as Buffer });
}

let maxUid = lastUid;
for (const { uid, source } of buffered) {
    const normalized = mapMailparser(await simpleParser(source));
    // ... poison-message check, processInboundEmail, markProcessed
}

Giới hạn scale

Buffer giữ toàn bộ raw message trong RAM cho tới MAX_PER_TICK = 50 mỗi connection mỗi tick — nếu một mailbox nhận email đính kèm lớn dồn dập, RAM tăng theo số connection đang poll song song. Throughput cap thực tế là 50 message/phút/connection; vượt ngưỡng thì phần còn lại đợi tick sau (watermark chỉ tiến tới message cuối cùng xử lý thành công).

Poison-message backoff

Một message lỗi liên tục (parse hỏng, pipeline throw) sẽ chặn watermark tiến lên mãi mãi nếu không có cơ chế bỏ qua. imap-source.ts đếm số lần fail theo messageId qua ctx.cache (best-effort, TTL 6h):

imap-source.ts — poison-message

if (normalized.messageId && (await failCount(ctx, normalized.messageId)) >= FAIL_LIMIT) {
    await markProcessed(client, uid, conn);
    maxUid = Math.max(maxUid, uid);
    continue;
}

FAIL_LIMIT = 3 — sau 3 lần thất bại trong 6 giờ, message bị đánh dấu processed và bỏ qua để tick sau không kẹt mãi ở cùng một mail.

Pipeline 8 bước: processInboundEmail

pipeline.ts

export async function processInboundEmail(
    ctx: AppContext,
    m: NormalizedInboundEmail,
    conn: HdConnection,
): Promise<InboundResult> {

1. Resolve inbox. Một connection có thể chỉ backing 1 inbox (dùng luôn), hoặc backing nhiều inbox dùng chung mailbox — khi đó định tuyến theo plus-address (support+project1@… khớp receiving_address của inbox nào). Không khớp inbox nào → dropped: no-inbox-match (không bao giờ rơi nhầm inbox).
2. Loop-guard. loopGuardReason() drop mail tự-gửi (from = chính inbox), bounce (mailer-daemon@, X-Failed-Recipients), auto-reply (Auto-Submitted, X-Autoreply), mailing-list (List-Id), hoặc chính marker outbound của hệ thống (X-Helpdesk-Loop — xem Outbound).
3. Sender filter. Mỗi inbox có inbound_filters (mode allow/deny + rule email/domain/pattern) trong channel_config. Người gửi không được phép → drop trước khi tạo contact/conversation.
4. Detect threats + sanitize. detectThreats() quét HTML thô tìm <script>, <iframe>, event-handler, javascript: URI… Nếu có, HTML được sanitizeInboundHtml() (sanitize-html, allowlist tag/attribute, strip script/style, buộc rel="noopener noreferrer nofollow" trên link) trước khi lưu — chỉ HTML sạch được ghi vào hd_messages.content.
5. Dedup theo Message-ID. Check nhanh hd_messages.external_message_id trùng → skipped: duplicate. DB unique constraint là chốt chặn cứng cuối (race concurrent → bắt lỗi unique-violation, coi như skip).
6. Contact choke point. ensureContactForRequester() — cửa duy nhất tạo/tìm c_contacts theo primary_email.
7. Threading — thử theo thứ tự tín hiệu mạnh nhất trước:
   • reply+<convId>@ trong To/Cc (mạnh nhất, sống sót qua mọi mail client).
   • In-Reply-To / References khớp external_message_id đã lưu.
   • Không có tín hiệu nào → luôn tạo hội thoại mới (không tự gộp mọi email cùng contact/inbox vào 1 thread).
8. Create/append + attachments + security flag. Hội thoại mới → createConversationWithFirstMessage; hội thoại có sẵn → createItemsService('hd_messages').createOne(...) (append), và nếu hội thoại đang resolved thì tự mở lại open. Cả 2 nhánh đều gọi storeAttachments() rồi maybeFlagSecurity() nếu bước 4 phát hiện threat.

pipeline.ts — resolve inbox (plus-routing)

async function resolveInbox(ctx: AppContext, m: NormalizedInboundEmail, conn: HdConnection) {
    const inboxes: any[] = await db('hd_inboxes').where('connection_id', conn.id);
    // ...
    if (inboxes.length === 1) return inboxes[0];
    const recipients = inboundRecipients(m);
    for (const ib of inboxes) {
        if (matchesReceivingAddress(recipients, inboxReceivingAddress(ib))) return ib;
    }
    return null;
}

Mã hoá credential: crypto.ts

hd_connections không dùng chung util mã hoá của @odp/api (package chỉ export . và ./types, không phải dependency của extension) — extension tự cài AES-256-GCM bằng node:crypto, mirror format enc:iv:tag:ct cho quen mắt:

crypto.ts

function keyFrom(secret: string): Buffer {
    return createHash('sha256').update(String(secret)).digest(); // 32 bytes
}

export function encrypt(plain: string, secret: string): string {
    const iv = randomBytes(12);
    const cipher = createCipheriv('aes-256-gcm', keyFrom(secret), iv);
    const ct = Buffer.concat([cipher.update(String(plain), 'utf8'), cipher.final()]);
    const tag = cipher.getAuthTag();
    return `${PREFIX}${iv.toString('hex')}:${tag.toString('hex')}:${ct.toString('hex')}`;
}

Key = SHA-256(env.SECRET). isEncrypted() chỉ kiểm tra prefix enc: — một chuỗi plaintext tình cờ bắt đầu bằng enc: sẽ bị coi là đã mã hoá và decrypt thất bại im lặng (decrypt() trả nguyên giá trị nếu tách iv:tag:ct lỗi).

False-positive heuristic

isEncrypted() chỉ dựa vào prefix chuỗi, không phải structural check. Đây là heuristic đủ dùng cho secret nội bộ (không phải input user tự do), nhưng không phải bằng chứng mã hoá tuyệt đối — đừng tái dùng hàm này để xác thực dữ liệu từ nguồn không tin cậy.

Bảo mật: security-alert.ts

Khi bước 4 phát hiện threat “đáng báo động” (script, iframe, object, embed, javascript-uri, data-html-uri, svg-script), flagConversationSecurity():

1. Gắn label đỏ “Security Risk” (#dc2626) vào hội thoại — tạo label nếu chưa tồn tại.
2. Emit notify.trigger action event với template helpdesk_security_alert (seed idempotent lúc boot) tới toàn bộ admin_emails cấu hình ở odp_notify_settings.

security-alert.ts

const ALERT_THREATS = new Set(['script', 'iframe', 'object', 'embed', 'javascript-uri', 'data-html-uri', 'svg-script']);
export function isAlertWorthy(threats: string[]): boolean {
    return threats.some((t) => ALERT_THREATS.has(t));
}

Giới hạn cần biết khi vận hành

Scale

• RAM buffer: tối đa 50 message thô/connection/tick giữ trong bộ nhớ tiến trình — nhiều connection cùng poll song song với mailbox lớn có thể cộng dồn đáng kể.
• Throughput cap: 50 message/phút/connection là trần cứng; mailbox nhận dồn dập hơn sẽ bị dồn sang tick kế tiếp (không mất mail, chỉ trễ).
• Poison-message TTL 6h dùng ctx.cache best-effort (không có cache thì luôn coi là 0 lần fail — pipeline sẽ retry vô hạn mỗi tick cho tới khi cache khả dụng).

Đọc tiếp

• Outbound (SMTP) — chiều gửi đi, threading ngược, loop-guard marker
• Master/Satellite & Trust model — contact choke point dùng ở bước 6
• Automation engine — trigger message_received chạy sau khi message được tạo