Extracted Entities Restart

export interface ExtractionRetryEvent extends BaseMessageEvent {  
    requestId: string;              // Unique ID for this retry request  
    fileIds: string[];              // Array of document hashes (DocumentHash from Documents table)  
    reason: string;                 // Reason for retry (e.g., "extraction_failed", "manual_retry", "manual_restart")  
    retryAttempt: number;           // Current retry attempt number (1-based)  
    maxRetries: number;             // Maximum retry attempts allowed  
    clientApplicationId: string;    // Client application identifier  
    initiatedBy: string;            // User or system that initiated retry (email or 'system')  
    timestamp: Date;                // When retry was requested  
    metadata?: Record<string, any>; // Optional metadata (e.g., { extractedEntityIds: [1,2,3] })  
}  
  
export interface BaseMessageEvent {  
    timestamp: Date;  
    clientApplicationId: string;  
}  

export interface ExtractionRetryResult extends BaseMessageEvent {  
    requestId: string;              // Original request ID  
    fileIds: string[];              // Files that were processed  
    successfulIds: string[];        // Successfully re-extracted file IDs  
    failedIds: string[];            // Failed file IDs  
    errors: Record<string, string>; // Map of fileId to error message  
    timestamp: Date;  
    clientApplicationId: string;  
}  

import { createPublisher } from '../../internal/pubsub/createPublisher.js';  
import { app } from '../../internal/app.js';  
import { cached } from '../../internal/cached.js';  
import { ENV_KEYS } from '../../constants/configuration.js';  
  
export const createServiceBusPublisher = async () =>  
    cached(async () =>  
        createPublisher(await app.lookup(ENV_KEYS.AZURE_SERVICE_BUS_CONNECTION_STRING))  
    );  

import { cached } from '../../internal/cached.js';  
import { createServiceBusPublisher } from './createServiceBusPublisher.js';  
import { app } from '../../internal/app.js';  
import { ENV_KEYS } from '../../constants/configuration.js';  
  
export const createSendToExtractionRetryTopic = async () => {  
    const topicName = await app.lookup(ENV_KEYS.TOPIC_EXTRACTION_RETRY);  
    const publisher = await cached(createServiceBusPublisher);  
    return publisher(topicName);  
};  

import { app } from '../../internal/app.js';  
import { cached } from '../../internal/cached.js';  
import { createSendToExtractionRetryTopic } from './createSendToExtractionRetryTopic.js';  
import { ExtractionRetryEvent } from '../../types/messaging.js';  
import { formatRetryMessage } from './formatRetryMessage.js';  
  
export const sendExtractionRetryEvent = async (event: ExtractionRetryEvent): Promise<void> => {  
    const sender = await cached(createSendToExtractionRetryTopic);  
    await sender(formatRetryMessage(event));  
     
    app.log.info(`Extraction retry event sent for ${event.fileIds.length} file(s)`, {  
        requestId: event.requestId,  
        fileIds: event.fileIds,  
        reason: event.reason,  
        retryAttempt: event.retryAttempt,  
    });  
};  

import { ExtractionRetryEvent } from '../../types/messaging.js';  
  
export const formatRetryMessage = (event: ExtractionRetryEvent) => ({  
    ...event,  
    timestamp: event.timestamp.toISOString(),  
});  

import { ServiceBusReceivedMessage } from '@azure/service-bus';  
import { app } from '../internal/app.js';  
import { SubscriptionHandler, SubscriptionMessageHandlers } from '../types/messaging.js';  
import { ExtractionService } from '../services/extractionService.js';  
import { FileService } from '../services/fileService.js';  
import { sendExtractionRetryResult } from '../services/messaging/sendExtractionRetryResult.js';  
  
export class ExtractionRetryHandler extends SubscriptionHandler {  
    constructor(  
        private readonly extractionService: ExtractionService,  
        private readonly fileService: FileService  
    ) {  
        super();  
    }  
  
    messageHandler: SubscriptionMessageHandlers = {  
        processMessage: async (message: ServiceBusReceivedMessage): Promise<void> => {  
            await this.handleExtractionRetry(message);  
        },  
    };  
  
    private async handleExtractionRetry(message: ServiceBusReceivedMessage): Promise<void> {  
        const event = message.body;  
         
        // Validate message  
        if (!event.fileIds || !Array.isArray(event.fileIds) || event.fileIds.length === 0) {  
            throw new Error('ExtractionRetryEvent must contain at least one fileId');  
        }  
  
        if (!event.requestId) {  
            throw new Error('ExtractionRetryEvent must contain a requestId');  
        }  
  
        app.log.info(`Processing extraction retry request`, {  
            requestId: event.requestId,  
            fileCount: event.fileIds.length,  
            reason: event.reason,  
            retryAttempt: event.retryAttempt,  
        });  
  
        const successfulIds: string[] = [];  
        const failedIds: string[] = [];  
        const errors: Record<string, string> = {};  
  
        // Process each file  
        for (const fileId of event.fileIds) {  
            try {  
                // Validate file exists  
                const file = await this.fileService.getFileById(fileId);  
                if (!file) {  
                    throw new Error(`File with ID ${fileId} not found`);  
                }  
  
                // Check retry limit  
                if (event.retryAttempt > event.maxRetries) {  
                    throw new Error(`Maximum retry attempts (${event.maxRetries}) exceeded`);  
                }  
  
                // Trigger re-extraction  
                app.log.info(`Re-extracting file ${fileId}`, { requestId: event.requestId });  
                await this.extractionService.extractDocument(fileId);  
                 
                successfulIds.push(fileId);  
                app.log.info(`Successfully re-extracted file ${fileId}`);  
                 
            } catch (error) {  
                const errorMessage = error instanceof Error ? error.message : 'Unknown error';  
                app.log.error(`Failed to re-extract file ${fileId}`, error);  
                failedIds.push(fileId);  
                errors[fileId] = errorMessage;  
            }  
        }  
  
        // Send result message  
        await sendExtractionRetryResult({  
            requestId: event.requestId,  
            fileIds: event.fileIds,  
            successfulIds,  
            failedIds,  
            errors,  
            timestamp: new Date(),  
            clientApplicationId: event.clientApplicationId,  
        });  
  
        app.log.info(`Extraction retry completed`, {  
            requestId: event.requestId,  
            total: event.fileIds.length,  
            successful: successfulIds.length,  
            failed: failedIds.length,  
        });  
    }  
}  

import { app } from '../internal/app.js';  
import { TopicSubscription } from '../types/messaging.js';  
import { ENV_KEYS } from '../constants/configuration.js';  
import { ExtractionRetryHandler } from '../subscriptions/extractionRetryHandler.js';  
  
export class TopicService {  
    constructor(  
        private readonly extractionRetryHandler: ExtractionRetryHandler  
    ) {}  
  
    async getMessageSubscribers(): Promise<TopicSubscription[]> {  
        const subscribers: TopicSubscription[] = [  
            {  
                topic: await app.lookup(ENV_KEYS.TOPIC_EXTRACTION_RETRY),  
                subscription: await app.lookup(ENV_KEYS.SUBSCRIPTION_EXTRACTION_RETRY),  
                handler: this.extractionRetryHandler,  
            },  
        ];  
        return subscribers;  
    }  
}  

import { sendExtractionRetryEvent } from './messaging/sendExtractionRetryEvent.js';  
import { ExtractionRetryEvent } from '../types/messaging.js';  
import { v4 as uuidv4 } from 'uuid';  
  
export class StageMapper {  
    /**  
     * Triggers re-extraction for a single file  
     */  
    async retryExtraction(  
        fileId: string,  
        clientApplicationId: string,  
        reason: string,  
        initiatedBy: string,  
        retryAttempt: number = 1,  
        maxRetries: number = 3  
    ): Promise<string> {  
        const requestId = uuidv4();  
         
        const event: ExtractionRetryEvent = {  
            requestId,  
            fileIds: [fileId],  
            reason,  
            retryAttempt,  
            maxRetries,  
            clientApplicationId,  
            initiatedBy,  
            timestamp: new Date(),  
        };  
  
        await sendExtractionRetryEvent(event);  
        return requestId;  
    }  
  
    /**  
     * Triggers re-extraction for multiple files (batch)  
     */  
    async retryExtractionBatch(  
        fileIds: string[],  
        clientApplicationId: string,  
        reason: string,  
        initiatedBy: string,  
        retryAttempt: number = 1,  
        maxRetries: number = 3  
    ): Promise<string> {  
        if (fileIds.length === 0) {  
            throw new Error('Cannot retry extraction for empty file list');  
        }  
  
        const requestId = uuidv4();  
         
        const event: ExtractionRetryEvent = {  
            requestId,  
            fileIds,  
            reason,  
            retryAttempt,  
            maxRetries,  
            clientApplicationId,  
            initiatedBy,  
            timestamp: new Date(),  
        };  
  
        await sendExtractionRetryEvent(event);  
        return requestId;  
    }  
}  

restartExtraction: {  
  security: [{ bearerAuth: [] }],  
  body: {  
    type: 'object',  
    properties: {  
      extractedEntityIds: {  
        type: 'array',  
        items: {  
          type: 'integer',  
          description: 'Extracted entity ID',  
        },  
        minItems: 1,  
        maxItems: 100,  
        description: 'Array of extracted entity IDs to restart extraction for (1-100 items)',  
      },  
      reason: {  
        type: 'string',  
        minLength: 1,  
        maxLength: 500,  
        description: 'Reason for restart (e.g., "manual_retry", "extraction_failed")',  
      },  
    },  
    required: ['extractedEntityIds'],  
  },  
  tags: ['Extracted Entities'],  
  response: {  
    200: {  
      type: 'object',  
      properties: {  
        requestId: {  
          type: 'string',  
          format: 'uuid',  
          description: 'Unique request ID for tracking this restart operation',  
        },  
        extractedEntityIds: {  
          type: 'array',  
          items: { type: 'integer' },  
          description: 'List of extracted entity IDs being restarted',  
        },  
        fileHashes: {  
          type: 'array',  
          items: { type: 'string' },  
          description: 'List of document hashes sent to Fluxx for extraction',  
        },  
        status: {  
          type: 'string',  
          enum: ['accepted', 'partial', 'failed'],  
          description: 'Overall status of the restart request',  
        },  
        message: {  
          type: 'string',  
          description: 'Human-readable status message',  
        },  
        errors: {  
          type: 'array',  
          items: {  
            type: 'object',  
            properties: {  
              extractedEntityId: { type: 'integer' },  
              error: { type: 'string' },  
            },  
          },  
          description: 'List of errors for entities that could not be restarted',  
        },  
      },  
    },  
    400: {  
      type: 'object',  
      properties: {  
        error: { type: 'string' },  
        message: { type: 'string' },  
      },  
    },  
    404: {  
      type: 'object',  
      properties: {  
        error: { type: 'string' },  
        message: { type: 'string' },  
      },  
    },  
    500: {  
      type: 'object',  
      properties: {  
        error: { type: 'string' },  
        message: { type: 'string' },  
      },  
    },  
  },  
},  

export interface RestartExtractionRequest {  
  extractedEntityIds: number[];  
  reason?: string;  
}  
  
export interface RestartExtractionResponse {  
  requestId: string;  
  extractedEntityIds: number[];  
  fileHashes: string[];  
  status: 'accepted' | 'partial' | 'failed';  
  message: string;  
  errors?: Array<{  
    extractedEntityId: number;  
    error: string;  
  }>;  
}  
  
export interface ExtractedEntityWithDocument extends Entity {  
  documentHash: string;  
  documentId: number;  
}  
  
// Entity interface uses existing status values  
export interface Entity {  
  entityID: number;  
  documentID: number;  
  extractedEntitiesPayload: string;  
  stage: string;  
  status: 'pending' | 'processing' | 'completed' | 'failed';  
  substatus: string | null;  
  createdOn: string;  
  modifiedOn?: string;  
  dateExported?: string | null;  
  dateFiled?: string | null;  
}  

static async getEntitiesWithDocumentsByIds(  
  entityIds: number[]  
): Promise<ExtractedEntityWithDocument[]> {  
  if (entityIds.length === 0) {  
    return [];  
  }  
  
  const db = getDb();  
   
  // Build parameterized query for IN clause  
  const params: Record<string, any> = {};  
  const paramNames = entityIds.map((id, index) => {  
    const paramName = `id${index}`;  
    params[paramName] = id;  
    return `@${paramName}`;  
  }).join(',');  
  
  const sql = `  
    SELECT  
      ee.ExtractedEntityId AS entityID,  
      ee.DocumentId AS documentID,  
      d.DocumentHash AS documentHash,  
      ee.Status AS status  
    FROM ExtractedEntities ee  
    INNER JOIN Documents d ON ee.DocumentId = d.DocumentId  
    WHERE ee.ExtractedEntityId IN (${paramNames});  
  `;  
  
  const result = await db.executeQuery(sql, params);  
  return result.recordset || [];  
}  

import { v4 as uuidv4 } from 'uuid';  
import { sendExtractionRetryEvent } from '../../../services/messaging/sendExtractionRetryEvent.js';  
import { ExtractionRetryEvent } from '../../../types/messaging.js';  
import {  
  RestartExtractionRequest,  
  RestartExtractionResponse,  
  ExtractedEntityWithDocument  
} from './extractedEntities.types.js';  
  
const EntitiesService = {  
  // ... existing methods  
  
  async restartExtraction(  
    request: RestartExtractionRequest,  
    initiatedBy: string,  
    clientApplicationId: string  
  ): Promise<RestartExtractionResponse> {  
    const { extractedEntityIds, reason = 'manual_restart' } = request;  
    const requestId = uuidv4();  
    const errors: Array<{ extractedEntityId: number; error: string }> = [];  
     
    // Validate extracted entity IDs  
    if (!extractedEntityIds || extractedEntityIds.length === 0) {  
      throw new Error('extractedEntityIds is required and must contain at least one ID');  
    }  
  
    if (extractedEntityIds.length > 100) {  
      throw new Error('Cannot restart more than 100 entities at once');  
    }  
  
    // Get entities with document information  
    const entities = await EntitiesModel.getEntitiesWithDocumentsByIds(extractedEntityIds);  
     
    // Check if all entities were found  
    const foundIds = entities.map(e => e.entityID);  
    const notFoundIds = extractedEntityIds.filter(id => !foundIds.includes(id));  
     
    if (notFoundIds.length > 0) {  
      notFoundIds.forEach(id => {  
        errors.push({  
          extractedEntityId: id,  
          error: `Extracted entity with ID ${id} not found`,  
        });  
      });  
    }  
  
    // Validate entities can be restarted (not already processing)  
    const validEntities: ExtractedEntityWithDocument[] = [];  
    entities.forEach(entity => {  
      if (entity.status === 'processing') {  
        errors.push({  
          extractedEntityId: entity.entityID,  
          error: 'Cannot restart entity that is currently processing',  
        });  
      } else {  
        validEntities.push(entity);  
      }  
    });  
  
    // If no valid entities, return failed status  
    if (validEntities.length === 0) {  
      return {  
        requestId,  
        extractedEntityIds: extractedEntityIds,  
        fileHashes: [],  
        status: 'failed',  
        message: 'No valid entities to restart',  
        errors,  
      };  
    }  
  
    // Extract file hashes and entity IDs  
    const validEntityIds = validEntities.map(e => e.entityID);  
    const fileHashes = validEntities.map(e => e.documentHash);  
  
    // Send message to Fluxx Pub/Sub for each file hash  
    const event: ExtractionRetryEvent = {  
      requestId,  
      fileIds: fileHashes, // Using documentHash as fileId for Fluxx  
      reason,  
      retryAttempt: 1,  
      maxRetries: 3,  
      clientApplicationId,  
      initiatedBy,  
      timestamp: new Date(),  
      metadata: {  
        extractedEntityIds: validEntityIds,  
      },  
    };  
  
    try {  
      await sendExtractionRetryEvent(event);  
    } catch (error) {  
      throw new Error(`Failed to send extraction retry message: ${error instanceof Error ? error.message : 'Unknown error'}`);  
    }  
  
    const status = errors.length === 0 ? 'accepted' : 'partial';  
    const message = errors.length === 0  
      ? `Successfully initiated restart for ${validEntities.length} extraction(s)`  
      : `Initiated restart for ${validEntities.length} extraction(s) with ${errors.length} error(s)`;  
  
    return {  
      requestId,  
      extractedEntityIds: validEntityIds,  
      fileHashes,  
      status,  
      message,  
      errors: errors.length > 0 ? errors : undefined,  
    };  
  },  
};  
  
export default EntitiesService;  

import { FastifyInstance } from 'fastify';  
import { RestartExtractionRequest } from './extractedEntities.types.js';  
  
const EntitiesController = {  
  // ... existing methods  
  
  async restartExtraction(  
    fastify: FastifyInstance,  
    request: RestartExtractionRequest,  
    initiatedBy: string,  
    clientApplicationId: string  
  ) {  
    return EntitiesService.restartExtraction(request, initiatedBy, clientApplicationId);  
  },  
};  
  
export default EntitiesController;  

fastify.post(`${basePath}/extracted_entities/restart`, {  
  schema: entitySchema.restartExtraction,  
  preValidation: [fastify.authenticate, fastify.authorize],  
  config: {  
    validRoles: [ROLES.Administrator, ROLES.User],  
    validScopes: [SCOPES.EXTRACTED_ENTITIES_WRITE],  
  },  
  handler: async (request: FastifyRequest, reply: FastifyReply) => {  
    try {  
      const body = request.body as RestartExtractionRequest;  
       
      // Extract user info from authenticated request  
      const initiatedBy = (request.user as any)?.email || (request.user as any)?.sub || 'unknown';  
      const clientApplicationId = (request.user as any)?.clientId || 'default';  
  
      const response = await EntitiesController.restartExtraction(  
        fastify,  
        body,  
        initiatedBy,  
        clientApplicationId  
      );  
  
      const statusCode = response.status === 'failed' ? 400 : 200;  
      reply.status(statusCode).send(response);  
    } catch (err) {  
      request.log.error(err);  
       
      if (err instanceof Error) {  
        if (err.message.includes('not found')) {  
          return reply.status(404).send({  
            error: 'Not Found',  
            message: err.message  
          });  
        }  
        if (err.message.includes('required') || err.message.includes('Cannot restart')) {  
          return reply.status(400).send({  
            error: 'Bad Request',  
            message: err.message  
          });  
        }  
      }  
       
      reply.status(500).send({  
        error: 'Internal Server Error',  
        message: 'Failed to restart extraction'  
      });  
    }  
  },  
});  

import { describe, it, expect, vi, beforeEach } from 'vitest';  
import EntitiesModel from '../../../../../src/api/v1/extractedEntity/extractedEntities.model.js';  
import { getDb } from '../../../../../src/plugins/db.js';  
  
vi.mock('../../../../../src/plugins/db.js');  
  
describe('EntitiesModel - Restart Extraction', () => {  
  beforeEach(() => {  
    vi.clearAllMocks();  
  });  
  
  it('should get entities with documents by IDs', async () => {  
    const mockDb = {  
      executeQuery: vi.fn().mockResolvedValue({  
        recordset: [  
          {  
            entityID: 1,  
            documentID: 100,  
            documentHash: 'hash123',  
            status: 'failed',  
          },  
        ],  
      }),  
    };  
    vi.mocked(getDb).mockReturnValue(mockDb as any);  
  
    const result = await EntitiesModel.getEntitiesWithDocumentsByIds([1]);  
  
    expect(result).toHaveLength(1);  
    expect(result[0].documentHash).toBe('hash123');  
    expect(mockDb.executeQuery).toHaveBeenCalledWith(  
      expect.stringContaining('INNER JOIN Documents'),  
      expect.objectContaining({ id0: 1 })  
    );  
  });  
});