email: push new-mail to /email via SSE (IMAP IDLE -> sidecar -> server -> EventSource)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -68,6 +68,26 @@ export const EmailList = () => {
|
|||||||
}
|
}
|
||||||
}, [folder, data?.total, allCount]);
|
}, [folder, data?.total, allCount]);
|
||||||
|
|
||||||
|
// Live updates: while /email is open, listen for new-mail pushes (IMAP IDLE → SSE) and refetch.
|
||||||
|
// The EventSource closes automatically when this component unmounts (i.e. when you leave /email).
|
||||||
|
useEffect(() => {
|
||||||
|
const token = localStorage.getItem('BEARER_TOKEN');
|
||||||
|
if (!token) return;
|
||||||
|
const es = new EventSource(`/api/email/events?token=${encodeURIComponent(token)}`);
|
||||||
|
es.onmessage = (ev) => {
|
||||||
|
try {
|
||||||
|
if ((JSON.parse(ev.data) as { type?: string }).type === 'new-mail') {
|
||||||
|
queryClient.invalidateQueries({ queryKey: ['email-messages'] });
|
||||||
|
queryClient.invalidateQueries({ queryKey: ['email-messages-all-count'] });
|
||||||
|
queryClient.invalidateQueries({ queryKey: ['email-accounts'] });
|
||||||
|
}
|
||||||
|
} catch {
|
||||||
|
/* ignore non-JSON keepalives */
|
||||||
|
}
|
||||||
|
};
|
||||||
|
return () => es.close();
|
||||||
|
}, [queryClient]);
|
||||||
|
|
||||||
const handleFolderChange = (newFolder: string) => {
|
const handleFolderChange = (newFolder: string) => {
|
||||||
setFolder(newFolder);
|
setFolder(newFolder);
|
||||||
setPage(1);
|
setPage(1);
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ import { findEntryByProxyId, touchEntry } from './servers/api/dev-server/router'
|
|||||||
import officerWeb from './apps/officer-web/index.html';
|
import officerWeb from './apps/officer-web/index.html';
|
||||||
import { startBrowserRelay } from './servers/api/browser/relay';
|
import { startBrowserRelay } from './servers/api/browser/relay';
|
||||||
import { registerSidecar, unregisterSidecar, handleSidecarMessage } from './servers/sidecar-registry';
|
import { registerSidecar, unregisterSidecar, handleSidecarMessage } from './servers/sidecar-registry';
|
||||||
|
import { broadcastEmailNew } from './servers/api/email/email';
|
||||||
import type { SidecarRegistration } from './servers/sidecar/registration-protocol';
|
import type { SidecarRegistration } from './servers/sidecar/registration-protocol';
|
||||||
import { toShellUsername } from './servers/data-path';
|
import { toShellUsername } from './servers/data-path';
|
||||||
|
|
||||||
@@ -71,6 +72,12 @@ const sidecarWebsocket = {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Email IDLE watcher reporting new mail → push to that user's /email SSE clients.
|
||||||
|
if (msg.type === 'email:new' && typeof msg.userEmail === 'string') {
|
||||||
|
broadcastEmailNew(msg.userEmail);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
const id = sidecarConnections.get(ws);
|
const id = sidecarConnections.get(ws);
|
||||||
if (id) {
|
if (id) {
|
||||||
handleSidecarMessage(id, msg);
|
handleSidecarMessage(id, msg);
|
||||||
|
|||||||
@@ -10,6 +10,60 @@ export const emailRouter = createRouter();
|
|||||||
|
|
||||||
emailRouter.route('/accounts', accountsRouter);
|
emailRouter.route('/accounts', accountsRouter);
|
||||||
|
|
||||||
|
// ── Real-time: per-user SSE stream of email events (fed by the IMAP IDLE watcher) ──
|
||||||
|
const emailSseClients = new Map<string, Set<ReadableStreamDefaultController<Uint8Array>>>();
|
||||||
|
|
||||||
|
// Called from the sidecar-message handler when the IDLE watcher saves new mail for a user.
|
||||||
|
export function broadcastEmailNew(userEmail: string): void {
|
||||||
|
const set = emailSseClients.get(userEmail);
|
||||||
|
if (!set || set.size === 0) return;
|
||||||
|
const payload = new TextEncoder().encode(`data: ${JSON.stringify({ type: 'new-mail' })}\n\n`);
|
||||||
|
for (const ctrl of set) {
|
||||||
|
try {
|
||||||
|
ctrl.enqueue(payload);
|
||||||
|
} catch {
|
||||||
|
/* dead controller — cleaned up on cancel */
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// EventSource endpoint (auth via ?token= handled by userMiddleware). The frontend opens this only
|
||||||
|
// while on /email, so events stop the moment you navigate away.
|
||||||
|
emailRouter.get('/events', (ctx) => {
|
||||||
|
const email = ctx.get('user').email;
|
||||||
|
const enc = new TextEncoder();
|
||||||
|
let controllerRef: ReadableStreamDefaultController<Uint8Array>;
|
||||||
|
let ping: ReturnType<typeof setInterval>;
|
||||||
|
const stream = new ReadableStream<Uint8Array>({
|
||||||
|
start(controller) {
|
||||||
|
controllerRef = controller;
|
||||||
|
let set = emailSseClients.get(email);
|
||||||
|
if (!set) {
|
||||||
|
set = new Set();
|
||||||
|
emailSseClients.set(email, set);
|
||||||
|
}
|
||||||
|
set.add(controller);
|
||||||
|
controller.enqueue(enc.encode('retry: 3000\n\n'));
|
||||||
|
ping = setInterval(() => {
|
||||||
|
try {
|
||||||
|
controller.enqueue(enc.encode(': ping\n\n'));
|
||||||
|
} catch {
|
||||||
|
/* closed */
|
||||||
|
}
|
||||||
|
}, 25_000);
|
||||||
|
},
|
||||||
|
cancel() {
|
||||||
|
clearInterval(ping);
|
||||||
|
const set = emailSseClients.get(email);
|
||||||
|
set?.delete(controllerRef);
|
||||||
|
if (set && set.size === 0) emailSseClients.delete(email);
|
||||||
|
},
|
||||||
|
});
|
||||||
|
return new Response(stream, {
|
||||||
|
headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive' },
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
emailRouter.get('/messages', async (ctx) => {
|
emailRouter.get('/messages', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const email = ctx.get('user').email;
|
||||||
const page = Number(ctx.req.query('page') ?? '1');
|
const page = Number(ctx.req.query('page') ?? '1');
|
||||||
|
|||||||
@@ -25,6 +25,7 @@ const RECONCILE_MS = 5 * 60_000;
|
|||||||
|
|
||||||
const watchers = new Map<number, Watcher>();
|
const watchers = new Map<number, Watcher>();
|
||||||
let reconcileTimer: ReturnType<typeof setInterval> | null = null;
|
let reconcileTimer: ReturnType<typeof setInterval> | null = null;
|
||||||
|
let notifyNewMail: ((userEmail: string) => void) | null = null;
|
||||||
|
|
||||||
async function resolveAuth(account: Account): Promise<{ user: string; pass?: string; accessToken?: string } | null> {
|
async function resolveAuth(account: Account): Promise<{ user: string; pass?: string; accessToken?: string } | null> {
|
||||||
if (account.authType === 'oauth') {
|
if (account.authType === 'oauth') {
|
||||||
@@ -46,7 +47,10 @@ async function runSync(w: Watcher): Promise<void> {
|
|||||||
const user = await getUserById(w.account.userId);
|
const user = await getUserById(w.account.userId);
|
||||||
if (!user) return;
|
if (!user) return;
|
||||||
const result = await performResync({ accountId: w.account.id, userEmail: user.email, userId: user.id });
|
const result = await performResync({ accountId: w.account.id, userEmail: user.email, userId: user.id });
|
||||||
if (result.saved > 0) console.log(`[email-idle] ${w.account.email}: ${result.saved} new`);
|
if (result.saved > 0) {
|
||||||
|
console.log(`[email-idle] ${w.account.email}: ${result.saved} new`);
|
||||||
|
notifyNewMail?.(user.email);
|
||||||
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
console.error(`[email-idle] sync failed for ${w.account.email}:`, err instanceof Error ? err.message : err);
|
console.error(`[email-idle] sync failed for ${w.account.email}:`, err instanceof Error ? err.message : err);
|
||||||
} finally {
|
} finally {
|
||||||
@@ -141,8 +145,9 @@ async function reconcile(): Promise<void> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
export function initEmailIdle(): void {
|
export function initEmailIdle(onNewMail?: (userEmail: string) => void): void {
|
||||||
if (reconcileTimer) return;
|
if (reconcileTimer) return;
|
||||||
|
notifyNewMail = onNewMail ?? null;
|
||||||
console.log('[email-idle] starting IMAP IDLE watchers');
|
console.log('[email-idle] starting IMAP IDLE watchers');
|
||||||
setTimeout(() => void reconcile(), 5_000);
|
setTimeout(() => void reconcile(), 5_000);
|
||||||
reconcileTimer = setInterval(() => void reconcile(), RECONCILE_MS);
|
reconcileTimer = setInterval(() => void reconcile(), RECONCILE_MS);
|
||||||
|
|||||||
@@ -85,8 +85,9 @@ const connection = createSidecarConnector({
|
|||||||
onConnected() {
|
onConnected() {
|
||||||
// Start email cron once connected (so queue commands can reach API server)
|
// Start email cron once connected (so queue commands can reach API server)
|
||||||
initEmailCron();
|
initEmailCron();
|
||||||
// Real-time push via IMAP IDLE; the cron above is the slow backstop.
|
// Real-time push via IMAP IDLE; the cron above is the slow backstop. On new mail, tell the API
|
||||||
initEmailIdle();
|
// server so it can push an SSE event to that user's open /email page.
|
||||||
|
initEmailIdle((userEmail) => connection.send({ type: 'email:new', userEmail }));
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
@@ -50,6 +50,8 @@ export type SidecarEvent =
|
|||||||
| { type: 'vnc:stopped'; id: string }
|
| { type: 'vnc:stopped'; id: string }
|
||||||
| { type: 'vnc:status'; id: string; session: VncSessionInfo | null }
|
| { type: 'vnc:status'; id: string; session: VncSessionInfo | null }
|
||||||
| { type: 'vnc:error'; id: string; error: string }
|
| { type: 'vnc:error'; id: string; error: string }
|
||||||
|
// Email
|
||||||
|
| { type: 'email:new'; userEmail: string }
|
||||||
// Generic
|
// Generic
|
||||||
| { type: 'error'; id?: string; error: string };
|
| { type: 'error'; id?: string; error: string };
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user