feat: add extensible mail flags

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,6 +1,7 @@
 import { Context } from "hono";
 import { handleMailListQuery } from "../common";
 import { resolveRawEmailRow } from "../gzip";
+import { deleteRawMails } from "../mail_flags";
 
 export default {
     getMails: async (c: Context<HonoCustomType>) => {
@@ -35,9 +36,7 @@ export default {
     },
     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

@@ -20,6 +20,15 @@ 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,

worker/src/admin_api/worker_config.ts

@@ -40,6 +40,8 @@ export default {
             "ENABLE_USER_CREATE_EMAIL": utils.getBooleanValue(c.env.ENABLE_USER_CREATE_EMAIL),
             "DISABLE_ANONYMOUS_USER_CREATE_EMAIL": utils.getBooleanValue(c.env.DISABLE_ANONYMOUS_USER_CREATE_EMAIL),
             "ENABLE_USER_DELETE_EMAIL": utils.getBooleanValue(c.env.ENABLE_USER_DELETE_EMAIL),
+            "ENABLE_MAIL_READ_STATUS": utils.getBooleanValue(c.env.ENABLE_MAIL_READ_STATUS),
+            "ENABLE_MAIL_FLAGGED": utils.getBooleanValue(c.env.ENABLE_MAIL_FLAGGED),
             "ENABLE_AUTO_REPLY": utils.getBooleanValue(c.env.ENABLE_AUTO_REPLY),
             "COPYRIGHT": c.env.COPYRIGHT,
             "ENABLE_WEBHOOK": utils.getBooleanValue(c.env.ENABLE_WEBHOOK),

worker/src/commom_api.ts

@@ -20,6 +20,8 @@ api.get('/open_api/settings', async (c) => {
     ) || {};
     const smtpProxyConfig = smtpImapProxyConfig.smtp || {};
     const imapProxyConfig = smtpImapProxyConfig.imap || {};
+    const enableMailReadStatus = utils.getBooleanValue(c.env.ENABLE_MAIL_READ_STATUS);
+    const enableMailFlagged = utils.getBooleanValue(c.env.ENABLE_MAIL_FLAGGED);
 
     return c.json({
         "title": c.env.TITLE,
@@ -39,6 +41,8 @@ api.get('/open_api/settings', async (c) => {
         "disableAnonymousUserCreateEmail": utils.getBooleanValue(c.env.DISABLE_ANONYMOUS_USER_CREATE_EMAIL),
         "disableCustomAddressName": utils.getBooleanValue(c.env.DISABLE_CUSTOM_ADDRESS_NAME),
         "enableUserDeleteEmail": utils.getBooleanValue(c.env.ENABLE_USER_DELETE_EMAIL),
+        ...(enableMailReadStatus ? { "enableMailReadStatus": true } : {}),
+        ...(enableMailFlagged ? { "enableMailFlagged": true } : {}),
         "enableAutoReply": utils.getBooleanValue(c.env.ENABLE_AUTO_REPLY),
         "enableIndexAbout": utils.getBooleanValue(c.env.ENABLE_INDEX_ABOUT),
         "copyright": c.env.COPYRIGHT,

worker/src/common.ts

@@ -7,6 +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 { deleteRawMails, serializeMailStates } from './mail_flags';
 
 const DEFAULT_NAME_REGEX = /[^a-z0-9]/g;
 const DEFAULT_RANDOM_SUBDOMAIN_LENGTH = 8;
@@ -527,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(`
@@ -569,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})`
@@ -626,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();
@@ -704,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
@@ -721,10 +732,11 @@ export const handleMailListQuery = async (
         ...params, limit, offset
     ).all();
     const resolvedResults = await resolveRawEmailList(results);
+    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;
-    return c.json({ results: resolvedResults, count });
+    return c.json({ results: serializedResults, count });
 }
 
 export const commonParseMail = async (parsedEmailContext: ParsedEmailContext): Promise<{

worker/src/constants.ts

@@ -3,7 +3,7 @@ export const CONSTANTS = {
 
     // DB Version
     DB_VERSION_KEY: 'db_version',
-    DB_VERSION: "v0.0.7",
+    DB_VERSION: "v0.0.8",
 
     // DB settings
     ADDRESS_BLOCK_LIST_KEY: 'address_block_list',

worker/src/email/index.ts

@@ -12,6 +12,7 @@ import { forwardEmail } from "./forward";
 import { EmailRuleSettings } from "../models";
 import { CONSTANTS } from "../constants";
 import { compressText } from "../gzip";
+import { initializeMailFlagsAfterInsert } from "../mail_flags";
 
 
 async function email(message: ForwardableEmailMessage, env: Bindings, ctx: ExecutionContext) {
@@ -67,7 +68,7 @@ async function email(message: ForwardableEmailMessage, env: Bindings, ctx: Execu
     const message_id = message.headers.get("Message-ID");
     // save email
     try {
-        let success = false;
+        let insertResult: D1Result | null = null;
         if (getBooleanValue(env.ENABLE_MAIL_GZIP)) {
             let compressed: ArrayBuffer | null = null;
             try {
@@ -77,42 +78,49 @@ async function email(message: ForwardableEmailMessage, env: Bindings, ctx: Execu
             }
             if (compressed) {
                 try {
-                    ({ success } = await env.DB.prepare(
+                    insertResult = await env.DB.prepare(
                         `INSERT INTO raw_mails (source, address, raw_blob, message_id) VALUES (?, ?, ?, ?)`
                     ).bind(
                         message.from, toAddress, compressed, message_id
-                    ).run());
+                    ).run();
                 } catch (dbError) {
                     // Fallback to plaintext only if raw_blob column is missing (migration not applied)
                     const errMsg = String(dbError);
                     if (errMsg.includes('raw_blob') || errMsg.includes('no such column')) {
                         console.error("raw_blob column missing, falling back to plaintext", dbError);
-                        ({ success } = await env.DB.prepare(
+                        insertResult = await env.DB.prepare(
                             `INSERT INTO raw_mails (source, address, raw, message_id) VALUES (?, ?, ?, ?)`
                         ).bind(
                             message.from, toAddress, parsedEmailContext.rawEmail, message_id
-                        ).run());
+                        ).run();
                     } else {
                         throw dbError;
                     }
                 }
             } else {
-                ({ success } = await env.DB.prepare(
+                insertResult = await env.DB.prepare(
                     `INSERT INTO raw_mails (source, address, raw, message_id) VALUES (?, ?, ?, ?)`
                 ).bind(
                     message.from, toAddress, parsedEmailContext.rawEmail, message_id
-                ).run());
+                ).run();
             }
         } else {
-            ({ success } = await env.DB.prepare(
+            insertResult = await env.DB.prepare(
                 `INSERT INTO raw_mails (source, address, raw, message_id) VALUES (?, ?, ?, ?)`
             ).bind(
                 message.from, toAddress, parsedEmailContext.rawEmail, message_id
-            ).run());
+            ).run();
         }
-        if (!success) {
+        if (!insertResult?.success) {
             message.setReject(`Failed save message to ${toAddress}`);
             console.error(`Failed save message from ${message.from} to ${toAddress}`);
+        } else {
+            await initializeMailFlagsAfterInsert(
+                env.DB,
+                env,
+                insertResult?.meta.last_row_id ?? 0,
+                toAddress,
+            );
         }
     }
     catch (error) {

worker/src/mail_flags.ts

@@ -0,0 +1,343 @@
+export enum MailFlag {
+    UNREAD = 0,
+    FLAGGED = 1,
+}
+
+export enum MailState {
+    ALL = 'all',
+    UNREAD = 'unread',
+    READ = 'read',
+}
+
+export type MailStateOption = {
+    value: string;
+    label_key: string;
+    unread?: boolean;
+    default?: boolean;
+};
+
+type MailStateDefinition = MailStateOption & {
+    filter?: { flag: MailFlag; present: boolean };
+};
+
+const MAIL_STATES: MailStateDefinition[] = [
+    { value: MailState.ALL, label_key: 'allMail', default: true },
+    {
+        value: MailState.UNREAD,
+        label_key: 'unread',
+        unread: true,
+        filter: { flag: MailFlag.UNREAD, present: true },
+    },
+    {
+        value: MailState.READ,
+        label_key: 'read',
+        unread: false,
+        filter: { flag: MailFlag.UNREAD, present: false },
+    },
+];
+
+export const getMailStateOptions = (): MailStateOption[] => {
+    return MAIL_STATES.map(({ filter: _filter, ...option }) => option);
+};
+
+const getMailStateDefinition = (value: unknown): MailStateDefinition | undefined => {
+    if (typeof value !== 'string') return undefined;
+    return MAIL_STATES.find(state => state.value === value);
+};
+
+const isEnabled = (value: boolean | string | undefined): boolean => {
+    return value === true || value === 'true';
+};
+
+export const isMailReadStatusEnabled = (env: Bindings): boolean => {
+    return isEnabled(env.ENABLE_MAIL_READ_STATUS);
+};
+
+export const isMailFlaggedEnabled = (env: Bindings): boolean => {
+    return isEnabled(env.ENABLE_MAIL_FLAGGED);
+};
+
+const isAnyMailFlagEnabled = (env: Bindings): boolean => {
+    return isMailReadStatusEnabled(env) || isMailFlaggedEnabled(env);
+};
+
+export const serializeMailStates = async <T extends Record<string, unknown>>(
+    db: D1Database,
+    rows: T[],
+    env: Bindings,
+): Promise<T[]> => {
+    const readStatusEnabled = isMailReadStatusEnabled(env);
+    const flaggedEnabled = isMailFlaggedEnabled(env);
+    if ((!readStatusEnabled && !flaggedEnabled) || rows.length === 0) return rows;
+
+    const hasBoolean = (row: T, field: string) => {
+        return [true, false, 0, 1].includes(row[field] as boolean | number);
+    };
+    const ids = [...new Set(rows
+        .filter(row => (readStatusEnabled && !hasBoolean(row, 'unread'))
+            || (flaggedEnabled && !hasBoolean(row, 'flagged')))
+        .map(row => Number(row.id)))]
+        .filter(id => Number.isInteger(id) && id > 0);
+    if (ids.length === 0) {
+        return rows.map(row => ({
+            ...row,
+            ...(readStatusEnabled ? { unread: Boolean(row.unread) } : {}),
+            ...(flaggedEnabled ? { flagged: Boolean(row.flagged) } : {}),
+        }));
+    }
+
+    const flags = [
+        ...(readStatusEnabled ? [MailFlag.UNREAD] : []),
+        ...(flaggedEnabled ? [MailFlag.FLAGGED] : []),
+    ];
+    const flagPlaceholders = flags.map(() => '?').join(',');
+    const placeholders = ids.map(() => '?').join(',');
+    const { results } = await db.prepare(
+        `SELECT mf.mail_id, mf.flag, a.name AS address FROM mail_flags mf`
+        + ` JOIN address a ON a.id = mf.address_id`
+        + ` WHERE mf.flag IN (${flagPlaceholders}) AND mf.mail_id IN (${placeholders})`
+    ).bind(...flags, ...ids).all<{
+        mail_id: number;
+        flag: number;
+        address: string;
+    }>();
+    const mailKey = (id: unknown, address: unknown) => `${Number(id)}\0${String(address)}`;
+    const unreadKeys = new Set(results
+        .filter(row => row.flag === MailFlag.UNREAD)
+        .map(row => mailKey(row.mail_id, row.address)));
+    const flaggedKeys = new Set(results
+        .filter(row => row.flag === MailFlag.FLAGGED)
+        .map(row => mailKey(row.mail_id, row.address)));
+
+    return rows.map(row => ({
+        ...row,
+        ...(readStatusEnabled ? {
+            unread: hasBoolean(row, 'unread')
+                ? Boolean(row.unread)
+                : unreadKeys.has(mailKey(row.id, row.address)),
+        } : {}),
+        ...(flaggedEnabled ? {
+            flagged: hasBoolean(row, 'flagged')
+                ? Boolean(row.flagged)
+                : flaggedKeys.has(mailKey(row.id, row.address)),
+        } : {}),
+    }));
+};
+
+export const serializeMailState = async <T extends Record<string, unknown>>(
+    db: D1Database,
+    row: T,
+    env: Bindings,
+): Promise<T> => {
+    const [result] = await serializeMailStates(db, [row], env);
+    return result;
+};
+
+export const initializeMailFlagsAfterInsert = async (
+    db: D1Database,
+    env: Bindings,
+    mailId: number,
+    address: string,
+): Promise<void> => {
+    if (!isMailReadStatusEnabled(env) || !Number.isInteger(mailId) || mailId <= 0) return;
+
+    try {
+        await db.prepare(
+            `INSERT OR IGNORE INTO mail_flags (mail_id, address_id, flag)`
+            + ` SELECT ?, id, ? FROM address WHERE name = ?`
+        ).bind(mailId, MailFlag.UNREAD, address).run();
+    } catch (error) {
+        console.error(`Failed to initialize mail flags for mail ${mailId}`, error);
+    }
+};
+
+export type MailStateQuery = {
+    join: string;
+    clause?: string;
+    orderBy?: string;
+    unread?: boolean;
+    flagged?: boolean;
+    params: number[];
+};
+
+export const getMailFlaggedQuery = (
+    value: string | undefined,
+    mailAlias: string,
+    addressIdColumn: string,
+): MailStateQuery | undefined | null => {
+    if (value === undefined) return undefined;
+    if (value !== 'true' && value !== 'false') return null;
+
+    const present = value === 'true';
+    return {
+        join: ` ${present ? 'JOIN' : 'LEFT JOIN'} mail_flags mail_flagged_flags`
+            + ` ON mail_flagged_flags.mail_id = ${mailAlias}.id`
+            + ` AND mail_flagged_flags.address_id = ${addressIdColumn}`
+            + ` AND mail_flagged_flags.flag = ?`,
+        clause: present ? undefined : 'mail_flagged_flags.mail_id IS NULL',
+        orderBy: present ? 'mail_flagged_flags.mail_id desc' : undefined,
+        flagged: present,
+        params: [MailFlag.FLAGGED],
+    };
+};
+
+export const getMailStateQuery = (
+    value: string | undefined,
+    mailAlias: string,
+    addressIdColumn: string,
+): MailStateQuery | undefined | null => {
+    if (value === undefined) return undefined;
+
+    const definition = getMailStateDefinition(value);
+    if (!definition) return null;
+    if (!definition.filter) return undefined;
+
+    const { flag, present } = definition.filter;
+    return {
+        join: ` ${present ? 'JOIN' : 'LEFT JOIN'} 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 MailFlagUpdate = {
+    ids: number[];
+    body: Record<string, unknown>;
+};
+
+const parseMailFlagUpdate = (value: unknown): MailFlagUpdate | 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;
+
+    const ids = [...new Set(body.ids.map(Number))];
+    if (ids.some(id => !Number.isInteger(id) || id <= 0)) return null;
+
+    return { ids, body };
+};
+
+type MailScope = {
+    clause: string;
+    params: (string | number)[];
+};
+
+const applyMailFlagUpdate = async (
+    db: D1Database,
+    scope: MailScope,
+    ids: number[],
+    flag: MailFlag,
+    present: boolean,
+    resultField: 'unread' | 'flagged',
+) => {
+    const placeholders = ids.map(() => '?').join(',');
+    const targetWhere = `rm.id IN (${placeholders}) AND (${scope.clause})`;
+    const mutation = present
+        ? 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(flag, ...ids, ...scope.params)
+        : db.prepare(
+            `DELETE FROM mail_flags WHERE flag = ? AND mail_id IN (`
+            + `SELECT rm.id FROM raw_mails rm JOIN address a ON a.name = rm.address`
+            + ` WHERE ${targetWhere})`
+        ).bind(flag, ...ids, ...scope.params);
+
+    const mutationResult = await mutation.run();
+    if (!mutationResult.success) {
+        return { success: false, changes: 0, results: [] };
+    }
+
+    const { results } = await db.prepare(
+        `SELECT rm.id FROM raw_mails rm JOIN address a ON a.name = rm.address`
+        + ` WHERE ${targetWhere}`
+    ).bind(...ids, ...scope.params).all<{ id: number }>();
+
+    return {
+        success: true,
+        changes: mutationResult.meta.changes ?? 0,
+        results: results.map(row => ({ id: row.id, [resultField]: present })),
+    };
+};
+
+export const applyMailStateUpdate = async (
+    db: D1Database,
+    scope: MailScope,
+    value: unknown,
+) => {
+    const update = parseMailFlagUpdate(value);
+    if (!update) return null;
+
+    const definition = getMailStateDefinition(update.body.state);
+    if (definition?.unread === undefined) return null;
+    return await applyMailFlagUpdate(
+        db, scope, update.ids, MailFlag.UNREAD, definition.unread, 'unread'
+    );
+};
+
+export const applyMailFlaggedUpdate = async (
+    db: D1Database,
+    scope: MailScope,
+    value: unknown,
+) => {
+    const update = parseMailFlagUpdate(value);
+    if (!update || typeof update.body.flagged !== 'boolean') return null;
+    return await applyMailFlagUpdate(
+        db, scope, update.ids, MailFlag.FLAGGED, update.body.flagged, 'flagged'
+    );
+};
+
+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 (!isAnyMailFlagEnabled(env)) return [deleteMail];
+
+    return [
+        db.prepare(
+            `DELETE FROM mail_flags WHERE mail_id IN (`
+            + `SELECT id FROM raw_mails WHERE ${whereClause})`
+        ).bind(...params),
+        deleteMail,
+    ];
+};
+
+export const deleteRawMails = async (
+    db: D1Database,
+    env: Bindings,
+    whereClause: string,
+    params: (string | number)[],
+): Promise<D1Result> => {
+    const statements = prepareRawMailDeleteStatements(db, env, whereClause, params);
+    if (statements.length === 1) return await statements[0].run();
+
+    const results = await db.batch(statements);
+    return results[results.length - 1];
+};
+
+export const cleanupOrphanMailFlags = async (
+    db: D1Database,
+    env: Bindings,
+    limit = 1000,
+): Promise<number> => {
+    if (!isAnyMailFlagEnabled(env) || !Number.isInteger(limit) || limit <= 0) 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`
+        + ` LEFT JOIN address a ON a.id = mf.address_id AND a.name = rm.address`
+        + ` WHERE rm.id IS NULL OR a.id IS NULL LIMIT ?)`
+    ).bind(limit).run();
+    return result.meta.changes ?? 0;
+};

worker/src/mails_api/index.ts

@@ -27,7 +27,10 @@ api.post('/api/attachment/get_url', s3_attachment.getSignedGetUrl)
 
 // mail crud
 api.get('/api/mails', mails_crud.listMails)
+api.get('/api/mail-states', mails_crud.getMailStates)
 api.get('/api/mail/:mail_id', mails_crud.getMail)
+api.patch('/api/mails/state', mails_crud.updateMailState)
+api.patch('/api/mails/flagged', mails_crud.updateMailFlagged)
 api.delete('/api/mails/:id', mails_crud.deleteMail)
 
 // parsed mail (server-side parsed subject/text/html/attachments)

worker/src/mails_api/mails_crud.ts

@@ -5,18 +5,63 @@ import { getBooleanValue } from '../utils';
 import { handleMailListQuery, deleteAddressWithData, updateAddressUpdatedAt } from '../common'
 import { resolveRawEmailRow } from '../gzip'
 import { getSendBalanceState } from './send_balance';
+import {
+    getMailStateQuery,
+    getMailFlaggedQuery,
+    getMailStateOptions,
+    applyMailStateUpdate,
+    applyMailFlaggedUpdate,
+    serializeMailState,
+    deleteRawMails,
+    isMailReadStatusEnabled,
+    isMailFlaggedEnabled,
+} from '../mail_flags';
 
 const listMails = async (c: Context<HonoCustomType>) => {
     const { address } = c.get("jwtPayload")
     if (!address) {
         return c.json({ "error": "No address" }, 400)
     }
-    const { limit, offset } = c.req.query();
+    const { limit, offset, mail_state, flagged } = c.req.query();
     if (Number.parseInt(offset) <= 0) updateAddressUpdatedAt(c, address);
+    const stateQuery = getMailStateQuery(mail_state, 'rm', 'a.id');
+    const flaggedQuery = getMailFlaggedQuery(flagged, 'rm', 'a.id');
+    if (stateQuery === null) return c.json({ error: "Invalid mail state filter" }, 400);
+    if (flaggedQuery === null) return c.json({ error: "Invalid flagged filter" }, 400);
+    if (stateQuery && !isMailReadStatusEnabled(c.env)) {
+        return c.json({ error: "Mail read status is disabled" }, 403);
+    }
+    if (flaggedQuery && !isMailFlaggedEnabled(c.env)) {
+        return c.json({ error: "Flagged mail is disabled" }, 403);
+    }
+
+    if (!stateQuery && !flaggedQuery) {
+        return await handleMailListQuery(c,
+            `SELECT * FROM raw_mails WHERE address = ?`,
+            `SELECT count(*) as count FROM raw_mails WHERE address = ?`,
+            [address], limit, offset
+        );
+    }
+
+    const filters = [`rm.address = ?`];
+    if (stateQuery?.clause) filters.push(stateQuery.clause);
+    if (flaggedQuery?.clause) filters.push(flaggedQuery.clause);
+    const fromQuery = ` FROM raw_mails rm`
+        + ` JOIN address a ON a.name = rm.address`
+        + (stateQuery?.join ?? '')
+        + (flaggedQuery?.join ?? '')
+        + ` WHERE ${filters.join(' AND ')}`;
+    const unreadSelect = stateQuery?.unread === undefined
+        ? ''
+        : `, ${stateQuery.unread ? 1 : 0} AS unread`;
+    const flaggedSelect = flaggedQuery?.flagged === undefined
+        ? ''
+        : `, ${flaggedQuery.flagged ? 1 : 0} AS flagged`;
     return await handleMailListQuery(c,
-        `SELECT * FROM raw_mails where address = ?`,
-        `SELECT count(*) as count FROM raw_mails where address = ?`,
-        [address], limit, offset
+        `SELECT rm.*${unreadSelect}${flaggedSelect}${fromQuery}`,
+        `SELECT count(*) as count${fromQuery}`,
+        [...(stateQuery?.params ?? []), ...(flaggedQuery?.params ?? []), address], limit, offset,
+        flaggedQuery?.orderBy ?? stateQuery?.orderBy ?? 'rm.id desc'
     );
 };
 
@@ -27,7 +72,11 @@ 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(await resolveRawEmailRow(result));
+    return c.json(await serializeMailState(
+        c.env.DB,
+        await resolveRawEmailRow(result),
+        c.env,
+    ));
 };
 
 const deleteMail = async (c: Context<HonoCustomType>) => {
@@ -38,12 +87,52 @@ 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 });
 };
 
+const updateMailState = async (c: Context<HonoCustomType>) => {
+    if (!isMailReadStatusEnabled(c.env)) {
+        return c.json({ error: "Mail read status is disabled" }, 403);
+    }
+    const { address } = c.get("jwtPayload");
+    const result = await applyMailStateUpdate(
+        c.env.DB,
+        { clause: 'rm.address = ?', params: [address] },
+        await c.req.json().catch(() => null),
+    );
+    if (!result) return c.json({ error: "Invalid mail state request" }, 400);
+    if (!result.success) return c.json(result, 500);
+    return c.json(result);
+};
+
+const updateMailFlagged = async (c: Context<HonoCustomType>) => {
+    if (!isMailFlaggedEnabled(c.env)) {
+        return c.json({ error: "Flagged mail is disabled" }, 403);
+    }
+    const { address } = c.get("jwtPayload");
+    const result = await applyMailFlaggedUpdate(
+        c.env.DB,
+        { clause: 'rm.address = ?', params: [address] },
+        await c.req.json().catch(() => null),
+    );
+    if (!result) return c.json({ error: "Invalid flagged request" }, 400);
+    if (!result.success) return c.json(result, 500);
+    return c.json(result);
+};
+
+const getMailStates = (c: Context<HonoCustomType>) => {
+    if (!isMailReadStatusEnabled(c.env)) {
+        return c.json({ error: "Mail read status is disabled" }, 403);
+    }
+    return c.json({ results: getMailStateOptions() });
+};
+
 const getSettings = async (c: Context<HonoCustomType>) => {
     const { address, address_id } = c.get("jwtPayload")
     const msgs = i18n.getMessagesbyContext(c);
@@ -93,9 +182,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)
     }
@@ -117,4 +209,7 @@ const clearSentItems = async (c: Context<HonoCustomType>) => {
     return c.json({ success });
 };
 
-export default { listMails, getMail, deleteMail, getSettings, deleteAddress, clearInbox, clearSentItems };
+export default {
+    listMails, getMail, deleteMail, updateMailState, updateMailFlagged, getMailStates,
+    getSettings, deleteAddress, clearInbox, clearSentItems
+};

worker/src/models/index.ts

@@ -213,6 +213,8 @@ export type RawMailRow = {
     raw?: string;
     raw_blob?: unknown;
     metadata?: string;
+    unread?: boolean;
+    flagged?: boolean;
     created_at?: string;
 }
 

worker/src/types.d.ts

@@ -117,6 +117,8 @@ type Bindings = {
 
     // gzip compression for raw_mails
     ENABLE_MAIL_GZIP: string | boolean | undefined
+    ENABLE_MAIL_READ_STATUS: string | boolean | undefined
+    ENABLE_MAIL_FLAGGED: string | boolean | undefined
     CLEANUP_BATCH_SIZE: string | number | undefined
 
     // E2E testing

worker/src/user_api/index.ts

@@ -16,6 +16,9 @@ api.get('/user_api/settings', settings.settings);
 
 // mail api
 api.get('/user_api/mails', user_mail_api.getMails);
+api.get('/user_api/mail-states', user_mail_api.getMailStates);
+api.patch('/user_api/mails/state', user_mail_api.updateMailState);
+api.patch('/user_api/mails/flagged', user_mail_api.updateMailFlagged);
 api.delete('/user_api/mails/:id', user_mail_api.deleteMail);
 
 // send mail api

worker/src/user_api/user_mail_api.ts

@@ -2,25 +2,62 @@ import { Context } from "hono";
 import i18n from "../i18n";
 import { handleMailListQuery } from "../common";
 import { getBooleanValue } from "../utils";
+import {
+    getMailStateQuery,
+    getMailFlaggedQuery,
+    getMailStateOptions,
+    applyMailStateUpdate,
+    applyMailFlaggedUpdate,
+    deleteRawMails,
+    isMailReadStatusEnabled,
+    isMailFlaggedEnabled,
+} from "../mail_flags";
 
 export default {
+    getMailStates: (c: Context<HonoCustomType>) => {
+        if (!isMailReadStatusEnabled(c.env)) {
+            return c.json({ error: "Mail read status is disabled" }, 403);
+        }
+        return c.json({ results: getMailStateOptions() });
+    },
     getMails: async (c: Context<HonoCustomType>) => {
         const { user_id } = c.get("userPayload");
-        const { address, limit, offset } = c.req.query();
+        const { address, limit, offset, mail_state, flagged } = c.req.query();
         const filterQuerys = [`ua.user_id = ?`];
         const filterParams = [String(user_id)];
         if (address) {
             filterQuerys.push(`rm.address = ?`);
             filterParams.push(address);
         }
+        const stateQuery = getMailStateQuery(mail_state, 'rm', 'a.id');
+        const flaggedQuery = getMailFlaggedQuery(flagged, 'rm', 'a.id');
+        if (stateQuery === null) return c.json({ error: "Invalid mail state filter" }, 400);
+        if (flaggedQuery === null) return c.json({ error: "Invalid flagged filter" }, 400);
+        if (stateQuery && !isMailReadStatusEnabled(c.env)) {
+            return c.json({ error: "Mail read status is disabled" }, 403);
+        }
+        if (flaggedQuery && !isMailFlaggedEnabled(c.env)) {
+            return c.json({ error: "Flagged mail is disabled" }, 403);
+        }
+        if (stateQuery?.clause) filterQuerys.push(stateQuery.clause);
+        if (flaggedQuery?.clause) filterQuerys.push(flaggedQuery.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 ?? '')
+            + (flaggedQuery?.join ?? '')
             + ` WHERE ${filterQuerys.join(" AND ")}`;
+        const unreadSelect = stateQuery?.unread === undefined
+            ? ''
+            : `, ${stateQuery.unread ? 1 : 0} AS unread`;
+        const flaggedSelect = flaggedQuery?.flagged === undefined
+            ? ''
+            : `, ${flaggedQuery.flagged ? 1 : 0} AS flagged`;
         return await handleMailListQuery(c,
-            `SELECT rm.*${fromQuery}`,
+            `SELECT rm.*${unreadSelect}${flaggedSelect}${fromQuery}`,
             `SELECT count(*) as count${fromQuery}`,
-            filterParams, limit, offset, 'rm.id desc'
+            [...(stateQuery?.params ?? []), ...(flaggedQuery?.params ?? []), ...filterParams],
+            limit, offset, flaggedQuery?.orderBy ?? stateQuery?.orderBy ?? 'rm.id desc'
         );
     },
     deleteMail: async (c: Context<HonoCustomType>) => {
@@ -30,16 +67,57 @@ 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
         })
+    },
+    updateMailState: async (c: Context<HonoCustomType>) => {
+        if (!isMailReadStatusEnabled(c.env)) {
+            return c.json({ error: "Mail read status is disabled" }, 403);
+        }
+        const { user_id } = c.get("userPayload");
+        const result = await applyMailStateUpdate(
+            c.env.DB,
+            {
+                clause: `a.id IN (`
+                    + `SELECT address_id FROM users_address WHERE user_id = ?`
+                    + `)`,
+                params: [user_id],
+            },
+            await c.req.json().catch(() => null),
+        );
+        if (!result) return c.json({ error: "Invalid mail state request" }, 400);
+        if (!result.success) return c.json(result, 500);
+        return c.json(result);
+    },
+    updateMailFlagged: async (c: Context<HonoCustomType>) => {
+        if (!isMailFlaggedEnabled(c.env)) {
+            return c.json({ error: "Flagged mail is disabled" }, 403);
+        }
+        const { user_id } = c.get("userPayload");
+        const result = await applyMailFlaggedUpdate(
+            c.env.DB,
+            {
+                clause: `a.id IN (`
+                    + `SELECT address_id FROM users_address WHERE user_id = ?`
+                    + `)`,
+                params: [user_id],
+            },
+            await c.req.json().catch(() => null),
+        );
+        if (!result) return c.json({ error: "Invalid flagged request" }, 400);
+        if (!result.success) return c.json(result, 500);
+        return c.json(result);
     }
 }

worker/src/utils.ts

@@ -3,6 +3,7 @@ import { createMimeMessage } from "mimetext";
 import { UserSettings, RoleAddressConfig } from "./models";
 import { CONSTANTS } from "./constants";
 import { compressText } from "./gzip";
+import { initializeMailFlagsAfterInsert } from "./mail_flags";
 
 export const getJsonObjectValue = <T = any>(
     value: string | any
@@ -371,7 +372,7 @@ export const sendAdminInternalMail = async (
         });
         const message_id = Math.random().toString(36).substring(2, 15);
         const rawText = msg.asRaw();
-        let success = false;
+        let insertResult: D1Result | null = null;
         if (getBooleanValue(c.env.ENABLE_MAIL_GZIP)) {
             let compressed: ArrayBuffer | null = null;
             try {
@@ -381,34 +382,41 @@ export const sendAdminInternalMail = async (
             }
             if (compressed) {
                 try {
-                    ({ success } = await c.env.DB.prepare(
+                    insertResult = await c.env.DB.prepare(
                         `INSERT INTO raw_mails (source, address, raw_blob, message_id) VALUES (?, ?, ?, ?)`
-                    ).bind("admin@internal", toMail, compressed, message_id).run());
+                    ).bind("admin@internal", toMail, compressed, message_id).run();
                 } catch (dbError) {
                     const errMsg = String(dbError);
                     if (errMsg.includes('raw_blob') || errMsg.includes('no such column')) {
                         console.error("raw_blob column missing, falling back to plaintext", dbError);
-                        ({ success } = await c.env.DB.prepare(
+                        insertResult = await c.env.DB.prepare(
                             `INSERT INTO raw_mails (source, address, raw, message_id) VALUES (?, ?, ?, ?)`
-                        ).bind("admin@internal", toMail, rawText, message_id).run());
+                        ).bind("admin@internal", toMail, rawText, message_id).run();
                     } else {
                         throw dbError;
                     }
                 }
             } else {
-                ({ success } = await c.env.DB.prepare(
+                insertResult = await c.env.DB.prepare(
                     `INSERT INTO raw_mails (source, address, raw, message_id) VALUES (?, ?, ?, ?)`
-                ).bind("admin@internal", toMail, rawText, message_id).run());
+                ).bind("admin@internal", toMail, rawText, message_id).run();
             }
         } else {
-            ({ success } = await c.env.DB.prepare(
+            insertResult = await c.env.DB.prepare(
                 `INSERT INTO raw_mails (source, address, raw, message_id) VALUES (?, ?, ?, ?)`
-            ).bind("admin@internal", toMail, rawText, message_id).run());
+            ).bind("admin@internal", toMail, rawText, message_id).run();
         }
-        if (!success) {
+        if (!insertResult?.success) {
             console.log(`Failed save message from admin@internal to ${toMail}`);
+        } else {
+            await initializeMailFlagsAfterInsert(
+                c.env.DB,
+                c.env,
+                insertResult?.meta.last_row_id ?? 0,
+                toMail,
+            );
         }
-        return success;
+        return insertResult?.success ?? false;
     } catch (error) {
         console.log("sendAdminInternalMail error", error);
         return false;

worker/wrangler.toml.template

@@ -77,6 +77,10 @@ ENABLE_USER_CREATE_EMAIL = true
 # DISABLE_ANONYMOUS_USER_CREATE_EMAIL = true
 # Allow users to delete messages
 ENABLE_USER_DELETE_EMAIL = true
+# Enable per-message read status. This adds one write for each new mail and another when it is read.
+# ENABLE_MAIL_READ_STATUS = true
+# Enable low-write Flagged/starred mail independently from read status.
+# ENABLE_MAIL_FLAGGED = true
 # Allow automatic replies to emails
 ENABLE_AUTO_REPLY = false
 # Allow webhook