email: store emails.db per account under email_accounts/<account>/
Reorganizes email storage: the DB moves from DATA_PATH/<user>/emails.db to DATA_PATH/<user>/email_accounts/<accountEmail>/emails.db, with a shared email_accounts/attachment_cache/ (was Gmail/emails/attachments). openEmailDb now takes (owner, account); a new openUserEmailDb(owner, userId) resolves the user's configured account (first enabled) for read paths. Threads the account through email.ts, accounts, resync, queue sync, channel handlers, and the email_db MCP tool path. Drops the dead getUserEmailDir helper. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -27,7 +27,8 @@ if (targetEmail) {
|
|||||||
function migrate(userEmail: string): void {
|
function migrate(userEmail: string): void {
|
||||||
console.log(`Migrating ${userEmail}...`);
|
console.log(`Migrating ${userEmail}...`);
|
||||||
const emailDir = join(DATA_PATH, userEmail, 'Gmail', 'emails');
|
const emailDir = join(DATA_PATH, userEmail, 'Gmail', 'emails');
|
||||||
const db = openEmailDb(userEmail);
|
// Obsolete one-off migration (old .eml-file store → SQLite); kept only to compile.
|
||||||
|
const db = openEmailDb(userEmail, userEmail);
|
||||||
|
|
||||||
let filenames: string[];
|
let filenames: string[];
|
||||||
try {
|
try {
|
||||||
|
|||||||
@@ -56,7 +56,7 @@ const GMAIL_PREFIX_RE = /^\[(?:Gmail|Google Mail)\]\//;
|
|||||||
const SKIP_SUFFIXES = new Set(['All Mail', 'Trash', 'Spam', 'Bin']);
|
const SKIP_SUFFIXES = new Set(['All Mail', 'Trash', 'Spam', 'Bin']);
|
||||||
|
|
||||||
const folders = await client.list();
|
const folders = await client.list();
|
||||||
const db = openEmailDb(userEmail);
|
const db = openEmailDb(userEmail, config.email as string);
|
||||||
|
|
||||||
let seeded = 0;
|
let seeded = 0;
|
||||||
|
|
||||||
|
|||||||
@@ -141,7 +141,7 @@ accountsRouter.post('/:id/sync', async (ctx) => {
|
|||||||
// Determine if this is a first sync or resync
|
// Determine if this is a first sync or resync
|
||||||
let isFirstSync = true;
|
let isFirstSync = true;
|
||||||
if (account.provider === 'gmail' && account.authType === 'oauth') {
|
if (account.provider === 'gmail' && account.authType === 'oauth') {
|
||||||
const db = openEmailDb(user.email);
|
const db = openEmailDb(user.email, account.email);
|
||||||
try {
|
try {
|
||||||
isFirstSync = !getSyncMeta(db, 'last_sync_at');
|
isFirstSync = !getSyncMeta(db, 'last_sync_at');
|
||||||
} finally {
|
} finally {
|
||||||
|
|||||||
@@ -1,8 +1,9 @@
|
|||||||
import { Database } from 'bun:sqlite';
|
import { Database } from 'bun:sqlite';
|
||||||
import { createHash } from 'node:crypto';
|
import { createHash } from 'node:crypto';
|
||||||
import { join } from 'node:path';
|
import { join, dirname } from 'node:path';
|
||||||
import { chmodSync } from 'node:fs';
|
import { chmodSync, mkdirSync } from 'node:fs';
|
||||||
import { DATA_PATH } from '@@/data-path';
|
import { getEmailDbPath } from '@@/data-path';
|
||||||
|
import { getEmailAccounts } from 'officerdb';
|
||||||
import type { EmailSummary } from 'types';
|
import type { EmailSummary } from 'types';
|
||||||
|
|
||||||
const SCHEMA_TABLES = `
|
const SCHEMA_TABLES = `
|
||||||
@@ -130,8 +131,9 @@ function fallbackThreadId(row: { id: string; subject: unknown; from_address: unk
|
|||||||
return `s:${norm}|${counterpart}`;
|
return `s:${norm}|${counterpart}`;
|
||||||
}
|
}
|
||||||
|
|
||||||
export function openEmailDb(email: string): Database {
|
export function openEmailDb(ownerEmail: string, accountEmail: string): Database {
|
||||||
const dbPath = join(DATA_PATH, email, 'emails.db');
|
const dbPath = getEmailDbPath(ownerEmail, accountEmail);
|
||||||
|
mkdirSync(dirname(dbPath), { recursive: true });
|
||||||
const db = new Database(dbPath, { create: true });
|
const db = new Database(dbPath, { create: true });
|
||||||
// The API and the email sidecar both open this file; wait out a concurrent writer (e.g. a resync
|
// The API and the email sidecar both open this file; wait out a concurrent writer (e.g. a resync
|
||||||
// or the one-time thread_id backfill) instead of failing immediately with "database is locked".
|
// or the one-time thread_id backfill) instead of failing immediately with "database is locked".
|
||||||
@@ -151,6 +153,17 @@ export function openEmailDb(email: string): Database {
|
|||||||
return db;
|
return db;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Open the email DB for a user's configured account. Emails are stored per account
|
||||||
|
* (email_accounts/<account>/emails.db); for now we use the user's first account. Returns null if the
|
||||||
|
* user has no email account configured yet.
|
||||||
|
*/
|
||||||
|
export async function openUserEmailDb(ownerEmail: string, userId: number): Promise<Database | null> {
|
||||||
|
const accounts = await getEmailAccounts(userId);
|
||||||
|
const account = accounts.find((a) => a.enabled) ?? accounts[0];
|
||||||
|
return account ? openEmailDb(ownerEmail, account.email) : null;
|
||||||
|
}
|
||||||
|
|
||||||
function migrate(db: Database): void {
|
function migrate(db: Database): void {
|
||||||
const cols = db.query('PRAGMA table_info(emails)').all() as Array<{ name: string }>;
|
const cols = db.query('PRAGMA table_info(emails)').all() as Array<{ name: string }>;
|
||||||
const colNames = new Set(cols.map((c) => c.name));
|
const colNames = new Set(cols.map((c) => c.name));
|
||||||
|
|||||||
@@ -4,9 +4,9 @@ import nodemailer from 'nodemailer';
|
|||||||
import type { EmailMessage } from 'types';
|
import type { EmailMessage } from 'types';
|
||||||
import { createRouter } from '../../create-router';
|
import { createRouter } from '../../create-router';
|
||||||
import * as errors from '@@/custom-errors';
|
import * as errors from '@@/custom-errors';
|
||||||
import { DATA_PATH } from '@@/data-path';
|
import { getEmailAttachmentCacheDir } from '@@/data-path';
|
||||||
import { getEmailAccounts } from 'officerdb';
|
import { getEmailAccounts } from 'officerdb';
|
||||||
import { openEmailDb, rowToSummary, getSyncMeta, searchEmails } from './email-db';
|
import { openEmailDb, openUserEmailDb, rowToSummary, getSyncMeta, searchEmails } from './email-db';
|
||||||
import { accountsRouter } from './accounts';
|
import { accountsRouter } from './accounts';
|
||||||
|
|
||||||
export const emailRouter = createRouter();
|
export const emailRouter = createRouter();
|
||||||
@@ -69,10 +69,11 @@ emailRouter.post('/send', async (ctx) => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
// GET /contacts?q= — address autocomplete from people you've received mail from, ranked by frequency.
|
// GET /contacts?q= — address autocomplete from people you've received mail from, ranked by frequency.
|
||||||
emailRouter.get('/contacts', (ctx) => {
|
emailRouter.get('/contacts', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
const like = `%${(ctx.req.query('q') ?? '').trim().toLowerCase()}%`;
|
const like = `%${(ctx.req.query('q') ?? '').trim().toLowerCase()}%`;
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.json([]);
|
||||||
try {
|
try {
|
||||||
const rows = db
|
const rows = db
|
||||||
.query(
|
.query(
|
||||||
@@ -146,14 +147,15 @@ emailRouter.get('/events', (ctx) => {
|
|||||||
|
|
||||||
// Full-text search across all mail (subject / sender / recipients / snippet / body), newest first.
|
// Full-text search across all mail (subject / sender / recipients / snippet / body), newest first.
|
||||||
emailRouter.get('/search', async (ctx) => {
|
emailRouter.get('/search', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
const q = (ctx.req.query('q') ?? '').trim();
|
const q = (ctx.req.query('q') ?? '').trim();
|
||||||
const page = Math.max(1, Number(ctx.req.query('page') ?? '1') || 1);
|
const page = Math.max(1, Number(ctx.req.query('page') ?? '1') || 1);
|
||||||
const limit = Math.min(100, Math.max(1, Number(ctx.req.query('limit') ?? '50') || 50));
|
const limit = Math.min(100, Math.max(1, Number(ctx.req.query('limit') ?? '50') || 50));
|
||||||
const offset = (page - 1) * limit;
|
const offset = (page - 1) * limit;
|
||||||
if (!q) return ctx.json({ messages: [], total: 0 });
|
if (!q) return ctx.json({ messages: [], total: 0 });
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.json({ messages: [], total: 0 });
|
||||||
try {
|
try {
|
||||||
const { rows, total } = searchEmails(db, q, limit, offset);
|
const { rows, total } = searchEmails(db, q, limit, offset);
|
||||||
return ctx.json({ messages: rows.map(rowToSummary), total });
|
return ctx.json({ messages: rows.map(rowToSummary), total });
|
||||||
@@ -163,7 +165,7 @@ emailRouter.get('/search', async (ctx) => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
emailRouter.get('/messages', async (ctx) => {
|
emailRouter.get('/messages', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
const page = Number(ctx.req.query('page') ?? '1');
|
const page = Number(ctx.req.query('page') ?? '1');
|
||||||
const limit = Number(ctx.req.query('limit') ?? '50');
|
const limit = Number(ctx.req.query('limit') ?? '50');
|
||||||
const folder = ctx.req.query('folder') ?? 'inbox';
|
const folder = ctx.req.query('folder') ?? 'inbox';
|
||||||
@@ -171,7 +173,8 @@ emailRouter.get('/messages', async (ctx) => {
|
|||||||
|
|
||||||
const folderWhere = folder === 'all' ? 'deleted = 0' : `deleted = 0 AND labels LIKE '%${folder}%'`;
|
const folderWhere = folder === 'all' ? 'deleted = 0' : `deleted = 0 AND labels LIKE '%${folder}%'`;
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.json({ messages: [], total: 0 });
|
||||||
try {
|
try {
|
||||||
// One row per conversation: the latest message in each thread within this folder, plus the
|
// One row per conversation: the latest message in each thread within this folder, plus the
|
||||||
// thread's message count and how many are unread. COALESCE guards any un-backfilled rows.
|
// thread's message count and how many are unread. COALESCE guards any un-backfilled rows.
|
||||||
@@ -225,10 +228,11 @@ function buildMessage(db: ReturnType<typeof openEmailDb>, row: Record<string, un
|
|||||||
}
|
}
|
||||||
|
|
||||||
emailRouter.get('/messages/:id', async (ctx) => {
|
emailRouter.get('/messages/:id', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
const id = ctx.req.param('id');
|
const id = ctx.req.param('id');
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.text('Not found', 404);
|
||||||
try {
|
try {
|
||||||
const row = db.query('SELECT * FROM emails WHERE id = ? AND deleted = 0').get(id) as Record<string, unknown> | null;
|
const row = db.query('SELECT * FROM emails WHERE id = ? AND deleted = 0').get(id) as Record<string, unknown> | null;
|
||||||
if (!row) return ctx.text('Not found', 404);
|
if (!row) return ctx.text('Not found', 404);
|
||||||
@@ -239,11 +243,12 @@ emailRouter.get('/messages/:id', async (ctx) => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
// GET /thread/:id — the full conversation containing message :id, oldest message first.
|
// GET /thread/:id — the full conversation containing message :id, oldest message first.
|
||||||
emailRouter.get('/thread/:id', (ctx) => {
|
emailRouter.get('/thread/:id', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
const id = ctx.req.param('id');
|
const id = ctx.req.param('id');
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.text('Not found', 404);
|
||||||
try {
|
try {
|
||||||
const head = db.query('SELECT thread_id, subject FROM emails WHERE id = ? AND deleted = 0').get(id) as
|
const head = db.query('SELECT thread_id, subject FROM emails WHERE id = ? AND deleted = 0').get(id) as
|
||||||
| { thread_id: string | null; subject: string }
|
| { thread_id: string | null; subject: string }
|
||||||
@@ -262,11 +267,12 @@ emailRouter.get('/thread/:id', (ctx) => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
// PATCH /thread/:id/read — mark every message in the conversation as read.
|
// PATCH /thread/:id/read — mark every message in the conversation as read.
|
||||||
emailRouter.patch('/thread/:id/read', (ctx) => {
|
emailRouter.patch('/thread/:id/read', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
const id = ctx.req.param('id');
|
const id = ctx.req.param('id');
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.text('Not found', 404);
|
||||||
try {
|
try {
|
||||||
const head = db.query('SELECT thread_id FROM emails WHERE id = ?').get(id) as { thread_id: string | null } | null;
|
const head = db.query('SELECT thread_id FROM emails WHERE id = ?').get(id) as { thread_id: string | null } | null;
|
||||||
if (!head) return ctx.text('Not found', 404);
|
if (!head) return ctx.text('Not found', 404);
|
||||||
@@ -279,11 +285,12 @@ emailRouter.patch('/thread/:id/read', (ctx) => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
emailRouter.post('/messages/:id/attachments/:index/extract', async (ctx) => {
|
emailRouter.post('/messages/:id/attachments/:index/extract', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
const id = ctx.req.param('id');
|
const id = ctx.req.param('id');
|
||||||
const index = Number(ctx.req.param('index'));
|
const index = Number(ctx.req.param('index'));
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.text('Attachment not found', 404);
|
||||||
try {
|
try {
|
||||||
const row = db.query('SELECT filename, content FROM attachments WHERE email_id = ? AND idx = ?').get(id, index) as {
|
const row = db.query('SELECT filename, content FROM attachments WHERE email_id = ? AND idx = ?').get(id, index) as {
|
||||||
filename: string;
|
filename: string;
|
||||||
@@ -292,7 +299,7 @@ emailRouter.post('/messages/:id/attachments/:index/extract', async (ctx) => {
|
|||||||
if (!row || !row.content) return ctx.text('Attachment not found', 404);
|
if (!row || !row.content) return ctx.text('Attachment not found', 404);
|
||||||
|
|
||||||
const fileName = row.filename ?? 'unknown';
|
const fileName = row.filename ?? 'unknown';
|
||||||
const attachDir = join(DATA_PATH, email, 'Gmail', 'emails', 'attachments');
|
const attachDir = getEmailAttachmentCacheDir(user.email);
|
||||||
const destPath = join(attachDir, fileName);
|
const destPath = join(attachDir, fileName);
|
||||||
|
|
||||||
const destFile = Bun.file(destPath);
|
const destFile = Bun.file(destPath);
|
||||||
@@ -302,7 +309,7 @@ emailRouter.post('/messages/:id/attachments/:index/extract', async (ctx) => {
|
|||||||
await Bun.write(destPath, binary);
|
await Bun.write(destPath, binary);
|
||||||
}
|
}
|
||||||
|
|
||||||
return ctx.json({ filePath: `Gmail/emails/attachments/${fileName}`, fileName, root: 'user-data' });
|
return ctx.json({ filePath: `email_accounts/attachment_cache/${fileName}`, fileName, root: 'user-data' });
|
||||||
} catch {
|
} catch {
|
||||||
return ctx.text('Failed to extract attachment', 500);
|
return ctx.text('Failed to extract attachment', 500);
|
||||||
} finally {
|
} finally {
|
||||||
@@ -311,10 +318,11 @@ emailRouter.post('/messages/:id/attachments/:index/extract', async (ctx) => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
emailRouter.patch('/messages/:id/read', async (ctx) => {
|
emailRouter.patch('/messages/:id/read', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
const id = ctx.req.param('id');
|
const id = ctx.req.param('id');
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.json({ ok: true });
|
||||||
try {
|
try {
|
||||||
db.run('UPDATE emails SET read = 1 WHERE id = ? AND read = 0', [id]);
|
db.run('UPDATE emails SET read = 1 WHERE id = ? AND read = 0', [id]);
|
||||||
return ctx.json({ ok: true });
|
return ctx.json({ ok: true });
|
||||||
@@ -324,10 +332,11 @@ emailRouter.patch('/messages/:id/read', async (ctx) => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
emailRouter.delete('/messages/:id', async (ctx) => {
|
emailRouter.delete('/messages/:id', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
const id = ctx.req.param('id');
|
const id = ctx.req.param('id');
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.text('Not found', 404);
|
||||||
try {
|
try {
|
||||||
const result = db.run('UPDATE emails SET deleted = 1 WHERE id = ? AND deleted = 0', [id]);
|
const result = db.run('UPDATE emails SET deleted = 1 WHERE id = ? AND deleted = 0', [id]);
|
||||||
if (result.changes === 0) return ctx.text('Not found', 404);
|
if (result.changes === 0) return ctx.text('Not found', 404);
|
||||||
@@ -338,9 +347,10 @@ emailRouter.delete('/messages/:id', async (ctx) => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
emailRouter.get('/sync-status', async (ctx) => {
|
emailRouter.get('/sync-status', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.json({ lastSyncAt: null });
|
||||||
try {
|
try {
|
||||||
const lastSyncAt = getSyncMeta(db, 'last_sync_at');
|
const lastSyncAt = getSyncMeta(db, 'last_sync_at');
|
||||||
return ctx.json({ lastSyncAt });
|
return ctx.json({ lastSyncAt });
|
||||||
@@ -350,12 +360,13 @@ emailRouter.get('/sync-status', async (ctx) => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
emailRouter.get('/stats', async (ctx) => {
|
emailRouter.get('/stats', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
const folder = ctx.req.query('folder') ?? 'inbox';
|
const folder = ctx.req.query('folder') ?? 'inbox';
|
||||||
|
|
||||||
const folderWhere = folder === 'all' ? 'deleted = 0' : `deleted = 0 AND labels LIKE '%${folder}%'`;
|
const folderWhere = folder === 'all' ? 'deleted = 0' : `deleted = 0 AND labels LIKE '%${folder}%'`;
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.json({ total: 0, byDomain: [], bySender: [] });
|
||||||
try {
|
try {
|
||||||
const total = (db.query(`SELECT COUNT(*) as count FROM emails WHERE ${folderWhere}`).get() as { count: number })
|
const total = (db.query(`SELECT COUNT(*) as count FROM emails WHERE ${folderWhere}`).get() as { count: number })
|
||||||
.count;
|
.count;
|
||||||
@@ -377,9 +388,10 @@ emailRouter.get('/stats', async (ctx) => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
emailRouter.get('/labels', async (ctx) => {
|
emailRouter.get('/labels', async (ctx) => {
|
||||||
const email = ctx.get('user').email;
|
const user = ctx.get('user');
|
||||||
|
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(user.email, user.id);
|
||||||
|
if (!db) return ctx.json({ labels: [] });
|
||||||
try {
|
try {
|
||||||
const rows = db.query('SELECT labels FROM emails WHERE deleted = 0 AND labels IS NOT NULL').all() as Array<{
|
const rows = db.query('SELECT labels FROM emails WHERE deleted = 0 AND labels IS NOT NULL').all() as Array<{
|
||||||
labels: string;
|
labels: string;
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ async function refreshCredentials(creds: GmailCredentials): Promise<GmailCredent
|
|||||||
return { ...creds, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt };
|
return { ...creds, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt };
|
||||||
}
|
}
|
||||||
|
|
||||||
async function gmailResync(userEmail: string): Promise<ResyncResult> {
|
async function gmailResync(userEmail: string, accountEmail: string): Promise<ResyncResult> {
|
||||||
let creds = await loadGmailCredentials(userEmail);
|
let creds = await loadGmailCredentials(userEmail);
|
||||||
if (!creds.accessToken) {
|
if (!creds.accessToken) {
|
||||||
throw new Error('OAuth not configured — connect Google in Settings → Integrations for resyncs');
|
throw new Error('OAuth not configured — connect Google in Settings → Integrations for resyncs');
|
||||||
@@ -48,7 +48,7 @@ async function gmailResync(userEmail: string): Promise<ResyncResult> {
|
|||||||
|
|
||||||
creds = await refreshCredentials(creds);
|
creds = await refreshCredentials(creds);
|
||||||
|
|
||||||
const db = openEmailDb(userEmail);
|
const db = openEmailDb(userEmail, accountEmail);
|
||||||
try {
|
try {
|
||||||
const result = await gmailApiSync({ creds, db });
|
const result = await gmailApiSync({ creds, db });
|
||||||
|
|
||||||
@@ -154,7 +154,7 @@ async function imapResync(account: ImapAccountInfo, userEmail: string): Promise<
|
|||||||
if (!freshAccount) throw new Error(`Email account ${account.id} not found`);
|
if (!freshAccount) throw new Error(`Email account ${account.id} not found`);
|
||||||
const syncMeta = (freshAccount.syncMeta ?? {}) as Record<string, unknown>;
|
const syncMeta = (freshAccount.syncMeta ?? {}) as Record<string, unknown>;
|
||||||
|
|
||||||
const db = openEmailDb(userEmail);
|
const db = openEmailDb(userEmail, account.email);
|
||||||
const existingIds = new Set<string>();
|
const existingIds = new Set<string>();
|
||||||
const rows = db.query('SELECT id FROM emails').all() as Array<{ id: string }>;
|
const rows = db.query('SELECT id FROM emails').all() as Array<{ id: string }>;
|
||||||
for (const row of rows) existingIds.add(row.id);
|
for (const row of rows) existingIds.add(row.id);
|
||||||
@@ -289,7 +289,7 @@ async function doResync({ accountId, userEmail, userId }: ResyncParams): Promise
|
|||||||
|
|
||||||
// Gmail API resync needs OAuth; a gmail account authed with an app password resyncs over IMAP.
|
// Gmail API resync needs OAuth; a gmail account authed with an app password resyncs over IMAP.
|
||||||
if (account.provider === 'gmail' && account.authType === 'oauth') {
|
if (account.provider === 'gmail' && account.authType === 'oauth') {
|
||||||
return gmailResync(userEmail);
|
return gmailResync(userEmail, account.email);
|
||||||
}
|
}
|
||||||
|
|
||||||
return imapResync(
|
return imapResync(
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ import { chunkMessage } from './chunker';
|
|||||||
import { listPiModels } from '@@/api/pi/list-models';
|
import { listPiModels } from '@@/api/pi/list-models';
|
||||||
import { enqueueJob } from '../../queue/init';
|
import { enqueueJob } from '../../queue/init';
|
||||||
import { readJob } from '@@/queue/storage';
|
import { readJob } from '@@/queue/storage';
|
||||||
import { openEmailDb } from '@@/api/email/email-db';
|
import { openUserEmailDb } from '@@/api/email/email-db';
|
||||||
import type { ModelInfo } from '@@/api/pi/types';
|
import type { ModelInfo } from '@@/api/pi/types';
|
||||||
import { toShellUsername } from '@@/data-path';
|
import { toShellUsername } from '@@/data-path';
|
||||||
|
|
||||||
@@ -45,15 +45,17 @@ type CommandContext = {
|
|||||||
};
|
};
|
||||||
|
|
||||||
async function handleEmailSync(ctx: CommandContext): Promise<void> {
|
async function handleEmailSync(ctx: CommandContext): Promise<void> {
|
||||||
const { channel, email } = ctx;
|
const { channel, email, userId } = ctx;
|
||||||
|
|
||||||
// Count emails before sync
|
// Count emails before sync
|
||||||
let countBefore = 0;
|
let countBefore = 0;
|
||||||
try {
|
try {
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(email, userId);
|
||||||
|
if (db) {
|
||||||
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
||||||
countBefore = row.count;
|
countBefore = row.count;
|
||||||
db.close();
|
db.close();
|
||||||
|
}
|
||||||
} catch {
|
} catch {
|
||||||
// DB might not exist yet
|
// DB might not exist yet
|
||||||
}
|
}
|
||||||
@@ -84,7 +86,8 @@ async function handleEmailSync(ctx: CommandContext): Promise<void> {
|
|||||||
|
|
||||||
// Count emails after sync and get newest ones
|
// Count emails after sync and get newest ones
|
||||||
try {
|
try {
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(email, userId);
|
||||||
|
if (!db) return;
|
||||||
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
||||||
const countAfter = row.count;
|
const countAfter = row.count;
|
||||||
const newCount = countAfter - countBefore;
|
const newCount = countAfter - countBefore;
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import { getTelegramBot } from './bot';
|
|||||||
import { listPiModels } from '@@/api/pi/list-models';
|
import { listPiModels } from '@@/api/pi/list-models';
|
||||||
import { enqueueJob } from '../../queue/init';
|
import { enqueueJob } from '../../queue/init';
|
||||||
import { readJob } from '@@/queue/storage';
|
import { readJob } from '@@/queue/storage';
|
||||||
import { openEmailDb } from '@@/api/email/email-db';
|
import { openUserEmailDb } from '@@/api/email/email-db';
|
||||||
import type { ModelInfo } from '@@/api/pi/types';
|
import type { ModelInfo } from '@@/api/pi/types';
|
||||||
import { toShellUsername } from '@@/data-path';
|
import { toShellUsername } from '@@/data-path';
|
||||||
|
|
||||||
@@ -46,14 +46,16 @@ type CommandContext = {
|
|||||||
};
|
};
|
||||||
|
|
||||||
async function handleEmailSync(ctx: CommandContext): Promise<void> {
|
async function handleEmailSync(ctx: CommandContext): Promise<void> {
|
||||||
const { send, email } = ctx;
|
const { send, email, userId } = ctx;
|
||||||
|
|
||||||
let countBefore = 0;
|
let countBefore = 0;
|
||||||
try {
|
try {
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(email, userId);
|
||||||
|
if (db) {
|
||||||
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
||||||
countBefore = row.count;
|
countBefore = row.count;
|
||||||
db.close();
|
db.close();
|
||||||
|
}
|
||||||
} catch {
|
} catch {
|
||||||
// DB might not exist yet
|
// DB might not exist yet
|
||||||
}
|
}
|
||||||
@@ -82,7 +84,8 @@ async function handleEmailSync(ctx: CommandContext): Promise<void> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(email, userId);
|
||||||
|
if (!db) return;
|
||||||
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
||||||
const countAfter = row.count;
|
const countAfter = row.count;
|
||||||
const newCount = countAfter - countBefore;
|
const newCount = countAfter - countBefore;
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ import { getWhatsAppClient } from './bot';
|
|||||||
import { listPiModels } from '@@/api/pi/list-models';
|
import { listPiModels } from '@@/api/pi/list-models';
|
||||||
import { enqueueJob } from '../../queue/init';
|
import { enqueueJob } from '../../queue/init';
|
||||||
import { readJob } from '@@/queue/storage';
|
import { readJob } from '@@/queue/storage';
|
||||||
import { openEmailDb } from '@@/api/email/email-db';
|
import { openUserEmailDb } from '@@/api/email/email-db';
|
||||||
import type { ModelInfo } from '@@/api/pi/types';
|
import type { ModelInfo } from '@@/api/pi/types';
|
||||||
import { toShellUsername } from '@@/data-path';
|
import { toShellUsername } from '@@/data-path';
|
||||||
|
|
||||||
@@ -50,14 +50,16 @@ type CommandContext = {
|
|||||||
};
|
};
|
||||||
|
|
||||||
async function handleEmailSync(ctx: CommandContext): Promise<void> {
|
async function handleEmailSync(ctx: CommandContext): Promise<void> {
|
||||||
const { send, email } = ctx;
|
const { send, email, userId } = ctx;
|
||||||
|
|
||||||
let countBefore = 0;
|
let countBefore = 0;
|
||||||
try {
|
try {
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(email, userId);
|
||||||
|
if (db) {
|
||||||
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
||||||
countBefore = row.count;
|
countBefore = row.count;
|
||||||
db.close();
|
db.close();
|
||||||
|
}
|
||||||
} catch {
|
} catch {
|
||||||
// DB might not exist yet
|
// DB might not exist yet
|
||||||
}
|
}
|
||||||
@@ -86,7 +88,8 @@ async function handleEmailSync(ctx: CommandContext): Promise<void> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const db = openEmailDb(email);
|
const db = await openUserEmailDb(email, userId);
|
||||||
|
if (!db) return;
|
||||||
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
|
||||||
const countAfter = row.count;
|
const countAfter = row.count;
|
||||||
const newCount = countAfter - countBefore;
|
const newCount = countAfter - countBefore;
|
||||||
|
|||||||
@@ -52,7 +52,12 @@ export const getTmpAttachmentsDir = (email: string) => join(DATA_PATH, email, 'a
|
|||||||
|
|
||||||
export const getAttachmentsDir = (email: string, sessionId: string) => join(DATA_PATH, email, 'attachments', sessionId);
|
export const getAttachmentsDir = (email: string, sessionId: string) => join(DATA_PATH, email, 'attachments', sessionId);
|
||||||
|
|
||||||
export const getUserEmailDir = (email: string) => join(DATA_PATH, email, 'Gmail', 'emails');
|
// Per-account email storage: DATA_PATH/<owner>/email_accounts/<accountEmail>/emails.db, with a
|
||||||
|
// single attachment_cache/ shared across the owner's accounts.
|
||||||
|
export const getEmailAccountsDir = (ownerEmail: string) => join(DATA_PATH, ownerEmail, 'email_accounts');
|
||||||
|
export const getEmailDbPath = (ownerEmail: string, accountEmail: string) =>
|
||||||
|
join(getEmailAccountsDir(ownerEmail), accountEmail, 'emails.db');
|
||||||
|
export const getEmailAttachmentCacheDir = (ownerEmail: string) => join(getEmailAccountsDir(ownerEmail), 'attachment_cache');
|
||||||
|
|
||||||
/** Derive a valid Linux username from a display username or email. */
|
/** Derive a valid Linux username from a display username or email. */
|
||||||
export const toShellUsername = (username: string, email: string): string => {
|
export const toShellUsername = (username: string, email: string): string => {
|
||||||
|
|||||||
@@ -139,7 +139,7 @@ const emailSyncHandler: JobHandler = {
|
|||||||
let errors = 0;
|
let errors = 0;
|
||||||
let allDone = false;
|
let allDone = false;
|
||||||
|
|
||||||
const db = openEmailDb(userEmail);
|
const db = openEmailDb(userEmail, account.email);
|
||||||
|
|
||||||
// Load existing IDs for dedup (once, shared across reconnections)
|
// Load existing IDs for dedup (once, shared across reconnections)
|
||||||
const existingIds = new Set<string>();
|
const existingIds = new Set<string>();
|
||||||
|
|||||||
@@ -556,9 +556,10 @@ const gmailSyncHandler: JobHandler = {
|
|||||||
run: async (ctx) => {
|
run: async (ctx) => {
|
||||||
const creds = await loadGmailCredentials(ctx.job.userId);
|
const creds = await loadGmailCredentials(ctx.job.userId);
|
||||||
const emailAccountId = (ctx.meta as Record<string, unknown>).emailAccountId as number | undefined;
|
const emailAccountId = (ctx.meta as Record<string, unknown>).emailAccountId as number | undefined;
|
||||||
|
const syncAccount = emailAccountId ? await getEmailAccount(emailAccountId) : undefined;
|
||||||
|
|
||||||
// Determine sync mode — check SQLite first, fall back to PostgreSQL syncMeta
|
// Determine sync mode — check SQLite first, fall back to PostgreSQL syncMeta
|
||||||
const db = openEmailDb(ctx.job.userId);
|
const db = openEmailDb(ctx.job.userId, syncAccount?.email ?? ctx.job.userId);
|
||||||
let lastSyncAt: string | null = null;
|
let lastSyncAt: string | null = null;
|
||||||
try {
|
try {
|
||||||
lastSyncAt = getSyncMeta(db, 'last_sync_at');
|
lastSyncAt = getSyncMeta(db, 'last_sync_at');
|
||||||
@@ -622,7 +623,9 @@ const gmailSyncHandler: JobHandler = {
|
|||||||
run: async (ctx) => {
|
run: async (ctx) => {
|
||||||
const creds = ctx.meta.creds as GmailCredentials;
|
const creds = ctx.meta.creds as GmailCredentials;
|
||||||
|
|
||||||
const db = openEmailDb(ctx.job.userId);
|
const emailAccountId = (ctx.meta as Record<string, unknown>).emailAccountId as number | undefined;
|
||||||
|
const syncAccount = emailAccountId ? await getEmailAccount(emailAccountId) : undefined;
|
||||||
|
const db = openEmailDb(ctx.job.userId, syncAccount?.email ?? ctx.job.userId);
|
||||||
try {
|
try {
|
||||||
let result: SyncResult;
|
let result: SyncResult;
|
||||||
|
|
||||||
|
|||||||
@@ -8,7 +8,7 @@ import { SANDBOX_DATA } from '../sandbox';
|
|||||||
import * as claudeManager from './claude-manager';
|
import * as claudeManager from './claude-manager';
|
||||||
import { createSidecarConnector } from '../connect';
|
import { createSidecarConnector } from '../connect';
|
||||||
import { sign } from '../../jwt';
|
import { sign } from '../../jwt';
|
||||||
import { getUserByEmail } from 'officerdb';
|
import { getUserByEmail, getEmailAccounts } from 'officerdb';
|
||||||
|
|
||||||
const email = process.env.CLAUDE_USER_EMAIL;
|
const email = process.env.CLAUDE_USER_EMAIL;
|
||||||
if (!email) {
|
if (!email) {
|
||||||
@@ -38,6 +38,10 @@ const OFFICER_AUTH_TOKEN = await sign(
|
|||||||
const homeDir =
|
const homeDir =
|
||||||
dbUser.role === 'Super Admin' ? (process.env.HOME_DIR ?? homedir()) : join(DATA_PATH, email, 'home');
|
dbUser.role === 'Super Admin' ? (process.env.HOME_DIR ?? homedir()) : join(DATA_PATH, email, 'home');
|
||||||
const globalToolsDir = join(DATA_PATH, 'tools');
|
const globalToolsDir = join(DATA_PATH, 'tools');
|
||||||
|
|
||||||
|
// The email_db MCP tool reads one account's SQLite store; use the user's first configured account.
|
||||||
|
const emailAccounts = await getEmailAccounts(dbUser.id);
|
||||||
|
const emailDbRel = join('email_accounts', emailAccounts[0]?.email ?? 'none', 'emails.db');
|
||||||
const userToolsDir = join(DATA_PATH, email, 'tools');
|
const userToolsDir = join(DATA_PATH, email, 'tools');
|
||||||
|
|
||||||
// ── Path setup ──
|
// ── Path setup ──
|
||||||
@@ -90,7 +94,7 @@ function generateMcpConfig(): McpPaths {
|
|||||||
args: ['run', MCP_SERVER_SCRIPT],
|
args: ['run', MCP_SERVER_SCRIPT],
|
||||||
env: {
|
env: {
|
||||||
PI_TOOLS_DIRS: sandboxToolsDirs,
|
PI_TOOLS_DIRS: sandboxToolsDirs,
|
||||||
OFFICER_EMAIL_DB: `${SANDBOX_DATA}/emails.db`,
|
OFFICER_EMAIL_DB: `${SANDBOX_DATA}/${emailDbRel}`,
|
||||||
MCP_TOOLS_LOG: `${SANDBOX_DATA}/logs/mcp-tools.log`,
|
MCP_TOOLS_LOG: `${SANDBOX_DATA}/logs/mcp-tools.log`,
|
||||||
OFFICER_API_URL,
|
OFFICER_API_URL,
|
||||||
OFFICER_AUTH_TOKEN,
|
OFFICER_AUTH_TOKEN,
|
||||||
@@ -110,7 +114,7 @@ function generateMcpConfig(): McpPaths {
|
|||||||
args: ['run', MCP_SERVER_SCRIPT],
|
args: ['run', MCP_SERVER_SCRIPT],
|
||||||
env: {
|
env: {
|
||||||
PI_TOOLS_DIRS: hostToolsDirs,
|
PI_TOOLS_DIRS: hostToolsDirs,
|
||||||
OFFICER_EMAIL_DB: join(userRoot, 'emails.db'),
|
OFFICER_EMAIL_DB: join(userRoot, emailDbRel),
|
||||||
MCP_TOOLS_LOG: join(userRoot, 'logs', 'mcp-tools.log'),
|
MCP_TOOLS_LOG: join(userRoot, 'logs', 'mcp-tools.log'),
|
||||||
OFFICER_API_URL,
|
OFFICER_API_URL,
|
||||||
OFFICER_AUTH_TOKEN,
|
OFFICER_AUTH_TOKEN,
|
||||||
|
|||||||
Reference in New Issue
Block a user