opengraph stuff

src/servers/queue/engine.ts

@@ -0,0 +1,217 @@
+import type { Job, JobProgress, EnqueueParams, StepContext } from './types';
+import { readJob, writeJob, listAllJobs } from './storage';
+import { getHandler } from './handler-registry';
+import { sendMail } from 'emailer';
+
+const activeLanes = new Map<string, boolean>();
+
+const PROGRESS_THROTTLE_MS = 1000;
+
+export async function enqueue(params: EnqueueParams): Promise<Job> {
+  const handler = getHandler(params.type);
+  if (!handler) throw new Error(`No handler registered for job type: ${params.type}`);
+
+  const job: Job = {
+    id: crypto.randomUUID(),
+    lane: params.lane,
+    type: params.type,
+    userId: params.userId,
+    status: 'queued',
+    steps: handler.steps.map((s) => ({ name: s.name, status: 'pending' as const })),
+    currentStep: 0,
+    createdAt: Date.now(),
+    meta: params.meta,
+  };
+
+  await writeJob(job);
+  console.log(`[queue] Enqueued job ${job.id} (${job.type}) in lane ${job.lane}`);
+  kickLane(job.lane);
+  return job;
+}
+
+export async function cancelJob(id: string): Promise<Job | null> {
+  const job = await readJob(id);
+  if (!job) return null;
+  if (job.status === 'completed' || job.status === 'failed' || job.status === 'cancelled') return job;
+
+  job.status = 'cancelled';
+  job.completedAt = Date.now();
+  for (const step of job.steps) {
+    if (step.status === 'pending' || step.status === 'running') {
+      step.status = 'failed';
+      step.error = 'Cancelled';
+    }
+  }
+  await writeJob(job);
+  console.log(`[queue] Cancelled job ${job.id}`);
+  return job;
+}
+
+export async function resumeInterruptedJobs() {
+  const jobs = await listAllJobs();
+  const lanesToKick = new Set<string>();
+
+  for (const job of jobs) {
+    if (job.status === 'running') {
+      job.status = 'queued';
+      job.startedAt = undefined;
+      for (const step of job.steps) {
+        if (step.status === 'running') {
+          step.status = 'pending';
+          step.startedAt = undefined;
+        }
+      }
+      await writeJob(job);
+      console.log(`[queue] Reset interrupted job ${job.id} back to queued`);
+      lanesToKick.add(job.lane);
+    } else if (job.status === 'queued') {
+      lanesToKick.add(job.lane);
+    }
+  }
+
+  for (const lane of lanesToKick) {
+    kickLane(lane);
+  }
+}
+
+function kickLane(lane: string) {
+  if (activeLanes.get(lane)) return;
+  activeLanes.set(lane, true);
+  processNextInLane(lane);
+}
+
+async function processNextInLane(lane: string) {
+  try {
+    const jobs = await listAllJobs();
+    const next = jobs
+      .filter((j) => j.lane === lane && j.status === 'queued')
+      .sort((a, b) => a.createdAt - b.createdAt)[0];
+
+    if (!next) {
+      activeLanes.set(lane, false);
+      return;
+    }
+
+    await runJob(next);
+  } catch (err) {
+    console.error(`[queue] Lane ${lane} processing error:`, err);
+  } finally {
+    const jobs = await listAllJobs();
+    const hasMore = jobs.some((j) => j.lane === lane && j.status === 'queued');
+    if (hasMore) {
+      processNextInLane(lane);
+    } else {
+      activeLanes.set(lane, false);
+    }
+  }
+}
+
+async function runJob(job: Job) {
+  const handler = getHandler(job.type);
+  if (!handler) {
+    job.status = 'failed';
+    job.error = `No handler for type: ${job.type}`;
+    job.completedAt = Date.now();
+    await writeJob(job);
+    return;
+  }
+
+  job.status = 'running';
+  job.startedAt = Date.now();
+  await writeJob(job);
+  console.log(`[queue] Running job ${job.id} (${job.type})`);
+
+  const sharedMeta: Record<string, unknown> = { ...(job.meta ?? {}) };
+
+  for (let i = 0; i < handler.steps.length; i++) {
+    const fresh = await readJob(job.id);
+    if (!fresh || fresh.status === 'cancelled') {
+      console.log(`[queue] Job ${job.id} was cancelled, stopping`);
+      return;
+    }
+
+    const handlerStep = handler.steps[i]!;
+    const step = fresh.steps[i]!;
+
+    fresh.currentStep = i;
+    step.status = 'running';
+    step.startedAt = Date.now();
+    await writeJob(fresh);
+    console.log(`[queue] Job ${fresh.id} step ${i + 1}/${handler.steps.length}: "${handlerStep.name}"`);
+
+    let lastProgressWrite = 0;
+    let pendingProgress: JobProgress | null = null;
+
+    const updateProgress = async (progress: JobProgress) => {
+      step.progress = progress;
+      const now = Date.now();
+      if (now - lastProgressWrite >= PROGRESS_THROTTLE_MS) {
+        lastProgressWrite = now;
+        pendingProgress = null;
+        await writeJob(fresh);
+      } else {
+        pendingProgress = progress;
+      }
+    };
+
+    const ctx: StepContext = { job: fresh, step, updateProgress, meta: sharedMeta };
+
+    try {
+      await handlerStep.run(ctx);
+
+      if (pendingProgress) {
+        step.progress = pendingProgress;
+      }
+      step.status = 'completed';
+      step.completedAt = Date.now();
+      await writeJob(fresh);
+    } catch (err) {
+      const errorMessage = err instanceof Error ? err.message : String(err);
+      step.status = 'failed';
+      step.error = errorMessage;
+      step.completedAt = Date.now();
+      fresh.status = 'failed';
+      fresh.error = `Step "${step.name}" failed: ${errorMessage}`;
+      fresh.completedAt = Date.now();
+      await writeJob(fresh);
+      console.error(`[queue] Job ${fresh.id} failed at step "${step.name}":`, errorMessage);
+      await notifyFailure(fresh);
+      return;
+    }
+  }
+
+  const final = await readJob(job.id);
+  if (final && final.status === 'running') {
+    final.status = 'completed';
+    final.completedAt = Date.now();
+    await writeJob(final);
+    console.log(`[queue] Job ${final.id} completed`);
+    await notifyCompletion(final);
+  }
+}
+
+async function notifyCompletion(job: Job) {
+  try {
+    await sendMail({
+      template: 'JobCompleted',
+      subject: `Job completed: ${job.type}`,
+      to: job.userId,
+      data: { job },
+    });
+  } catch {
+    // SMTP might not be configured — non-fatal
+  }
+}
+
+async function notifyFailure(job: Job) {
+  try {
+    await sendMail({
+      template: 'JobFailed',
+      subject: `Job failed: ${job.type}`,
+      to: job.userId,
+      data: { job },
+    });
+  } catch {
+    // SMTP might not be configured — non-fatal
+  }
+}

src/servers/queue/handler-registry.ts

@@ -0,0 +1,11 @@
+import type { JobHandler } from './types';
+
+const handlers = new Map<string, JobHandler>();
+
+export function registerHandler(handler: JobHandler) {
+  handlers.set(handler.type, handler);
+}
+
+export function getHandler(type: string): JobHandler | undefined {
+  return handlers.get(type);
+}

src/servers/queue/handlers/gmail-sync.ts

@@ -0,0 +1,271 @@
+import { mkdirSync, readdirSync, writeFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { homedir } from 'node:os';
+import type { JobHandler } from '../types';
+import { registerHandler } from '../handler-registry';
+import { DATA_PATH } from '../../data-path';
+
+type GoogleCredentials = {
+  accessToken: string;
+  refreshToken: string;
+  expiresAt: number;
+  clientId: string;
+  clientSecret: string;
+};
+
+const googleConfigPath = join(homedir(), '.config', 'officer.dev', 'google-oauth.json');
+
+async function loadCredentials(userId: string): Promise<GoogleCredentials> {
+  const config = await Bun.file(googleConfigPath).json().catch(() => null);
+  if (!config?.clientId || !config?.clientSecret) {
+    throw new Error('Google OAuth not configured — ask your admin to set up credentials');
+  }
+
+  const tokenPath = join(DATA_PATH, userId, 'integrations', 'google.json');
+  const token = await Bun.file(tokenPath).json().catch(() => null);
+  if (!token?.accessToken) {
+    throw new Error('Google account not connected — connect in Settings → Integrations');
+  }
+
+  return {
+    accessToken: token.accessToken,
+    refreshToken: token.refreshToken ?? '',
+    expiresAt: token.expiresAt ?? 0,
+    clientId: config.clientId,
+    clientSecret: config.clientSecret,
+  };
+}
+
+let cachedAccessToken: string | null = null;
+let cachedExpiresAt = 0;
+
+async function getValidAccessToken(creds: GoogleCredentials): Promise<string> {
+  if (cachedAccessToken && cachedExpiresAt > Date.now() + 5 * 60 * 1000) {
+    return cachedAccessToken;
+  }
+
+  if (creds.expiresAt > Date.now() + 5 * 60 * 1000) {
+    cachedAccessToken = creds.accessToken;
+    cachedExpiresAt = creds.expiresAt;
+    return creds.accessToken;
+  }
+
+  if (!creds.refreshToken) throw new Error('Token expired and no refresh token available');
+
+  const res = await fetch('https://oauth2.googleapis.com/token', {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
+    body: new URLSearchParams({
+      client_id: creds.clientId,
+      client_secret: creds.clientSecret,
+      refresh_token: creds.refreshToken,
+      grant_type: 'refresh_token',
+    }),
+  });
+
+  if (!res.ok) {
+    const error = await res.text().catch(() => '');
+    throw new Error(`Token refresh failed (${res.status}): ${error}`);
+  }
+
+  const data = (await res.json()) as { access_token: string; expires_in?: number };
+  cachedAccessToken = data.access_token;
+  cachedExpiresAt = Date.now() + (data.expires_in ?? 3600) * 1000;
+  return data.access_token;
+}
+
+const GMAIL_BASE = 'https://gmail.googleapis.com/gmail/v1/users/me';
+
+async function gmailGet(token: string, path: string, params?: Record<string, string>): Promise<unknown> {
+  const url = new URL(`${GMAIL_BASE}${path}`);
+  if (params) {
+    for (const [k, v] of Object.entries(params)) {
+      if (v) url.searchParams.set(k, v);
+    }
+  }
+
+  const res = await fetch(url.toString(), {
+    headers: { Authorization: `Bearer ${token}` },
+  });
+
+  if (!res.ok) {
+    const error = await res.text().catch(() => '');
+    throw new Error(`Gmail API error (${res.status}): ${error}`);
+  }
+
+  return res.json();
+}
+
+function slugify(text: string, maxLen = 60): string {
+  return text
+    .toLowerCase()
+    .replace(/[^a-z0-9]+/g, '-')
+    .replace(/^-+|-+$/g, '')
+    .slice(0, maxLen)
+    .replace(/-+$/, '');
+}
+
+function buildEmlFilename(id: string, internalDate: string | undefined, rawEmail: string): string {
+  const subjectMatch = rawEmail.match(/^Subject:\s*(.+)$/mi);
+  const subject = subjectMatch?.[1]?.trim() || 'no-subject';
+  const ts = parseInt(internalDate || '0');
+  const d = new Date(ts);
+  const dateStr =
+    ts > 0
+      ? `${d.getFullYear()}-${String(d.getMonth() + 1).padStart(2, '0')}-${String(d.getDate()).padStart(2, '0')}`
+      : 'unknown-date';
+  return `${dateStr}_${slugify(subject)}_${id}.eml`;
+}
+
+type SyncProgress = { saved: number; skipped: number; errors: number; page: number };
+type OnProgress = (progress: SyncProgress) => void;
+
+async function syncInbox(
+  token: string,
+  outputDir: string,
+  query?: string,
+  onProgress?: OnProgress,
+): Promise<{ saved: number; skipped: number; errors: number }> {
+  mkdirSync(outputDir, { recursive: true });
+
+  const existingIds = new Set<string>();
+  try {
+    for (const file of readdirSync(outputDir)) {
+      const match = file.match(/_([a-f0-9]+)\.eml$/i);
+      if (match) existingIds.add(match[1]!);
+    }
+  } catch {
+    /* dir might not exist yet */
+  }
+
+  let saved = 0;
+  let skipped = 0;
+  let errors = 0;
+  let page = 0;
+  let pageToken: string | undefined;
+
+  do {
+    const params: Record<string, string> = { maxResults: '100' };
+    if (query) params.q = query;
+    if (pageToken) params.pageToken = pageToken;
+
+    const list = (await gmailGet(token, '/messages', params)) as {
+      messages?: Array<{ id: string }>;
+      nextPageToken?: string;
+    };
+
+    const messages = list.messages ?? [];
+    if (messages.length === 0) break;
+
+    for (let i = 0; i < messages.length; i += 5) {
+      const batch = messages.slice(i, i + 5);
+      await Promise.all(
+        batch.map(async ({ id }) => {
+          if (existingIds.has(id)) {
+            skipped++;
+            return;
+          }
+          try {
+            const msg = (await gmailGet(token, `/messages/${id}`, { format: 'raw' })) as {
+              id: string;
+              internalDate?: string;
+              raw: string;
+            };
+            const rawEmail = Buffer.from(msg.raw, 'base64url').toString('utf-8');
+            const filename = buildEmlFilename(msg.id, msg.internalDate, rawEmail);
+            writeFileSync(join(outputDir, filename), rawEmail);
+            existingIds.add(id);
+            saved++;
+          } catch {
+            errors++;
+          }
+        }),
+      );
+    }
+
+    page++;
+    console.log(`[gmail-sync] Page ${page}: saved ${saved}, skipped ${skipped}, errors ${errors}`);
+    onProgress?.({ saved, skipped, errors, page });
+
+    pageToken = list.nextPageToken;
+  } while (pageToken);
+
+  return { saved, skipped, errors };
+}
+
+const MONTH_NAMES = ['January', 'February', 'March', 'April', 'May', 'June', 'July', 'August', 'September', 'October', 'November', 'December'];
+
+function buildMonthRanges(year: number): Array<{ label: string; after: string; before: string }> {
+  const now = new Date();
+  const currentMonth = now.getFullYear() === year ? now.getMonth() : 11;
+  const ranges: Array<{ label: string; after: string; before: string }> = [];
+
+  for (let m = 0; m <= currentMonth; m++) {
+    const after = `${year}/${m + 1}/1`;
+    const before = m < 11 ? `${year}/${m + 2}/1` : `${year + 1}/1/1`;
+    ranges.push({ label: `${MONTH_NAMES[m]!} ${year}`, after, before });
+  }
+
+  return ranges;
+}
+
+const gmailSyncHandler: JobHandler = {
+  type: 'gmail-sync',
+  steps: [
+    {
+      name: 'Verify connection',
+      run: async (ctx) => {
+        const creds = await loadCredentials(ctx.job.userId);
+        const token = await getValidAccessToken(creds);
+        // Store token in shared meta for the next step
+        ctx.meta.accessToken = token;
+        ctx.meta.outputDir = join(DATA_PATH, ctx.job.userId, 'Gmail', 'emails');
+      },
+    },
+    {
+      name: 'Sync emails',
+      run: async (ctx) => {
+        const token = ctx.meta.accessToken as string;
+        const outputDir = ctx.meta.outputDir as string;
+        const year = ctx.meta.year as number | undefined;
+
+        let totalSaved = 0;
+        let totalSkipped = 0;
+        let totalErrors = 0;
+
+        if (year) {
+          // Year-scoped sync: month by month with progress
+          const months = buildMonthRanges(year);
+          for (let i = 0; i < months.length; i++) {
+            const month = months[i]!;
+            await ctx.updateProgress({ current: i, total: months.length, label: month.label });
+            const query = `after:${month.after} before:${month.before}`;
+            const { saved, skipped, errors } = await syncInbox(token, outputDir, query, (p) => {
+              const label = `${month.label} — ${p.saved} saved`;
+              ctx.updateProgress({ current: i, total: months.length, label });
+            });
+            totalSaved += saved;
+            totalSkipped += skipped;
+            totalErrors += errors;
+          }
+          await ctx.updateProgress({ current: months.length, total: months.length, label: 'Done' });
+        } else {
+          // Full sync: all emails with per-page progress
+          await ctx.updateProgress({ current: 0, total: 0, label: 'Starting sync' });
+          const { saved, skipped, errors } = await syncInbox(token, outputDir, undefined, (p) => {
+            const label = `Saved ${p.saved}, skipped ${p.skipped} (page ${p.page})`;
+            ctx.updateProgress({ current: p.saved + p.skipped + p.errors, total: 0, label });
+          });
+          totalSaved = saved;
+          totalSkipped = skipped;
+          totalErrors = errors;
+          await ctx.updateProgress({ current: 1, total: 1, label: 'Done' });
+        }
+
+        console.log(`[gmail-sync] Saved ${totalSaved}, skipped ${totalSkipped}, errors ${totalErrors}`);
+      },
+    },
+  ],
+};
+
+registerHandler(gmailSyncHandler);

src/servers/queue/handlers/index.ts

@@ -0,0 +1 @@
+import './gmail-sync';

src/servers/queue/index.ts

@@ -0,0 +1,14 @@
+import { ensureQueueDir } from './storage';
+import { resumeInterruptedJobs } from './engine';
+import './handlers';
+
+export { enqueue, cancelJob } from './engine';
+export { readJob, listAllJobs } from './storage';
+export { registerHandler } from './handler-registry';
+export type { Job, JobStep, JobStatus, JobProgress, StepContext, JobHandler, JobHandlerStep, EnqueueParams } from './types';
+
+export async function initQueue() {
+  await ensureQueueDir();
+  await resumeInterruptedJobs();
+  console.log('[queue] Initialized');
+}

src/servers/queue/storage.ts

@@ -0,0 +1,51 @@
+import type { Job } from './types';
+import { join } from 'node:path';
+import { readdir, mkdir, unlink } from 'node:fs/promises';
+import { DATA_PATH } from '../data-path';
+
+const QUEUE_DIR = join(DATA_PATH, 'queue', 'jobs');
+
+export async function ensureQueueDir() {
+  await mkdir(QUEUE_DIR, { recursive: true });
+}
+
+export async function readJob(id: string): Promise<Job | null> {
+  const file = Bun.file(join(QUEUE_DIR, `${id}.json`));
+  if (!(await file.exists())) return null;
+  try {
+    return await file.json();
+  } catch {
+    return null;
+  }
+}
+
+export async function writeJob(job: Job): Promise<void> {
+  await Bun.write(join(QUEUE_DIR, `${job.id}.json`), JSON.stringify(job, null, 2));
+}
+
+export async function listAllJobs(): Promise<Job[]> {
+  try {
+    const entries = await readdir(QUEUE_DIR);
+    const jobs: Job[] = [];
+    for (const entry of entries) {
+      if (!entry.endsWith('.json')) continue;
+      const file = Bun.file(join(QUEUE_DIR, entry));
+      try {
+        jobs.push(await file.json());
+      } catch {
+        // corrupted file — skip
+      }
+    }
+    return jobs.sort((a, b) => b.createdAt - a.createdAt);
+  } catch {
+    return [];
+  }
+}
+
+export async function deleteJobFile(id: string): Promise<void> {
+  try {
+    await unlink(join(QUEUE_DIR, `${id}.json`));
+  } catch {
+    // file doesn't exist — fine
+  }
+}

src/servers/queue/types.ts

@@ -0,0 +1,57 @@
+export type JobStatus = 'queued' | 'running' | 'completed' | 'failed' | 'cancelled';
+
+export type JobStepStatus = 'pending' | 'running' | 'completed' | 'failed';
+
+export type JobProgress = {
+  current: number;
+  total: number;
+  label?: string;
+};
+
+export type JobStep = {
+  name: string;
+  status: JobStepStatus;
+  startedAt?: number;
+  completedAt?: number;
+  error?: string;
+  progress?: JobProgress;
+};
+
+export type Job = {
+  id: string;
+  lane: string;
+  type: string;
+  userId: string;
+  status: JobStatus;
+  steps: JobStep[];
+  currentStep: number;
+  error?: string;
+  createdAt: number;
+  startedAt?: number;
+  completedAt?: number;
+  meta?: Record<string, unknown>;
+};
+
+export type StepContext = {
+  job: Job;
+  step: JobStep;
+  updateProgress: (progress: JobProgress) => Promise<void>;
+  meta: Record<string, unknown>;
+};
+
+export type JobHandlerStep = {
+  name: string;
+  run: (ctx: StepContext) => Promise<void>;
+};
+
+export type JobHandler = {
+  type: string;
+  steps: JobHandlerStep[];
+};
+
+export type EnqueueParams = {
+  lane: string;
+  type: string;
+  userId: string;
+  meta?: Record<string, unknown>;
+};