refactor: store mail flags in sparse relation table

worker/src/admin_api/address_api.ts

@@ -4,6 +4,7 @@ import { Jwt } from 'hono/utils/jwt'
 import i18n from '../i18n'
 import { getBooleanValue } from '../utils'
 import { newAddress, handleListQuery } from '../common'
+import { deleteRawMails, prepareRawMailDeleteStatements } from '../mail_flags'
 
 const listAddresses = async (c: Context<HonoCustomType>) => {
     const { limit, offset, query, sort_by, sort_order } = c.req.query();
@@ -74,10 +75,12 @@ const deleteAddress = async (c: Context<HonoCustomType>) => {
     // deleted first and the address row last, so the name subqueries still
     // resolve and a failed statement rolls back the whole deletion
     const results = await c.env.DB.batch([
-        c.env.DB.prepare(
-            `DELETE FROM raw_mails WHERE address IN`
-            + ` (select name from address where id = ?) `
-        ).bind(id),
+        ...prepareRawMailDeleteStatements(
+            c.env.DB,
+            c.env,
+            `address IN (select name from address where id = ?)`,
+            [id],
+        ),
         c.env.DB.prepare(
             `DELETE FROM address_sender WHERE address IN`
             + ` (select name from address where id = ?) `
@@ -107,10 +110,12 @@ const deleteAddress = async (c: Context<HonoCustomType>) => {
 const clearInbox = async (c: Context<HonoCustomType>) => {
     const msgs = i18n.getMessagesbyContext(c);
     const { id } = c.req.param();
-    const { success: mailSuccess } = await c.env.DB.prepare(
-        `DELETE FROM raw_mails WHERE address IN`
-        + ` (select name from address where id = ?) `
-    ).bind(id).run();
+    const { success: mailSuccess } = await deleteRawMails(
+        c.env.DB,
+        c.env,
+        `address IN (select name from address where id = ?)`,
+        [id],
+    );
     if (!mailSuccess) {
         return c.text(msgs.OperationFailedMsg, 500)
     }

worker/src/admin_api/admin_mail_api.ts

@@ -1,7 +1,7 @@
 import { Context } from "hono";
 import { handleMailListQuery } from "../common";
 import { resolveRawEmailRow } from "../gzip";
-import { serializeMailState } from "../mail_flags";
+import { deleteRawMails, serializeMailState } from "../mail_flags";
 
 export default {
     getMails: async (c: Context<HonoCustomType>) => {
@@ -32,16 +32,15 @@ export default {
             `SELECT * FROM raw_mails WHERE id = ?`
         ).bind(id).first();
         if (!result) return c.json(null);
-        return c.json(serializeMailState(
+        return c.json(await serializeMailState(
+            c.env.DB,
             await resolveRawEmailRow(result),
             c.env,
         ));
     },
     deleteMail: async (c: Context<HonoCustomType>) => {
         const { id } = c.req.param();
-        const { success } = await c.env.DB.prepare(
-            `DELETE FROM raw_mails WHERE id = ? `
-        ).bind(id).run();
+        const { success } = await deleteRawMails(c.env.DB, c.env, `id = ?`, [id]);
         return c.json({
             success: success
         })

worker/src/admin_api/cleanup_api.ts

@@ -6,6 +6,7 @@ import { getJsonSetting, saveSetting } from '../utils';
 import { CleanupSettings, CustomSqlCleanup } from '../models';
 import i18n from '../i18n';
 import { LocaleMessages } from '../i18n/type';
+import { cleanupOrphanMailFlags } from '../mail_flags';
 
 // SQL validation error types
 type SqlValidationError = 'empty' | 'too_long' | 'not_delete' | 'multiple_statements' | 'has_comments';
@@ -84,6 +85,7 @@ export const executeCustomSqlCleanup = async (
         console.log(`Executing custom SQL cleanup [${customSql.name}]: ${sql}`);
         const result = await c.env.DB.prepare(sql).run();
         const rowsAffected = result.meta?.changes ?? 0;
+        await cleanupOrphanMailFlags(c.env.DB, c.env);
         console.log(`Custom SQL cleanup [${customSql.name}] completed, rows affected: ${rowsAffected}`);
         return { success: true, rowsAffected };
     } catch (error) {

worker/src/admin_api/db_api.ts

@@ -11,7 +11,6 @@ CREATE TABLE IF NOT EXISTS raw_mails (
     raw TEXT,
     raw_blob BLOB,
     metadata TEXT,
-    flags INTEGER,
     created_at DATETIME DEFAULT CURRENT_TIMESTAMP
 );
 
@@ -21,6 +20,16 @@ CREATE INDEX IF NOT EXISTS idx_raw_mails_created_at ON raw_mails(created_at);
 
 CREATE INDEX IF NOT EXISTS idx_raw_mails_message_id ON raw_mails(message_id);
 
+CREATE TABLE IF NOT EXISTS mail_flags (
+    mail_id INTEGER NOT NULL,
+    address_id INTEGER NOT NULL,
+    flag INTEGER NOT NULL,
+    PRIMARY KEY (mail_id, flag)
+) WITHOUT ROWID;
+
+CREATE INDEX IF NOT EXISTS idx_mail_flags_address_flag_mail
+ON mail_flags(address_id, flag, mail_id DESC);
+
 CREATE TABLE IF NOT EXISTS address (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     name TEXT UNIQUE,
@@ -198,16 +207,17 @@ export default {
                 await c.env.DB.exec(`ALTER TABLE raw_mails ADD COLUMN raw_blob BLOB;`);
             }
         }
-        if (version && version <= "v0.0.7") {
-            const tableInfo = await c.env.DB.prepare(
-                `PRAGMA table_info(raw_mails)`
-            ).all();
-            const hasFlags = tableInfo.results?.some(
-                (col: any) => col.name === 'flags'
-            );
-            if (!hasFlags) {
-                await c.env.DB.exec(`ALTER TABLE raw_mails ADD COLUMN flags INTEGER;`);
-            }
+        if (version && version <= "v0.0.8") {
+            await c.env.DB.exec(`
+                CREATE TABLE IF NOT EXISTS mail_flags (
+                    mail_id INTEGER NOT NULL,
+                    address_id INTEGER NOT NULL,
+                    flag INTEGER NOT NULL,
+                    PRIMARY KEY (mail_id, flag)
+                ) WITHOUT ROWID;
+                CREATE INDEX IF NOT EXISTS idx_mail_flags_address_flag_mail
+                ON mail_flags(address_id, flag, mail_id DESC);
+            `);
         }
         if (version != CONSTANTS.DB_VERSION) {
             // remove all \r and \n characters from the query string

worker/src/admin_api/e2e_test_api.ts

@@ -78,4 +78,18 @@ const receiveMail = async (c: Context<HonoCustomType>) => {
     });
 };
 
-export default { seedMail, receiveMail };
+const getMailFlags = async (c: Context<HonoCustomType>) => {
+    if (!getBooleanValue(c.env.E2E_TEST_MODE)) {
+        return c.text("Not available", 404);
+    }
+    const mailId = Number(c.req.query('mail_id'));
+    if (!Number.isInteger(mailId) || mailId <= 0) {
+        return c.text("Invalid mail_id", 400);
+    }
+    const { results } = await c.env.DB.prepare(
+        `SELECT mail_id, address_id, flag FROM mail_flags WHERE mail_id = ? ORDER BY flag`
+    ).bind(mailId).all();
+    return c.json({ results });
+};
+
+export default { seedMail, receiveMail, getMailFlags };

worker/src/admin_api/index.ts

@@ -112,3 +112,4 @@ api.post('/admin/ai_extract/settings', ai_extract_settings.saveAiExtractSettings
 // E2E test endpoints
 api.post('/admin/test/seed_mail', e2e_test_api.seedMail)
 api.post('/admin/test/receive_mail', e2e_test_api.receiveMail)
+api.get('/admin/test/mail_flags', e2e_test_api.getMailFlags)

worker/src/common.ts

@@ -7,7 +7,7 @@ import { unbindTelegramByAddress } from './telegram_api/common';
 import { CONSTANTS } from './constants';
 import { AddressCreationSettings, AdminWebhookSettings, ExtractResult, WebhookMail, WebhookSettings } from './models';
 import i18n from './i18n';
-import { serializeMailState } from './mail_flags';
+import { deleteRawMails, serializeMailStates } from './mail_flags';
 
 const DEFAULT_NAME_REGEX = /[^a-z0-9]/g;
 const DEFAULT_RANDOM_SUBDOMAIN_LENGTH = 8;
@@ -528,20 +528,25 @@ export const cleanup = async (
             )
             break;
         case "mails":
-            await c.env.DB.prepare(`
-                DELETE FROM raw_mails WHERE id IN (
+            await deleteRawMails(
+                c.env.DB,
+                c.env,
+                `id IN (
                     SELECT id FROM raw_mails
                     WHERE created_at < datetime('now', ?)
                     ORDER BY created_at, id
-                    LIMIT ?
-                )`
-            ).bind(`-${cleanDays} day`, cleanupBatchSize).run();
+                    LIMIT ?)`,
+                [`-${cleanDays} day`, cleanupBatchSize],
+            );
             break;
         case "mails_unknow":
-            await c.env.DB.prepare(`
-                DELETE FROM raw_mails WHERE address NOT IN
-                (select name from address) AND created_at < datetime('now', '-${cleanDays} day')`
-            ).run();
+            await deleteRawMails(
+                c.env.DB,
+                c.env,
+                `address NOT IN (select name from address)`
+                    + ` AND created_at < datetime('now', '-${cleanDays} day')`,
+                [],
+            );
             break;
         case "sendbox":
             await c.env.DB.prepare(`
@@ -570,10 +575,12 @@ const batchDeleteAddressWithData = async (
     c: Context<HonoCustomType>,
     addressQueryCondition: string,
 ): Promise<boolean> => {
-    await c.env.DB.prepare(
-        `DELETE FROM raw_mails WHERE address IN ( ` +
-        `SELECT name FROM address WHERE ${addressQueryCondition})`
-    ).run();
+    await deleteRawMails(
+        c.env.DB,
+        c.env,
+        `address IN (SELECT name FROM address WHERE ${addressQueryCondition})`,
+        [],
+    );
     await c.env.DB.prepare(
         `DELETE FROM sendbox WHERE address IN ( ` +
         `SELECT name FROM address WHERE ${addressQueryCondition})`
@@ -627,9 +634,12 @@ export const deleteAddressWithData = async (
     // unbind telegram
     await unbindTelegramByAddress(c, address);
     // delete address and related data
-    const { success: mailSuccess } = await c.env.DB.prepare(
-        `DELETE FROM raw_mails WHERE address = ? `
-    ).bind(address).run();
+    const { success: mailSuccess } = await deleteRawMails(
+        c.env.DB,
+        c.env,
+        `address = ?`,
+        [address],
+    );
     const { success: sendAccess } = await c.env.DB.prepare(
         `DELETE FROM address_sender WHERE address = ? `
     ).bind(address).run();
@@ -705,7 +715,7 @@ export const hideObjectFields = <T extends Record<string, unknown>>(
  */
 export const handleMailListQuery = async (
     c: Context<HonoCustomType>,
-    query: string, countQuery: string, params: string[],
+    query: string, countQuery: string, params: (string | number)[],
     limit: string | number | undefined | null,
     offset: string | number | undefined | null,
     orderBy?: string
@@ -722,10 +732,7 @@ export const handleMailListQuery = async (
         ...params, limit, offset
     ).all();
     const resolvedResults = await resolveRawEmailList(results);
-    const serializedResults = resolvedResults.map(row => serializeMailState(
-        row,
-        c.env,
-    ));
+    const serializedResults = await serializeMailStates(c.env.DB, resolvedResults, c.env);
     const count = offset == 0 ? await c.env.DB.prepare(
         countQuery
     ).bind(...params).first("count") : 0;

worker/src/mail_flags.ts

@@ -1,13 +1,11 @@
-import { getBooleanValue } from './utils';
-
-export const MAIL_FLAGS = {
-    UNREAD: 1 << 0,
-    ANSWERED: 1 << 1,
-    FLAGGED: 1 << 2,
-    DELETED: 1 << 3,
-    DRAFT: 1 << 4,
-    JUNK: 1 << 5,
-} as const;
+export enum MailFlag {
+    UNREAD = 0,
+    ANSWERED = 1,
+    FLAGGED = 2,
+    DELETED = 3,
+    DRAFT = 4,
+    JUNK = 5,
+}
 
 export const CUSTOM_MAIL_FLAG_OFFSET = 10;
 export const CUSTOM_MAIL_FLAG_COUNT = 10;
@@ -26,9 +24,14 @@ export type MailStateOption = {
     default?: boolean;
 };
 
+type MailStateMutation = {
+    add?: number;
+    remove?: number[];
+};
+
 export type MailStateDefinition = MailStateOption & {
-    filter?: { mask: number; set: boolean };
-    mutation?: { add: number; remove: number };
+    filter?: { flag: number; present: boolean };
+    mutation?: MailStateMutation;
 };
 
 const SYSTEM_MAIL_STATES: MailStateDefinition[] = [
@@ -37,15 +40,15 @@ const SYSTEM_MAIL_STATES: MailStateDefinition[] = [
         value: MailState.UNREAD,
         label_key: 'unread',
         unread: true,
-        filter: { mask: MAIL_FLAGS.UNREAD, set: true },
-        mutation: { add: MAIL_FLAGS.UNREAD, remove: 0 },
+        filter: { flag: MailFlag.UNREAD, present: true },
+        mutation: { add: MailFlag.UNREAD },
     },
     {
         value: MailState.READ,
         label_key: 'read',
         unread: false,
-        filter: { mask: MAIL_FLAGS.UNREAD, set: false },
-        mutation: { add: 0, remove: MAIL_FLAGS.UNREAD },
+        filter: { flag: MailFlag.UNREAD, present: false },
+        mutation: { remove: [MailFlag.UNREAD] },
     },
 ];
 
@@ -68,9 +71,9 @@ const getMailStateDefinition = (
 
 export const getCustomMailFlag = (slot: number): number => {
     if (!Number.isInteger(slot) || slot < 0 || slot >= CUSTOM_MAIL_FLAG_COUNT) {
-        throw new Error("Invalid custom mail flag slot");
+        throw new Error('Invalid custom mail flag slot');
     }
-    return 1 << (CUSTOM_MAIL_FLAG_OFFSET + slot);
+    return CUSTOM_MAIL_FLAG_OFFSET + slot;
 };
 
 export type CustomMailStateConfig = {
@@ -78,7 +81,10 @@ export type CustomMailStateConfig = {
     name: string;
 };
 
-const CUSTOM_MAIL_FLAGS_MASK = ((1 << CUSTOM_MAIL_FLAG_COUNT) - 1) << CUSTOM_MAIL_FLAG_OFFSET;
+const CUSTOM_MAIL_FLAGS = Array.from(
+    { length: CUSTOM_MAIL_FLAG_COUNT },
+    (_, slot) => getCustomMailFlag(slot),
+);
 
 export const createCustomMailStateDefinitions = (
     configs: CustomMailStateConfig[],
@@ -88,23 +94,58 @@ export const createCustomMailStateDefinitions = (
         return {
             value: `custom:${config.slot}`,
             label: config.name,
-            filter: { mask: flag, set: true },
-            mutation: { add: flag, remove: CUSTOM_MAIL_FLAGS_MASK & ~flag },
+            filter: { flag, present: true },
+            mutation: { add: flag, remove: CUSTOM_MAIL_FLAGS.filter(value => value !== flag) },
         };
     });
 };
 
-export const serializeMailState = <T extends Record<string, unknown>>(
+const withoutLegacyFlags = <T extends Record<string, unknown>>(row: T): T => {
+    const result = { ...row };
+    delete result.flags;
+    return result;
+};
+
+const isMailFlagsEnabled = (env: Bindings): boolean => {
+    return env.ENABLE_MAIL_FLAGS === true || env.ENABLE_MAIL_FLAGS === 'true';
+};
+
+export const serializeMailStates = async <T extends Record<string, unknown>>(
+    db: D1Database,
+    rows: T[],
+    env: Bindings,
+): Promise<T[]> => {
+    const results = rows.map(withoutLegacyFlags);
+    if (!isMailFlagsEnabled(env) || results.length === 0) return results;
+
+    const needsUnread = (row: T) => ![true, false, 0, 1].includes(row.unread as boolean | number);
+    const ids = [...new Set(results.filter(needsUnread).map(row => Number(row.id)))]
+        .filter(id => Number.isInteger(id) && id > 0);
+    if (ids.length === 0) {
+        return results.map(row => ({ ...row, unread: Boolean(row.unread) }));
+    }
+
+    const placeholders = ids.map(() => '?').join(',');
+    const { results: flagRows } = await db.prepare(
+        `SELECT mf.mail_id FROM mail_flags mf`
+        + ` JOIN raw_mails rm ON rm.id = mf.mail_id`
+        + ` JOIN address a ON a.id = mf.address_id AND a.name = rm.address`
+        + ` WHERE mf.flag = ? AND mf.mail_id IN (${placeholders})`
+    ).bind(MailFlag.UNREAD, ...ids).all<{ mail_id: number }>();
+    const unreadIds = new Set(flagRows.map(row => Number(row.mail_id)));
+
+    return results.map(row => ({
+        ...row,
+        unread: needsUnread(row) ? unreadIds.has(Number(row.id)) : Boolean(row.unread),
+    }));
+};
+
+export const serializeMailState = async <T extends Record<string, unknown>>(
+    db: D1Database,
     row: T,
     env: Bindings,
-): T => {
-    const result = { ...row };
-    const flags = Number(result.flags ?? 0);
-    delete result.flags;
-    if (!getBooleanValue(env.ENABLE_MAIL_FLAGS)) {
-        return result;
-    }
-    result.unread = (flags & MAIL_FLAGS.UNREAD) !== 0;
+): Promise<T> => {
+    const [result] = await serializeMailStates(db, [row], env);
     return result;
 };
 
@@ -112,8 +153,8 @@ const resolveInitialMailFlags = async (
     _env: Bindings,
     _address: string,
     _parsedEmailContext: ParsedEmailContext,
-): Promise<number> => {
-    return MAIL_FLAGS.UNREAD;
+): Promise<number[]> => {
+    return [MailFlag.UNREAD];
 };
 
 export const updateInitialMailFlags = async (
@@ -125,43 +166,62 @@ export const updateInitialMailFlags = async (
     parsedEmailContext: ParsedEmailContext,
 ) => {
     if (!enabled || !Number.isInteger(mailId) || mailId <= 0) return;
+
     const flags = await resolveInitialMailFlags(env, address, parsedEmailContext);
-    await db.prepare(`UPDATE raw_mails SET flags = ? WHERE id = ?`).bind(flags, mailId).run();
+    if (flags.length === 0) return;
+
+    const selects = flags.map(() => 'SELECT ?, id, ? FROM address WHERE name = ?').join(' UNION ALL ');
+    await db.prepare(
+        `INSERT OR IGNORE INTO mail_flags (mail_id, address_id, flag) ${selects}`
+    ).bind(...flags.flatMap(flag => [mailId, flag, address])).run();
 };
 
-export type MailStateUpdate = {
-    ids: number[];
-    state: string;
-    add: number;
-    remove: number;
-};
-
-export type MailReadStatusQuery = {
-    clause: string;
-    params: string[];
+export type MailStateQuery = {
+    join: string;
+    clause?: string;
+    orderBy?: string;
+    unread?: boolean;
+    params: number[];
 };
 
 export const getMailStateQuery = (
     value: string | undefined,
-    column: 'flags' | 'rm.flags',
+    mailAlias: string,
+    addressIdColumn: string,
     customStates: MailStateDefinition[] = [],
-): MailReadStatusQuery | undefined | null => {
+): MailStateQuery | undefined | null => {
     if (value === undefined) return undefined;
+
     const definition = getMailStateDefinition(value, customStates);
     if (!definition) return null;
     if (!definition.filter) return undefined;
-    const operator = definition.filter.set ? '!=' : '=';
+
+    const { flag, present } = definition.filter;
+    const joinType = present ? 'JOIN' : 'LEFT JOIN';
     return {
-        clause: `(COALESCE(${column}, 0) & ?) ${operator} 0`,
-        params: [String(definition.filter.mask)],
+        join: ` ${joinType} mail_flags mail_state_flags`
+            + ` ON mail_state_flags.mail_id = ${mailAlias}.id`
+            + ` AND mail_state_flags.address_id = ${addressIdColumn}`
+            + ` AND mail_state_flags.flag = ?`,
+        clause: present ? undefined : 'mail_state_flags.mail_id IS NULL',
+        orderBy: present ? 'mail_state_flags.mail_id desc' : undefined,
+        unread: definition.unread,
+        params: [flag],
     };
 };
 
+type MailStateUpdate = {
+    ids: number[];
+    mutation: MailStateMutation;
+    unread?: boolean;
+};
+
 const parseMailStateUpdate = (
     value: unknown,
     customStates: MailStateDefinition[] = [],
 ): MailStateUpdate | null => {
     if (!value || typeof value !== 'object') return null;
+
     const body = value as Record<string, unknown>;
     if (!Array.isArray(body.ids) || body.ids.length === 0 || body.ids.length > 100) return null;
     if (body.ids.some(id => typeof id !== 'number')) return null;
@@ -171,24 +231,7 @@ const parseMailStateUpdate = (
 
     const definition = getMailStateDefinition(body.state, customStates);
     if (!definition?.mutation) return null;
-    return {
-        ids,
-        state: definition.value,
-        add: definition.mutation.add,
-        remove: definition.mutation.remove,
-    };
-};
-
-const getMailStateUpdateExpression = (
-    update: MailStateUpdate,
-    column = 'flags',
-): { expression: string; params: number[]; condition?: string; conditionParams?: number[] } => {
-    return {
-        expression: `((COALESCE(${column}, 0) | ?) & ~?)`,
-        params: [update.add, update.remove],
-        condition: `((COALESCE(${column}, 0) & ?) != ? OR (COALESCE(${column}, 0) & ?) != 0)`,
-        conditionParams: [update.add, update.add, update.remove],
-    };
+    return { ids, mutation: definition.mutation, unread: definition.unread };
 };
 
 type MailScope = {
@@ -206,27 +249,87 @@ export const applyMailStateUpdate = async (
     const update = parseMailStateUpdate(value, customStates);
     if (!update) return null;
 
-    const placeholders = update.ids.map(() => '?').join(',');
-    const statusUpdate = getMailStateUpdateExpression(update);
-    const condition = statusUpdate.condition ? ` AND ${statusUpdate.condition}` : '';
-    const result = await db.prepare(
-        `UPDATE raw_mails SET flags = ${statusUpdate.expression}`
-        + ` WHERE id IN (${placeholders}) AND (${scope.clause})${condition}`
-    ).bind(
-        ...statusUpdate.params,
-        ...update.ids,
-        ...scope.params,
-        ...(statusUpdate.conditionParams ?? []),
-    ).run();
-    if (!result.success) return { success: false, changes: 0, results: [] };
+    const idPlaceholders = update.ids.map(() => '?').join(',');
+    const targetWhere = `rm.id IN (${idPlaceholders}) AND (${scope.clause})`;
+    const statements: D1PreparedStatement[] = [];
 
+    if (update.mutation.remove?.length) {
+        const flagPlaceholders = update.mutation.remove.map(() => '?').join(',');
+        statements.push(db.prepare(
+            `DELETE FROM mail_flags WHERE flag IN (${flagPlaceholders})`
+            + ` AND mail_id IN (`
+            + `SELECT rm.id FROM raw_mails rm JOIN address a ON a.name = rm.address`
+            + ` WHERE ${targetWhere})`
+        ).bind(...update.mutation.remove, ...update.ids, ...scope.params));
+    }
+
+    if (update.mutation.add !== undefined) {
+        statements.push(db.prepare(
+            `INSERT OR IGNORE INTO mail_flags (mail_id, address_id, flag)`
+            + ` SELECT rm.id, a.id, ?`
+            + ` FROM raw_mails rm JOIN address a ON a.name = rm.address`
+            + ` WHERE ${targetWhere}`
+        ).bind(update.mutation.add, ...update.ids, ...scope.params));
+    }
+
+    const mutationResults = await db.batch(statements);
+    if (mutationResults.some(result => !result.success)) {
+        return { success: false, changes: 0, results: [] };
+    }
+
+    const unreadSelect = update.unread === undefined ? '' : `, ${update.unread ? 1 : 0} AS unread`;
     const { results } = await db.prepare(
-        `SELECT id, flags FROM raw_mails`
-        + ` WHERE id IN (${placeholders}) AND (${scope.clause})`
+        `SELECT rm.id${unreadSelect} FROM raw_mails rm JOIN address a ON a.name = rm.address`
+        + ` WHERE ${targetWhere}`
     ).bind(...update.ids, ...scope.params).all();
+
     return {
         success: true,
-        changes: result.meta.changes ?? 0,
-        results: results.map(row => serializeMailState(row, env)),
+        changes: mutationResults.reduce((total, result) => total + (result.meta.changes ?? 0), 0),
+        results: await serializeMailStates(db, results, env),
     };
 };
+
+export const prepareRawMailDeleteStatements = (
+    db: D1Database,
+    env: Bindings,
+    whereClause: string,
+    params: (string | number)[],
+): D1PreparedStatement[] => {
+    const deleteMail = db.prepare(
+        `DELETE FROM raw_mails WHERE ${whereClause}`
+    ).bind(...params);
+    if (!isMailFlagsEnabled(env)) return [deleteMail];
+
+    const deleteFlags = db.prepare(
+        `DELETE FROM mail_flags WHERE mail_id IN (`
+        + `SELECT id FROM raw_mails WHERE ${whereClause})`
+    ).bind(...params);
+    return [deleteFlags, deleteMail];
+};
+
+export const deleteRawMails = async (
+    db: D1Database,
+    env: Bindings,
+    whereClause: string,
+    params: (string | number)[],
+): Promise<D1Result> => {
+    const results = await db.batch(prepareRawMailDeleteStatements(db, env, whereClause, params));
+    return results[results.length - 1];
+};
+
+export const cleanupOrphanMailFlags = async (
+    db: D1Database,
+    env: Bindings,
+    limit = 1000,
+): Promise<number> => {
+    if (!isMailFlagsEnabled(env)) return 0;
+
+    const result = await db.prepare(
+        `DELETE FROM mail_flags WHERE (mail_id, flag) IN (`
+        + `SELECT mf.mail_id, mf.flag FROM mail_flags mf`
+        + ` LEFT JOIN raw_mails rm ON rm.id = mf.mail_id`
+        + ` WHERE rm.id IS NULL LIMIT ?)`
+    ).bind(limit).run();
+    return result.meta.changes ?? 0;
+};

worker/src/mails_api/mails_crud.ts

@@ -10,6 +10,7 @@ import {
     getMailStateOptions,
     applyMailStateUpdate,
     serializeMailState,
+    deleteRawMails,
 } from '../mail_flags';
 
 const listMails = async (c: Context<HonoCustomType>) => {
@@ -19,23 +20,33 @@ const listMails = async (c: Context<HonoCustomType>) => {
     }
     const { limit, offset, mail_state } = c.req.query();
     if (Number.parseInt(offset) <= 0) updateAddressUpdatedAt(c, address);
-    const stateQuery = getMailStateQuery(mail_state, 'flags');
+    const stateQuery = getMailStateQuery(mail_state, 'rm', 'a.id');
     if (stateQuery === null) return c.json({ error: "Invalid mail state filter" }, 400);
     if (stateQuery && !getBooleanValue(c.env.ENABLE_MAIL_FLAGS)) {
         return c.json({ error: "Mail states are disabled" }, 403);
     }
 
-    const filters = [`address = ?`];
-    const params = [address];
-    if (stateQuery) {
-        filters.push(stateQuery.clause);
-        params.push(...stateQuery.params);
+    if (!stateQuery) {
+        return await handleMailListQuery(c,
+            `SELECT * FROM raw_mails WHERE address = ?`,
+            `SELECT count(*) as count FROM raw_mails WHERE address = ?`,
+            [address], limit, offset
+        );
     }
-    const whereClause = filters.join(' AND ');
+
+    const filters = [`rm.address = ?`];
+    if (stateQuery.clause) filters.push(stateQuery.clause);
+    const fromQuery = ` FROM raw_mails rm`
+        + ` JOIN address a ON a.name = rm.address`
+        + stateQuery.join
+        + ` WHERE ${filters.join(' AND ')}`;
+    const unreadSelect = stateQuery.unread === undefined
+        ? ''
+        : `, ${stateQuery.unread ? 1 : 0} AS unread`;
     return await handleMailListQuery(c,
-        `SELECT * FROM raw_mails WHERE ${whereClause}`,
-        `SELECT count(*) as count FROM raw_mails WHERE ${whereClause}`,
-        params, limit, offset
+        `SELECT rm.*${unreadSelect}${fromQuery}`,
+        `SELECT count(*) as count${fromQuery}`,
+        [...stateQuery.params, address], limit, offset, stateQuery.orderBy ?? 'rm.id desc'
     );
 };
 
@@ -46,7 +57,8 @@ const getMail = async (c: Context<HonoCustomType>) => {
         `SELECT * FROM raw_mails where id = ? and address = ?`
     ).bind(mail_id, address).first();
     if (!result) return c.json(null);
-    return c.json(serializeMailState(
+    return c.json(await serializeMailState(
+        c.env.DB,
         await resolveRawEmailRow(result),
         c.env,
     ));
@@ -60,9 +72,12 @@ const deleteMail = async (c: Context<HonoCustomType>) => {
     const { address } = c.get("jwtPayload")
     const { id } = c.req.param();
     // TODO: add toLowerCase() to handle old data
-    const { success } = await c.env.DB.prepare(
-        `DELETE FROM raw_mails WHERE address = ? and id = ? `
-    ).bind(address.toLowerCase(), id).run();
+    const { success } = await deleteRawMails(
+        c.env.DB,
+        c.env,
+        `address = ? and id = ?`,
+        [address.toLowerCase(), id],
+    );
     return c.json({ success });
 };
 
@@ -74,7 +89,7 @@ const updateMailState = async (c: Context<HonoCustomType>) => {
     const result = await applyMailStateUpdate(
         c.env.DB,
         c.env,
-        { clause: 'address = ?', params: [address] },
+        { clause: 'rm.address = ?', params: [address] },
         await c.req.json().catch(() => null),
     );
     if (!result) return c.json({ error: "Invalid mail state request" }, 400);
@@ -138,9 +153,12 @@ const clearInbox = async (c: Context<HonoCustomType>) => {
         return c.text(msgs.UserDeleteEmailDisabledMsg, 403)
     }
     const { address } = c.get("jwtPayload")
-    const { success } = await c.env.DB.prepare(
-        `DELETE FROM raw_mails WHERE address = ?`
-    ).bind(address).run();
+    const { success } = await deleteRawMails(
+        c.env.DB,
+        c.env,
+        `address = ?`,
+        [address],
+    );
     if (!success) {
         return c.text(msgs.FailedClearInboxMsg, 500)
     }

worker/src/mails_api/parsed_mail_api.ts

@@ -46,7 +46,8 @@ const getParsedMail = async (c: Context<HonoCustomType>) => {
         `SELECT * FROM raw_mails where id = ? and address = ?`
     ).bind(mail_id, address).first();
     if (!row) return c.json(null);
-    const resolved = serializeMailState(
+    const resolved = await serializeMailState(
+        c.env.DB,
         await resolveRawEmailRow(row),
         c.env,
     );

worker/src/models/index.ts

@@ -213,7 +213,6 @@ export type RawMailRow = {
     raw?: string;
     raw_blob?: unknown;
     metadata?: string;
-    flags?: number | null;
     unread?: boolean;
     created_at?: string;
 }

worker/src/telegram_api/miniapp.ts

@@ -146,7 +146,7 @@ async function getMail(c: Context<HonoCustomType>): Promise<Response> {
             if (!result) {
                 return c.text("Mail not found", 404);
             }
-            return c.json(serializeMailState(await resolveRawEmailRow(result), c.env));
+            return c.json(await serializeMailState(c.env.DB, await resolveRawEmailRow(result), c.env));
         }
         const userId = await checkTelegramAuth(c, initData);
         const jwtList = await c.env.KV.get<string[]>(`${CONSTANTS.TG_KV_PREFIX}:${userId}`, 'json') || [];
@@ -169,7 +169,7 @@ async function getMail(c: Context<HonoCustomType>): Promise<Response> {
                 return c.text(msgs.TgNoPermissionViewMailMsg, 403);
             }
         }
-        return c.json(serializeMailState(await resolveRawEmailRow(result), c.env));
+        return c.json(await serializeMailState(c.env.DB, await resolveRawEmailRow(result), c.env));
     }
     catch (e) {
         return c.text((e as Error).message, 400);

worker/src/user_api/user_mail_api.ts

@@ -6,6 +6,7 @@ import {
     getMailStateQuery,
     getMailStateOptions,
     applyMailStateUpdate,
+    deleteRawMails,
 } from "../mail_flags";
 
 export default {
@@ -24,23 +25,24 @@ export default {
             filterQuerys.push(`rm.address = ?`);
             filterParams.push(address);
         }
-        const stateQuery = getMailStateQuery(mail_state, 'rm.flags');
+        const stateQuery = getMailStateQuery(mail_state, 'rm', 'a.id');
         if (stateQuery === null) return c.json({ error: "Invalid mail state filter" }, 400);
         if (stateQuery && !getBooleanValue(c.env.ENABLE_MAIL_FLAGS)) {
             return c.json({ error: "Mail states are disabled" }, 403);
         }
-        if (stateQuery) {
-            filterQuerys.push(stateQuery.clause);
-            filterParams.push(...stateQuery.params);
-        }
+        if (stateQuery?.clause) filterQuerys.push(stateQuery.clause);
         const fromQuery = ` FROM users_address ua`
             + ` JOIN address a ON a.id = ua.address_id`
             + ` JOIN raw_mails rm ON rm.address = a.name`
+            + (stateQuery?.join ?? '')
             + ` WHERE ${filterQuerys.join(" AND ")}`;
+        const unreadSelect = stateQuery?.unread === undefined
+            ? ''
+            : `, ${stateQuery.unread ? 1 : 0} AS unread`;
         return await handleMailListQuery(c,
-            `SELECT rm.*${fromQuery}`,
+            `SELECT rm.*${unreadSelect}${fromQuery}`,
             `SELECT count(*) as count${fromQuery}`,
-            filterParams, limit, offset, 'rm.id desc'
+            [...(stateQuery?.params ?? []), ...filterParams], limit, offset, 'rm.id desc'
         );
     },
     deleteMail: async (c: Context<HonoCustomType>) => {
@@ -50,14 +52,17 @@ export default {
         }
         const { id } = c.req.param();
         const { user_id } = c.get("userPayload");
-        const { success } = await c.env.DB.prepare(
-            `DELETE FROM raw_mails WHERE id = ?`
+        const { success } = await deleteRawMails(
+            c.env.DB,
+            c.env,
+            `id = ?`
             + ` AND EXISTS (`
             + `SELECT 1 FROM users_address ua`
             + ` JOIN address a ON a.id = ua.address_id`
             + ` WHERE ua.user_id = ? AND a.name = raw_mails.address`
-            + `)`
-        ).bind(id, user_id).run();
+            + `)`,
+            [id, user_id],
+        );
         return c.json({
             success: success
         })
@@ -71,11 +76,9 @@ export default {
             c.env.DB,
             c.env,
             {
-                clause: `EXISTS (`
-                + `SELECT 1 FROM users_address ua`
-                + ` JOIN address a ON a.id = ua.address_id`
-                + ` WHERE ua.user_id = ? AND a.name = raw_mails.address`
-                + `)`,
+                clause: `a.id IN (`
+                    + `SELECT address_id FROM users_address WHERE user_id = ?`
+                    + `)`,
                 params: [user_id],
             },
             await c.req.json().catch(() => null),