feat(import): add cache-backed import job store

api/cache/getLogStores.js

@@ -68,6 +68,7 @@ const namespaces = {
     CacheKeys.ADMIN_OAUTH_EXCHANGE,
     Time.THIRTY_SECONDS,
   ),
+  [CacheKeys.IMPORT_JOBS]: standardCache(CacheKeys.IMPORT_JOBS, Time.ONE_DAY),
 };
 
 /**

packages/api/src/import/job.spec.ts

@@ -0,0 +1,72 @@
+import Keyv from 'keyv';
+
+import { ImportJobStore } from './job';
+
+describe('ImportJobStore', () => {
+  let store: ImportJobStore;
+
+  beforeEach(() => {
+    store = new ImportJobStore(new Keyv(), 60000);
+  });
+
+  it('creates a job awaiting confirmation', async () => {
+    const job = await store.create({
+      userId: 'u1',
+      filepath: '/tmp/a.zip',
+      filename: 'a.zip',
+    });
+
+    expect(job.phase).toBe('queued');
+    expect(job.status).toBe('active');
+    expect(job.jobId).toHaveLength(36);
+    expect(job.progress).toEqual({
+      conversations: { done: 0, total: 0 },
+      messages: { done: 0, total: 0 },
+      assets: { done: 0, total: 0 },
+    });
+  });
+
+  it('reads back a job for its owner only', async () => {
+    const job = await store.create({ userId: 'u1', filepath: '/tmp/a.zip', filename: 'a.zip' });
+
+    expect(await store.get('u1', job.jobId)).not.toBeNull();
+    expect(await store.get('u2', job.jobId)).toBeNull();
+  });
+
+  it('merges patches and bumps updatedAt', async () => {
+    const job = await store.create({ userId: 'u1', filepath: '/tmp/a.zip', filename: 'a.zip' });
+
+    const patched = await store.patch('u1', job.jobId, {
+      phase: 'conversations',
+      progress: {
+        conversations: { done: 5, total: 10 },
+        messages: { done: 40, total: 80 },
+        assets: { done: 0, total: 3 },
+      },
+    });
+
+    expect(patched?.phase).toBe('conversations');
+    expect(patched?.progress.conversations.done).toBe(5);
+    expect(patched?.filename).toBe('a.zip');
+    expect(patched?.updatedAt).toBeGreaterThanOrEqual(job.updatedAt);
+  });
+
+  it('marks a job cancelled and reports it', async () => {
+    const job = await store.create({ userId: 'u1', filepath: '/tmp/a.zip', filename: 'a.zip' });
+
+    expect(await store.isCancelled('u1', job.jobId)).toBe(false);
+    expect(await store.cancel('u1', job.jobId)).toBe(true);
+    expect(await store.isCancelled('u1', job.jobId)).toBe(true);
+    expect((await store.get('u1', job.jobId))?.status).toBe('cancelled');
+  });
+
+  it('does not cancel another user’s job', async () => {
+    const job = await store.create({ userId: 'u1', filepath: '/tmp/a.zip', filename: 'a.zip' });
+    expect(await store.cancel('u2', job.jobId)).toBe(false);
+    expect((await store.get('u1', job.jobId))?.status).toBe('active');
+  });
+
+  it('returns null when patching a job that does not exist', async () => {
+    expect(await store.patch('u1', 'missing', { phase: 'failed' })).toBeNull();
+  });
+});

packages/api/src/import/job.ts

@@ -0,0 +1,78 @@
+import Keyv from 'keyv';
+import { v4 as uuidv4 } from 'uuid';
+
+import type { ImportJob } from './types';
+
+const DEFAULT_TTL = 24 * 60 * 60 * 1000;
+
+function emptyProgress(): ImportJob['progress'] {
+  return {
+    conversations: { done: 0, total: 0 },
+    messages: { done: 0, total: 0 },
+    assets: { done: 0, total: 0 },
+  };
+}
+
+export class ImportJobStore {
+  private readonly store: Keyv;
+  private readonly ttl: number;
+
+  constructor(store: Keyv, ttl: number = DEFAULT_TTL) {
+    this.store = store;
+    this.ttl = ttl;
+  }
+
+  private key(userId: string, jobId: string): string {
+    return `${userId}:${jobId}`;
+  }
+
+  async create(input: { userId: string; filepath: string; filename: string }): Promise<ImportJob> {
+    const now = Date.now();
+    const job: ImportJob = {
+      jobId: uuidv4(),
+      userId: input.userId,
+      filepath: input.filepath,
+      filename: input.filename,
+      phase: 'queued',
+      status: 'active',
+      summary: null,
+      progress: emptyProgress(),
+      report: null,
+      error: null,
+      createdAt: now,
+      updatedAt: now,
+    };
+
+    await this.store.set(this.key(input.userId, job.jobId), job, this.ttl);
+    return job;
+  }
+
+  async get(userId: string, jobId: string): Promise<ImportJob | null> {
+    const job = await this.store.get<ImportJob>(this.key(userId, jobId));
+    return job ?? null;
+  }
+
+  async patch(userId: string, jobId: string, patch: Partial<ImportJob>): Promise<ImportJob | null> {
+    const existing = await this.get(userId, jobId);
+    if (!existing) {
+      return null;
+    }
+
+    const updated: ImportJob = { ...existing, ...patch, updatedAt: Date.now() };
+    await this.store.set(this.key(userId, jobId), updated, this.ttl);
+    return updated;
+  }
+
+  async cancel(userId: string, jobId: string): Promise<boolean> {
+    const updated = await this.patch(userId, jobId, {
+      status: 'cancelled',
+      phase: 'cancelled',
+    });
+    return updated !== null;
+  }
+
+  async isCancelled(userId: string, jobId: string): Promise<boolean> {
+    const job = await this.get(userId, jobId);
+    return job?.status === 'cancelled';
+  }
+}

packages/data-provider/src/config.ts

@@ -2472,6 +2472,10 @@ export enum CacheKeys {
    * Key for admin panel OAuth exchange codes (one-time-use, short TTL).
    */
   ADMIN_OAUTH_EXCHANGE = 'ADMIN_OAUTH_EXCHANGE',
+  /**
+   * Key for cached ChatGPT import job state.
+   */
+  IMPORT_JOBS = 'IMPORT_JOBS',
 }
 
 export const AUTH_USER_DOC_BY_ID_PREFIX = 'auth-user-doc-byid';