From afbb3717c47430bcbc03c41dbbd9fcc5894c2b4b Mon Sep 17 00:00:00 2001 From: Marco Beretta <81851188+berry-13@users.noreply.github.com> Date: Sat, 25 Jul 2026 02:45:39 +0200 Subject: [PATCH] feat(import): add cache-backed import job store --- api/cache/getLogStores.js | 1 + packages/api/src/import/job.spec.ts | 72 +++++++++++++++++++++++++ packages/api/src/import/job.ts | 78 ++++++++++++++++++++++++++++ packages/data-provider/src/config.ts | 4 ++ 4 files changed, 155 insertions(+) create mode 100644 packages/api/src/import/job.spec.ts create mode 100644 packages/api/src/import/job.ts diff --git a/api/cache/getLogStores.js b/api/cache/getLogStores.js index b7f943cc55..84af1de5b2 100644 --- a/api/cache/getLogStores.js +++ b/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), }; /** diff --git a/packages/api/src/import/job.spec.ts b/packages/api/src/import/job.spec.ts new file mode 100644 index 0000000000..0522141430 --- /dev/null +++ b/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(); + }); +}); diff --git a/packages/api/src/import/job.ts b/packages/api/src/import/job.ts new file mode 100644 index 0000000000..506446de7e --- /dev/null +++ b/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 { + 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 { + const job = await this.store.get(this.key(userId, jobId)); + return job ?? null; + } + + async patch(userId: string, jobId: string, patch: Partial): Promise { + 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 { + const updated = await this.patch(userId, jobId, { + status: 'cancelled', + phase: 'cancelled', + }); + return updated !== null; + } + + async isCancelled(userId: string, jobId: string): Promise { + const job = await this.get(userId, jobId); + return job?.status === 'cancelled'; + } +} diff --git a/packages/data-provider/src/config.ts b/packages/data-provider/src/config.ts index 83fed8dfd0..a05143a220 100644 --- a/packages/data-provider/src/config.ts +++ b/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';