feat: implement asynchronous session processing with PENDING status and background execution

Author: Thavarshan

src/controllers/session-controller.ts

@@ -47,6 +47,12 @@ const CreateSessionSchema = z.object({
       experienceLevel: z.enum(['beginner', 'intermediate', 'advanced']).optional(),
     })
     .optional(),
+  /**
+   * If true, returns immediately with PENDING status and processes in background.
+   * Client must poll GET /api/sessions/:id to check completion.
+   * Default: false (synchronous processing for small files)
+   */
+  async: z.boolean().optional().default(false),
 });
 
 /**
@@ -133,7 +139,28 @@ export class SessionController {
         return res.status(401).json({ message: 'Unauthorized' });
       }
 
-      // Run workflow and persist result
+      // Async mode: Create PENDING session and process in background
+      if (payload.async) {
+        const pendingSession = await this.sessionRepository.createPending({
+          sessionId: payload.sessionId,
+          userId,
+          uploadUrl: payload.uploadUrl,
+          userContext: payload.userContext,
+        });
+
+        // Process in background (don't await)
+        this.sessionService.runSession(payload, userId).catch((error) => {
+          console.error(`Background session ${payload.sessionId} failed:`, error);
+        });
+
+        return res.status(202).json({
+          sessionId: pendingSession.sessionId,
+          status: 'PENDING',
+          message: 'Session queued for processing. Poll GET /api/sessions/:id to check status.',
+        });
+      }
+
+      // Sync mode: Run workflow and wait for completion (original behavior)
       const result = await this.sessionService.runSession(payload, userId);
       return res.status(201).json(result);
     } catch (error) {
@@ -258,9 +285,12 @@ export class SessionController {
         return res.status(404).json({ message: 'Session not found' });
       }
 
-      // Parse and return complete EngineState
+      // Parse EngineState and include database status
       const state = JSON.parse(session.payload) as EngineState;
-      return res.status(200).json(state);
+      return res.status(200).json({
+        ...state,
+        status: session.status, // PENDING, PROCESSING, COMPLETED, FAILED
+      });
     } catch (error) {
       return next(error);
     }

src/repositories/session-repository.ts

@@ -68,18 +68,22 @@ export class SessionRepository {
    * @throws Prisma errors if database operation fails
    */
   async save(state: EngineState, userId: string) {
+    const status = state.error ? 'FAILED' : 'COMPLETED';
+
     await prisma.session.upsert({
       where: { sessionId: state.sessionId },
       create: {
         sessionId: state.sessionId,
         userId,
         uploadUrl: state.uploadUrl,
         payload: JSON.stringify(state),
+        status,
         error: state.error,
       },
       update: {
         uploadUrl: state.uploadUrl,
         payload: JSON.stringify(state),
+        status,
         error: state.error,
       },
     });
@@ -111,6 +115,44 @@ export class SessionRepository {
     });
   }
 
+  /**
+   * Creates a new PENDING session for async processing.
+   *
+   * Used when client requests async processing (large files).
+   * Creates minimal session record with PENDING status that will be
+   * updated once background processing completes.
+   *
+   * **Workflow:**
+   * 1. Client sends POST /api/sessions with async=true
+   * 2. Controller calls createPending (returns immediately)
+   * 3. Background job processes the session
+   * 4. Background job calls save() to update with results
+   * 5. Client polls GET /api/sessions/:id to check status
+   *
+   * @param data - Session initialization data
+   * @returns Promise resolving to created Session record
+   */
+  async createPending(data: {
+    sessionId: string;
+    userId: string;
+    uploadUrl: string;
+    userContext?: Record<string, unknown>;
+  }): Promise<Session> {
+    return prisma.session.create({
+      data: {
+        sessionId: data.sessionId,
+        userId: data.userId,
+        uploadUrl: data.uploadUrl,
+        payload: JSON.stringify({
+          sessionId: data.sessionId,
+          uploadUrl: data.uploadUrl,
+          userContext: data.userContext,
+        }),
+        status: 'PENDING',
+      },
+    });
+  }
+
   /**
    * Retrieves a specific session by ID, with user ownership verification.
    *
@@ -142,4 +184,20 @@ export class SessionRepository {
       },
     });
   }
+
+  /**
+   * Updates session status to PROCESSING when workflow starts.
+   *
+   * Called at the beginning of async job processing to indicate
+   * that work has begun. This allows clients polling the status
+   * to see progress beyond just PENDING.
+   *
+   * @param sessionId - Session to mark as processing
+   */
+  async markProcessing(sessionId: string): Promise<void> {
+    await prisma.session.update({
+      where: { sessionId },
+      data: { status: 'PROCESSING' },
+    });
+  }
 }

src/services/session-service.ts

@@ -104,10 +104,17 @@ export class SessionService {
    * ```
    */
   async runSession(input: RunSessionInput, userId: string): Promise<EngineState> {
+    // Mark as PROCESSING if session exists (async workflow)
+    try {
+      await this.repository.markProcessing(input.sessionId);
+    } catch {
+      // Ignore error if session doesn't exist yet (sync workflow will create it)
+    }
+
     // Execute complete workflow (metadata → analysis → feedback)
     const state = await runAuralyzeSession(input, this.deps);
 
-    // Persist result (success or error) to database
+    // Persist result (success or error) to database with final status
     await this.repository.save(state, userId);
 
     return state;