E-Mail Cache und Konto-Synchronisation vorbereiten
KI-AGENT: Ergänzt Tabellen für lokalen E-Mail-Cache, IMAP-Sync-Service und Inbox-API. Überarbeitet außerdem die E-Mail-Konto-Seiten mit sicherer Passwortbehandlung und manuellem Sync.
This commit is contained in:
455
backend/src/modules/email/email.sync.service.ts
Normal file
455
backend/src/modules/email/email.sync.service.ts
Normal file
@@ -0,0 +1,455 @@
|
||||
import { FastifyInstance } from "fastify"
|
||||
import { and, desc, eq } from "drizzle-orm"
|
||||
import { ImapFlow } from "imapflow"
|
||||
import { simpleParser } from "mailparser"
|
||||
|
||||
import {
|
||||
emailAttachments,
|
||||
emailMailboxes,
|
||||
emailMessageBodies,
|
||||
emailMessages,
|
||||
emailSyncState,
|
||||
userCredentials,
|
||||
} from "../../../db/schema"
|
||||
import { decrypt } from "../../utils/crypt"
|
||||
|
||||
type EmailAddress = {
|
||||
name?: string | null
|
||||
address?: string | null
|
||||
}
|
||||
|
||||
type SyncOptions = {
|
||||
mailbox?: string
|
||||
limit?: number
|
||||
}
|
||||
|
||||
type MailAccountConnection = {
|
||||
id: string
|
||||
tenantId: number
|
||||
userId: string
|
||||
email: string
|
||||
password: string
|
||||
imapHost: string
|
||||
imapPort: number
|
||||
imapSsl: boolean
|
||||
}
|
||||
|
||||
const decryptValue = (value: unknown) => value ? decrypt(value as any) : ""
|
||||
|
||||
const normalizeAddressList = (addresses: any): EmailAddress[] => {
|
||||
const value = Array.isArray(addresses?.value) ? addresses.value : []
|
||||
return value.map((item: any) => ({
|
||||
name: item.name || null,
|
||||
address: item.address || null,
|
||||
}))
|
||||
}
|
||||
|
||||
const previewText = (text?: string | false | null) => {
|
||||
if (!text) return null
|
||||
return text.replace(/\s+/g, " ").trim().slice(0, 240) || null
|
||||
}
|
||||
|
||||
const flagsFromMessage = (flags: Set<string> | string[] | undefined) => {
|
||||
if (!flags) return []
|
||||
return Array.isArray(flags) ? flags : Array.from(flags)
|
||||
}
|
||||
|
||||
const mailboxDisplayName = (path: string) => {
|
||||
const parts = path.split(/[/.]/).filter(Boolean)
|
||||
return parts[parts.length - 1] || path
|
||||
}
|
||||
|
||||
export function emailSyncService(server: FastifyInstance) {
|
||||
const getAccount = async (
|
||||
tenantId: number,
|
||||
userId: string,
|
||||
accountId: string,
|
||||
): Promise<MailAccountConnection | null> => {
|
||||
const rows = await server.db
|
||||
.select()
|
||||
.from(userCredentials)
|
||||
.where(and(
|
||||
eq(userCredentials.id, accountId),
|
||||
eq(userCredentials.tenantId, tenantId),
|
||||
eq(userCredentials.userId, userId),
|
||||
eq(userCredentials.type, "mail"),
|
||||
))
|
||||
.limit(1)
|
||||
|
||||
const row = rows[0]
|
||||
if (!row) return null
|
||||
|
||||
return {
|
||||
id: row.id,
|
||||
tenantId: row.tenantId,
|
||||
userId: row.userId,
|
||||
email: decryptValue(row.emailEncrypted),
|
||||
password: decryptValue(row.passwordEncrypted),
|
||||
imapHost: decryptValue(row.imapHostEncrypted),
|
||||
imapPort: Number(row.imapPort || 993),
|
||||
imapSsl: row.imapSsl !== false,
|
||||
}
|
||||
}
|
||||
|
||||
const createClient = (account: MailAccountConnection) => new ImapFlow({
|
||||
host: account.imapHost,
|
||||
port: account.imapPort,
|
||||
secure: account.imapSsl,
|
||||
auth: {
|
||||
user: account.email,
|
||||
pass: account.password,
|
||||
},
|
||||
logger: false,
|
||||
})
|
||||
|
||||
const upsertMailbox = async (
|
||||
account: MailAccountConnection,
|
||||
mailbox: any,
|
||||
status?: { exists?: number; unseen?: number },
|
||||
) => {
|
||||
const path = mailbox.path || mailbox.name
|
||||
const [saved] = await server.db
|
||||
.insert(emailMailboxes)
|
||||
.values({
|
||||
tenantId: account.tenantId,
|
||||
userId: account.userId,
|
||||
accountId: account.id,
|
||||
path,
|
||||
delimiter: mailbox.delimiter || null,
|
||||
name: mailbox.name || mailboxDisplayName(path),
|
||||
specialUse: mailbox.specialUse || null,
|
||||
flags: flagsFromMessage(mailbox.flags),
|
||||
exists: status?.exists || 0,
|
||||
unseen: status?.unseen || 0,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [emailMailboxes.accountId, emailMailboxes.path],
|
||||
set: {
|
||||
delimiter: mailbox.delimiter || null,
|
||||
name: mailbox.name || mailboxDisplayName(path),
|
||||
specialUse: mailbox.specialUse || null,
|
||||
flags: flagsFromMessage(mailbox.flags),
|
||||
exists: status?.exists || 0,
|
||||
unseen: status?.unseen || 0,
|
||||
updatedAt: new Date(),
|
||||
},
|
||||
})
|
||||
.returning()
|
||||
|
||||
return saved
|
||||
}
|
||||
|
||||
const syncMailboxes = async (account: MailAccountConnection, client: ImapFlow) => {
|
||||
const savedMailboxes = []
|
||||
|
||||
for await (const mailbox of await client.list()) {
|
||||
savedMailboxes.push(await upsertMailbox(account, mailbox))
|
||||
}
|
||||
|
||||
return savedMailboxes
|
||||
}
|
||||
|
||||
const loadSyncState = async (account: MailAccountConnection, mailbox: any) => {
|
||||
const rows = await server.db
|
||||
.select()
|
||||
.from(emailSyncState)
|
||||
.where(and(
|
||||
eq(emailSyncState.accountId, account.id),
|
||||
eq(emailSyncState.mailboxPath, mailbox.path),
|
||||
))
|
||||
.limit(1)
|
||||
|
||||
return rows[0] || null
|
||||
}
|
||||
|
||||
const saveSyncState = async (
|
||||
account: MailAccountConnection,
|
||||
mailbox: any,
|
||||
highestUid: number,
|
||||
uidValidity?: number | null,
|
||||
syncError?: string | null,
|
||||
) => {
|
||||
await server.db
|
||||
.insert(emailSyncState)
|
||||
.values({
|
||||
tenantId: account.tenantId,
|
||||
userId: account.userId,
|
||||
accountId: account.id,
|
||||
mailboxId: mailbox.id,
|
||||
mailboxPath: mailbox.path,
|
||||
uidValidity: uidValidity || null,
|
||||
highestUid,
|
||||
lastSyncedAt: new Date(),
|
||||
syncError: syncError || null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [emailSyncState.accountId, emailSyncState.mailboxPath],
|
||||
set: {
|
||||
mailboxId: mailbox.id,
|
||||
uidValidity: uidValidity || null,
|
||||
highestUid,
|
||||
lastSyncedAt: new Date(),
|
||||
syncError: syncError || null,
|
||||
updatedAt: new Date(),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
const storeMessage = async (
|
||||
account: MailAccountConnection,
|
||||
mailbox: any,
|
||||
message: any,
|
||||
) => {
|
||||
if (!message.source) return null
|
||||
|
||||
const parsed = await simpleParser(message.source)
|
||||
const flags = flagsFromMessage(message.flags)
|
||||
const receivedAt = parsed.date || message.envelope?.date || new Date()
|
||||
const threadId = parsed.inReplyTo || parsed.references?.[0] || parsed.messageId || message.emailId || null
|
||||
|
||||
const [saved] = await server.db
|
||||
.insert(emailMessages)
|
||||
.values({
|
||||
tenantId: account.tenantId,
|
||||
userId: account.userId,
|
||||
accountId: account.id,
|
||||
mailboxId: mailbox.id,
|
||||
mailboxPath: mailbox.path,
|
||||
uid: Number(message.uid),
|
||||
emailId: message.emailId || null,
|
||||
messageId: parsed.messageId || null,
|
||||
inReplyTo: parsed.inReplyTo || null,
|
||||
threadId,
|
||||
subject: parsed.subject || "(kein Betreff)",
|
||||
from: normalizeAddressList(parsed.from),
|
||||
to: normalizeAddressList(parsed.to),
|
||||
cc: normalizeAddressList(parsed.cc),
|
||||
bcc: normalizeAddressList(parsed.bcc),
|
||||
replyTo: normalizeAddressList(parsed.replyTo),
|
||||
preview: previewText(parsed.text),
|
||||
flags,
|
||||
seen: flags.includes("\\Seen"),
|
||||
flagged: flags.includes("\\Flagged"),
|
||||
hasAttachments: Boolean(parsed.attachments?.length),
|
||||
size: message.size || null,
|
||||
sentAt: parsed.date || null,
|
||||
receivedAt,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [emailMessages.mailboxId, emailMessages.uid],
|
||||
set: {
|
||||
emailId: message.emailId || null,
|
||||
messageId: parsed.messageId || null,
|
||||
inReplyTo: parsed.inReplyTo || null,
|
||||
threadId,
|
||||
subject: parsed.subject || "(kein Betreff)",
|
||||
from: normalizeAddressList(parsed.from),
|
||||
to: normalizeAddressList(parsed.to),
|
||||
cc: normalizeAddressList(parsed.cc),
|
||||
bcc: normalizeAddressList(parsed.bcc),
|
||||
replyTo: normalizeAddressList(parsed.replyTo),
|
||||
preview: previewText(parsed.text),
|
||||
flags,
|
||||
seen: flags.includes("\\Seen"),
|
||||
flagged: flags.includes("\\Flagged"),
|
||||
hasAttachments: Boolean(parsed.attachments?.length),
|
||||
size: message.size || null,
|
||||
sentAt: parsed.date || null,
|
||||
receivedAt,
|
||||
updatedAt: new Date(),
|
||||
},
|
||||
})
|
||||
.returning()
|
||||
|
||||
await server.db
|
||||
.insert(emailMessageBodies)
|
||||
.values({
|
||||
messageId: saved.id,
|
||||
text: parsed.text || null,
|
||||
html: parsed.html || null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: emailMessageBodies.messageId,
|
||||
set: {
|
||||
text: parsed.text || null,
|
||||
html: parsed.html || null,
|
||||
updatedAt: new Date(),
|
||||
},
|
||||
})
|
||||
|
||||
if (parsed.attachments?.length) {
|
||||
for (const attachment of parsed.attachments) {
|
||||
await server.db
|
||||
.insert(emailAttachments)
|
||||
.values({
|
||||
messageId: saved.id,
|
||||
filename: attachment.filename || null,
|
||||
contentType: attachment.contentType || null,
|
||||
contentId: attachment.contentId || null,
|
||||
disposition: attachment.contentDisposition || null,
|
||||
size: attachment.size || null,
|
||||
checksum: attachment.checksum || null,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
}
|
||||
}
|
||||
|
||||
return saved
|
||||
}
|
||||
|
||||
const syncMailboxMessages = async (
|
||||
account: MailAccountConnection,
|
||||
client: ImapFlow,
|
||||
mailbox: any,
|
||||
limit: number,
|
||||
) => {
|
||||
const lock = await client.getMailboxLock(mailbox.path)
|
||||
let highestUid = 0
|
||||
|
||||
try {
|
||||
const opened: any = await client.mailboxOpen(mailbox.path)
|
||||
await upsertMailbox(account, mailbox, {
|
||||
exists: opened.exists || 0,
|
||||
unseen: opened.unseen || 0,
|
||||
})
|
||||
|
||||
const state = await loadSyncState(account, mailbox)
|
||||
const searchResult = await client.search({ all: true }, { uid: true })
|
||||
const allUids = Array.isArray(searchResult) ? searchResult : []
|
||||
const newUids = allUids
|
||||
.filter((uid: number) => !state?.highestUid || uid > state.highestUid)
|
||||
.slice(-limit)
|
||||
|
||||
highestUid = Math.max(state?.highestUid || 0, ...newUids, 0)
|
||||
|
||||
if (newUids.length) {
|
||||
for await (const message of client.fetch(newUids, {
|
||||
uid: true,
|
||||
envelope: true,
|
||||
flags: true,
|
||||
source: true,
|
||||
size: true,
|
||||
}, { uid: true })) {
|
||||
await storeMessage(account, mailbox, message)
|
||||
}
|
||||
}
|
||||
|
||||
await saveSyncState(account, mailbox, highestUid, Number(opened.uidValidity || 0))
|
||||
return { path: mailbox.path, fetched: newUids.length, highestUid }
|
||||
} catch (err: any) {
|
||||
await saveSyncState(account, mailbox, highestUid, null, err.message || "Sync fehlgeschlagen")
|
||||
throw err
|
||||
} finally {
|
||||
lock.release()
|
||||
}
|
||||
}
|
||||
|
||||
const syncAccount = async (
|
||||
tenantId: number,
|
||||
userId: string,
|
||||
accountId: string,
|
||||
options: SyncOptions = {},
|
||||
) => {
|
||||
const account = await getAccount(tenantId, userId, accountId)
|
||||
if (!account) {
|
||||
throw new Error("E-Mail Konto nicht gefunden")
|
||||
}
|
||||
|
||||
const client = createClient(account)
|
||||
const limit = Math.min(Math.max(Number(options.limit || 50), 1), 200)
|
||||
|
||||
await client.connect()
|
||||
|
||||
try {
|
||||
const mailboxes = await syncMailboxes(account, client)
|
||||
const syncTargets = options.mailbox
|
||||
? mailboxes.filter((mailbox) => mailbox.path === options.mailbox)
|
||||
: mailboxes.filter((mailbox) => mailbox.specialUse === "\\Inbox" || mailbox.path.toUpperCase() === "INBOX")
|
||||
|
||||
const synced = []
|
||||
for (const mailbox of syncTargets.length ? syncTargets : mailboxes.slice(0, 1)) {
|
||||
synced.push(await syncMailboxMessages(account, client, mailbox, limit))
|
||||
}
|
||||
|
||||
return {
|
||||
accountId,
|
||||
mailboxes: mailboxes.length,
|
||||
synced,
|
||||
}
|
||||
} finally {
|
||||
await client.logout().catch(() => client.close())
|
||||
}
|
||||
}
|
||||
|
||||
const listMailboxes = async (tenantId: number, userId: string, accountId: string) => {
|
||||
return await server.db
|
||||
.select()
|
||||
.from(emailMailboxes)
|
||||
.where(and(
|
||||
eq(emailMailboxes.tenantId, tenantId),
|
||||
eq(emailMailboxes.userId, userId),
|
||||
eq(emailMailboxes.accountId, accountId),
|
||||
))
|
||||
.orderBy(emailMailboxes.specialUse, emailMailboxes.name)
|
||||
}
|
||||
|
||||
const listMessages = async (
|
||||
tenantId: number,
|
||||
userId: string,
|
||||
accountId: string,
|
||||
mailboxPath = "INBOX",
|
||||
limit = 50,
|
||||
) => {
|
||||
return await server.db
|
||||
.select()
|
||||
.from(emailMessages)
|
||||
.where(and(
|
||||
eq(emailMessages.tenantId, tenantId),
|
||||
eq(emailMessages.userId, userId),
|
||||
eq(emailMessages.accountId, accountId),
|
||||
eq(emailMessages.mailboxPath, mailboxPath),
|
||||
))
|
||||
.orderBy(desc(emailMessages.receivedAt))
|
||||
.limit(Math.min(Math.max(Number(limit), 1), 200))
|
||||
}
|
||||
|
||||
const getMessage = async (tenantId: number, userId: string, messageId: string) => {
|
||||
const rows = await server.db
|
||||
.select({
|
||||
message: emailMessages,
|
||||
body: emailMessageBodies,
|
||||
})
|
||||
.from(emailMessages)
|
||||
.leftJoin(emailMessageBodies, eq(emailMessageBodies.messageId, emailMessages.id))
|
||||
.where(and(
|
||||
eq(emailMessages.tenantId, tenantId),
|
||||
eq(emailMessages.userId, userId),
|
||||
eq(emailMessages.id, messageId),
|
||||
))
|
||||
.limit(1)
|
||||
|
||||
if (!rows[0]) return null
|
||||
|
||||
const attachments = await server.db
|
||||
.select()
|
||||
.from(emailAttachments)
|
||||
.where(eq(emailAttachments.messageId, messageId))
|
||||
|
||||
return {
|
||||
...rows[0].message,
|
||||
body: rows[0].body,
|
||||
attachments,
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
syncAccount,
|
||||
listMailboxes,
|
||||
listMessages,
|
||||
getMessage,
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user