> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://cla-demo.postmansolutions.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://cla-demo.postmansolutions.com/_mcp/server.

# Extracted Entities Restart

## 🆕 What's New - Restart Endpoint

This story has been **updated** to include a new REST API endpoint for manually restarting extractions:

**Endpoint**: `POST /v1/extracted_entities/restart`

**Key Features**:

* Accept array of `extractedEntityIds` (1-100 items)

* Query `ExtractedEntities` JOIN `Documents` to retrieve document hashes

* Send messages to Fluxx Pub/Sub with document hashes for re-extraction

* Return detailed response with request tracking ID and errors

* Support partial success (some entities succeed, others fail)

* No database updates required (read-only operation)

**Use Cases**:

* Operators manually restart failed extractions via API

* Batch restart multiple extractions at once (up to 100)

* Support for monitoring/alerting systems to auto-restart

* Clear audit trail of who initiated restarts and why

---

## Business Value

* **Automated Retry**: Enable automatic re-extraction of failed document processing without manual intervention

* **Batch Processing**: Support batch retry operations for multiple files simultaneously

* **Manual Restart**: Allow operators to manually restart extraction for specific extracted entities via REST API

* **Decoupled Architecture**: Stage mapper service remains decoupled from extraction service through message-based communication

* **Reliability**: Failed extractions can be retried without data loss or manual file resubmission

* **Audit Trail**: All retry requests are logged and traceable through Azure Service Bus

* **Scalability**: Message-based architecture scales horizontally with processing demands

* **Status Tracking**: Clear visibility into extraction restart status through entity status updates

---

## Current State

### Existing Architecture

The web-api-documents-ai service currently:

* Processes documents through a stage mapper that coordinates extraction workflow

* Has no automated mechanism to trigger re-extraction of files

* Requires manual intervention or API calls to retry failed extractions

* Lacks integration with Azure Service Bus messaging infrastructure

### Gap Analysis

1. No pub/sub messaging infrastructure in place

2. No message publishing capability from stage mapper

3. No subscription handler to receive and process retry requests

4. No environment configuration for Service Bus topics and subscriptions

5. No structured message format for extraction retry events

---

## Future State

### Architecture Pattern

Implement Azure Service Bus pub/sub messaging following the pattern established in `fluxx-file-intake`:

```

┌─────────────────────────────────────────────────────────────────┐  

│                    web-api-documents-ai                          │  

├─────────────────────────────────────────────────────────────────┤  

│                                                                   │  

│  ┌──────────────────────┐         ┌─────────────────────────┐   │  

│  │   Stage Mapper       │         │  Extraction Service     │   │  

│  │   (Publisher)        │         │  (Subscriber)           │   │  

│  └──────────┬───────────┘         └─────────▲───────────────┘   │  

│             │                               │                   │  

│             │ Publish                       │ Subscribe         │  

│             │ ExtractionRetryEvent          │                   │  

│             │                               │                   │  

│  ┌──────────▼───────────────────────────────┴───────────────┐   │  

│  │        Azure Service Bus Topic                           │   │  

│  │        "extraction-retry"                                │   │  

│  │                                                           │   │  

│  │    Subscription: "extractor-retry-subscription"          │   │  

│  └──────────────────────────────────────────────────────────┘   │  

│                                                                   │  

└─────────────────────────────────────────────────────────────────┘  

```

### Message Flow

1. **Trigger**: Stage mapper detects extraction failure or receives retry request

2. **Publish**: Stage mapper publishes `ExtractionRetryEvent` to `extraction-retry` topic

3. **Queue**: Azure Service Bus queues message in subscription

4. **Subscribe**: Extraction service receives message from subscription

5. **Process**: Extraction handler processes retry request and re-extracts file(s)

6. **Complete**: Message is completed upon successful processing or abandoned for retry

---

## Design

### Message Types

#### ExtractionRetryEvent

This event is published to the Fluxx Pub/Sub queue to trigger extraction.

**Important**: The `fileIds` field contains **document hashes** (from the `Documents.DocumentHash` column), not database IDs. Fluxx uses these file hashes to identify and process the documents.

```typescript

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;  

}  

```

**Message Flow for Restart Endpoint**:

1. Receive `extractedEntityIds` from API request

2. Query `ExtractedEntities` JOIN `Documents` to get document hashes (read-only)

3. Send `ExtractionRetryEvent` with document hashes in `fileIds` field

4. Fluxx receives message and initiates extraction for each file hash

5. Fluxx processes the documents and updates status accordingly

#### ExtractionRetryResult (Response Event)

```typescript

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;  

}  

```

---

### Database Requirements

**Note**: The restart endpoint is **read-only** for the database. No status updates or database modifications are performed.

#### Table Relationships

```

Documents (1) ----< (N) ExtractedEntities  

  

Documents:  

  - DocumentId (PK)  

  - DocumentHash (indexed) - Used as fileId for Fluxx messages  

  

ExtractedEntities:  

  - ExtractedEntityId (PK)  

  - DocumentId (FK) - Links to Documents.DocumentId  

```

**Query Pattern** (Read-Only):

The endpoint only reads data to retrieve document hashes. No UPDATE, INSERT, or DELETE operations are performed.

---

### Component Architecture

#### 1. Publisher Infrastructure

**File**: `src/services/messaging/createServiceBusPublisher.ts`

```typescript

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))  

    );  

```

**File**: `src/services/messaging/createSendToExtractionRetryTopic.ts`

```typescript

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);  

};  

```

**File**: `src/services/messaging/sendExtractionRetryEvent.ts`

```typescript

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,  

    });  

};  

```

**File**: `src/services/messaging/formatRetryMessage.ts`

```typescript

import { ExtractionRetryEvent } from '../../types/messaging.js';  

  

export const formatRetryMessage = (event: ExtractionRetryEvent) => ({  

    ...event,  

    timestamp: event.timestamp.toISOString(),  

});  

```

#### 2. Subscriber Infrastructure

**File**: `src/subscriptions/extractionRetryHandler.ts`

```typescript

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,  

        });  

    }  

}  

```

#### 3. Message Receiver Integration

**File**: `src/services/topicService.ts` (Update existing)

```typescript

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;  

    }  

}  

```

#### 4. Stage Mapper Integration

**File**: `src/services/stageMapper.ts` (Update existing)

```typescript

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;  

    }  

}  

```

---

### Phase 3: REST API Endpoint for Extraction Restart

#### Task 3.1: Add REST Endpoint Schema

**File**: `src/api/v1/extractedEntity/extractedEntities.schema.ts`

Add new schema for restart endpoint:

```typescript

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' },  

      },  

    },  

  },  

},  

```

#### Task 3.2: Add Service Types

**File**: `src/api/v1/extractedEntity/extractedEntities.types.ts`

Add new types for restart functionality:

```typescript

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;  

}  

```

#### Task 3.3: Add Model Methods

**File**: `src/api/v1/extractedEntity/extractedEntities.model.ts`

Add method to get entities with document information:

```typescript

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 || [];  

}  

```

#### Task 3.4: Add Service Method

**File**: `src/api/v1/extractedEntity/extractedEntities.service.ts`

Add restart extraction service method:

```typescript

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;  

```

#### Task 3.5: Add Controller Method

**File**: `src/api/v1/extractedEntity/extractedEntities.controller.ts`

Add restart extraction controller method:

```typescript

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;  

```

#### Task 3.6: Add Route

**File**: `src/api/v1/extractedEntity/extractedEntities.route.ts`

Add restart endpoint route:

```typescript

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'  

      });  

    }  

  },  

});  

```

---

### Phase 4: Testing

#### Test 4.1: Unit Test - Model Methods

**File**: `test/unit/api/v1/extractedEntity/extractedEntities.model.test.ts`

```typescript

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 })  

    );  

  });  

});  

```

#### Test 4.2: Unit Test - Service Method

**File**: `test/unit/api/v1/extractedEntity/extractedEntities.service.test.ts`

```typescript

import { describe, it, expect, vi, beforeEach } from 'vitest';  

import EntitiesService from '../../../../../src/api/v1/extractedEntity/extractedEntities.service.js';  

import EntitiesModel from '../../../../../src/api/v1/extractedEntity/extractedEntities.model.js';  

import { sendExtractionRetryEvent } from '../../../../../src/services/messaging/sendExtractionRetryEvent.js';  

  

vi.mock('../../../../../src/api/v1/extractedEntity/extractedEntities.model.js');  

vi.mock('../../../../../src/services/messaging/sendExtractionRetryEvent.js');  

  

describe('EntitiesService - Restart Extraction', () => {  

  beforeEach(() => {  

    vi.clearAllMocks();  

  });  

  

  it('should successfully restart extraction for valid entities', async () => {  

    vi.mocked(EntitiesModel.getEntitiesWithDocumentsByIds).mockResolvedValue([  

      {  

        entityID: 1,  

        documentID: 100,  

        documentHash: 'hash123',  

        status: 'failed',  

        extractedEntitiesPayload: '{}',  

        stage: 'extraction',  

        substatus: null,  

        createdOn: '2026-01-01',  

      },  

    ]);  

  

    vi.mocked(sendExtractionRetryEvent).mockResolvedValue(undefined);  

  

    const result = await EntitiesService.restartExtraction(  

      { extractedEntityIds: [1] },  

      'test@example.com',  

      'test-client'  

    );  

  

    expect(result.status).toBe('accepted');  

    expect(result.fileHashes).toEqual(['hash123']);  

    expect(result.extractedEntityIds).toEqual([1]);  

    expect(sendExtractionRetryEvent).toHaveBeenCalledWith(  

      expect.objectContaining({  

        fileIds: ['hash123'],  

        reason: 'manual_restart',  

      })  

    );  

  });  

  

  it('should handle entities not found', async () => {  

    vi.mocked(EntitiesModel.getEntitiesWithDocumentsByIds).mockResolvedValue([]);  

  

    const result = await EntitiesService.restartExtraction(  

      { extractedEntityIds: [999] },  

      'test@example.com',  

      'test-client'  

    );  

  

    expect(result.status).toBe('failed');  

    expect(result.errors).toHaveLength(1);  

    expect(result.errors![0].extractedEntityId).toBe(999);  

  });  

  

  it('should reject entities currently processing', async () => {  

    vi.mocked(EntitiesModel.getEntitiesWithDocumentsByIds).mockResolvedValue([  

      {  

        entityID: 1,  

        documentID: 100,  

        documentHash: 'hash123',  

        status: 'processing',  

        extractedEntitiesPayload: '{}',  

        stage: 'extraction',  

        substatus: null,  

        createdOn: '2026-01-01',  

      },  

    ]);  

  

    const result = await EntitiesService.restartExtraction(  

      { extractedEntityIds: [1] },  

      'test@example.com',  

      'test-client'  

    );  

  

    expect(result.status).toBe('failed');  

    expect(result.errors).toHaveLength(1);  

    expect(result.errors![0].error).toContain('currently processing');  

  });  

  

  it('should handle partial failures', async () => {  

    vi.mocked(EntitiesModel.getEntitiesWithDocumentsByIds).mockResolvedValue([  

      {  

        entityID: 1,  

        documentID: 100,  

        documentHash: 'hash123',  

        status: 'failed',  

        extractedEntitiesPayload: '{}',  

        stage: 'extraction',  

        substatus: null,  

        createdOn: '2026-01-01',  

      },  

    ]);  

  

    vi.mocked(sendExtractionRetryEvent).mockResolvedValue(undefined);  

  

    const result = await EntitiesService.restartExtraction(  

      { extractedEntityIds: [1, 999] }, // 1 exists, 999 doesn't  

      'test@example.com',  

      'test-client'  

    );  

  

    expect(result.status).toBe('partial');  

    expect(result.extractedEntityIds).toEqual([1]);  

    expect(result.errors).toHaveLength(1);  

    expect(result.errors![0].extractedEntityId).toBe(999);  

  });  

  

  it('should handle message send failure', async () => {  

    vi.mocked(EntitiesModel.getEntitiesWithDocumentsByIds).mockResolvedValue([  

      {  

        entityID: 1,  

        documentID: 100,  

        documentHash: 'hash123',  

        status: 'failed',  

        extractedEntitiesPayload: '{}',  

        stage: 'extraction',  

        substatus: null,  

        createdOn: '2026-01-01',  

      },  

    ]);  

  

    vi.mocked(sendExtractionRetryEvent).mockRejectedValue(new Error('Service Bus unavailable'));  

  

    await expect(  

      EntitiesService.restartExtraction(  

        { extractedEntityIds: [1] },  

        'test@example.com',  

        'test-client'  

      )  

    ).rejects.toThrow('Failed to send extraction retry message');  

  });  

  

  it('should reject more than 100 entities', async () => {  

    const entityIds = Array.from({ length: 101 }, (_, i) => i + 1);  

  

    await expect(  

      EntitiesService.restartExtraction(  

        { extractedEntityIds: entityIds },  

        'test@example.com',  

        'test-client'  

      )  

    ).rejects.toThrow('Cannot restart more than 100 entities at once');  

  });  

});  

```

#### Test 4.3: Integration Test - Restart Endpoint

**File**: `test/integration/extractedEntities.restart.test.ts`

```typescript

import { describe, it, expect, beforeAll, afterAll } from 'vitest';  

import { build } from '../../src/app.js';  

import { FastifyInstance } from 'fastify';  

  

describe('POST /v1/extracted_entities/restart', () => {  

  let app: FastifyInstance;  

  

  beforeAll(async () => {  

    app = await build();  

  });  

  

  afterAll(async () => {  

    await app.close();  

  });  

  

  it('should restart extraction for valid entity IDs', async () => {  

    const response = await app.inject({  

      method: 'POST',  

      url: '/v1/extracted_entities/restart',  

      headers: {  

        authorization: 'Bearer <valid-token>',  

      },  

      payload: {  

        extractedEntityIds: [1, 2],  

        reason: 'manual_retry',  

      },  

    });  

  

    expect(response.statusCode).toBe(200);  

    const body = JSON.parse(response.body);  

    expect(body.status).toBe('accepted');  

    expect(body.requestId).toBeDefined();  

    expect(body.fileHashes).toHaveLength(2);  

  });  

  

  it('should return 400 for empty entity array', async () => {  

    const response = await app.inject({  

      method: 'POST',  

      url: '/v1/extracted_entities/restart',  

      headers: {  

        authorization: 'Bearer <valid-token>',  

      },  

      payload: {  

        extractedEntityIds: [],  

      },  

    });  

  

    expect(response.statusCode).toBe(400);  

  });  

  

  it('should return 404 for non-existent entity', async () => {  

    const response = await app.inject({  

      method: 'POST',  

      url: '/v1/extracted_entities/restart',  

      headers: {  

        authorization: 'Bearer <valid-token>',  

      },  

      payload: {  

        extractedEntityIds: [999999],  

      },  

    });  

  

    expect(response.statusCode).toBe(400);  

    const body = JSON.parse(response.body);  

    expect(body.status).toBe('failed');  

  });  

});  

```

---

## Implementation Steps

### Phase 1: Infrastructure Setup

#### Task 1.1: Add Environment Configuration

**File**: `src/constants/configuration.ts`

```typescript

export const ENV_KEYS = {  

    // ... existing keys  

    AZURE_SERVICE_BUS_CONNECTION_STRING: 'AZURE_SERVICE_BUS_CONNECTION_STRING',  

     

    // Topic Names  

    TOPIC_EXTRACTION_RETRY: 'TOPIC_EXTRACTION_RETRY',  

    TOPIC_EXTRACTION_RESULT: 'TOPIC_EXTRACTION_RESULT',  

     

    // Subscription Names  

    SUBSCRIPTION_EXTRACTION_RETRY: 'SUBSCRIPTION_EXTRACTION_RETRY',  

    SUBSCRIPTION_EXTRACTION_RESULT: 'SUBSCRIPTION_EXTRACTION_RESULT',  

} as const;  

```

#### Task 1.2: Add Environment Variables

**File**: `.env.local.example`

```bash

# Azure Service Bus Configuration  

AZURE_SERVICE_BUS_CONNECTION_STRING='Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;'  

  

# Service Bus Topics  

TOPIC_EXTRACTION_RETRY=extraction-retry  

TOPIC_EXTRACTION_RESULT=extraction-result  

  

# Service Bus Subscriptions  

SUBSCRIPTION_EXTRACTION_RETRY=extractor-retry-subscription  

SUBSCRIPTION_EXTRACTION_RESULT=mapper-result-subscription  

```

**Required Environment Variables**:

* `AZURE_SERVICE_BUS_CONNECTION_STRING` (Required): Connection string for Azure Service Bus namespace

\- Format: `Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=<key-name>;SharedAccessKey=<key-value>`

\- Development: Can use local emulator with `UseDevelopmentEmulator=true`

* `TOPIC_EXTRACTION_RETRY` (Required): Name of the topic for extraction retry events (e.g., "extraction-retry")

* `TOPIC_EXTRACTION_RESULT` (Required): Name of the topic for extraction results (e.g., "extraction-result")

* `SUBSCRIPTION_EXTRACTION_RETRY` (Required): Name of subscription for retry handler (e.g., "extractor-retry-subscription")

* `SUBSCRIPTION_EXTRACTION_RESULT` (Optional): Name of subscription for result handler (e.g., "mapper-result-subscription")

#### Task 1.3: Define Message Types

**File**: `src/types/messaging.ts`

```typescript

import { ProcessErrorArgs, ServiceBusReceivedMessage } from '@azure/service-bus';  

  

export interface BaseMessageEvent {  

    timestamp: Date;  

    clientApplicationId: string;  

}  

  

export interface ExtractionRetryEvent extends BaseMessageEvent {  

    requestId: string;  

    fileIds: string[];  

    reason: string;  

    retryAttempt: number;  

    maxRetries: number;  

    initiatedBy: string;  

    metadata?: Record<string, any>;  

}  

  

export interface ExtractionRetryResult extends BaseMessageEvent {  

    requestId: string;  

    fileIds: string[];  

    successfulIds: string[];  

    failedIds: string[];  

    errors: Record<string, string>;  

}  

  

export interface TopicSubscription {  

    topic: string;  

    subscription: string;  

    handler: SubscriptionHandler;  

}  

  

export abstract class SubscriptionHandler {  

    public abstract messageHandler: SubscriptionMessageHandlers;  

}  

  

export interface SubscriptionMessageHandlers {  

    processMessage(message: ServiceBusReceivedMessage): Promise<void>;  

    processError?(args: ProcessErrorArgs): Promise<void>;  

}  

```

#### Task 1.4: Create Publisher Infrastructure

Create the following files:

* `src/internal/pubsub/createPublisher.ts` (from fluxx-file-intake pattern)

* `src/internal/pubsub/createMessageSender.ts` (from fluxx-file-intake pattern)

* `src/internal/pubsub/createServiceBusClient.ts` (from fluxx-file-intake pattern)

* `src/services/messaging/createServiceBusPublisher.ts`

* `src/services/messaging/createSendToExtractionRetryTopic.ts`

* `src/services/messaging/formatRetryMessage.ts`

* `src/services/messaging/sendExtractionRetryEvent.ts`

#### Task 1.5: Create Subscriber Infrastructure

Create the following files:

* `src/subscriptions/extractionRetryHandler.ts`

* `src/subscriptions/messageReceiver.ts` (from fluxx-file-intake pattern)

* `src/services/topicService.ts`

---

### Phase 2: Integration

#### Task 2.1: Update Stage Mapper

Add retry methods to `StageMapper` class:

```typescript

// src/services/stageMapper.ts  

async retryExtraction(fileId: string, ...): Promise<string>  

async retryExtractionBatch(fileIds: string[], ...): Promise<string>  

```

#### Task 2.2: Initialize Message Receiver

**File**: `src/index.ts` (Update existing)

```typescript

import { MessageReceiver } from './subscriptions/messageReceiver.js';  

import { TopicService } from './services/topicService.js';  

import { ExtractionRetryHandler } from './subscriptions/extractionRetryHandler.js';  

  

// ... existing initialization  

  

// Initialize message receiver  

const extractionService = diContainer.resolve('extractionService');  

const fileService = diContainer.resolve('fileService');  

  

const extractionRetryHandler = new ExtractionRetryHandler(  

    extractionService,  

    fileService  

);  

  

const topicService = new TopicService(extractionRetryHandler);  

  

const serviceBusConnectionString = await app.lookup(  

    ENV_KEYS.AZURE_SERVICE_BUS_CONNECTION_STRING  

);  

  

const messageReceiver = new MessageReceiver(  

    topicService,  

    serviceBusConnectionString  

);  

  

await messageReceiver.startListening();  

  

app.log.info('Message receiver initialized and listening');  

```

#### Task 2.3: Add API Endpoints (Optional)

**File**: `src/routes/v1/retry.ts`

```typescript

import { FastifyInstance } from 'fastify';  

import { StageMapper } from '../../services/stageMapper.js';  

  

export async function retryRoutes(fastify: FastifyInstance) {  

    const stageMapper = fastify.diContainer.resolve<StageMapper>('stageMapper');  

  

    // Single file retry  

    fastify.post('/retry/:fileId', async (request, reply) => {  

        const { fileId } = request.params as { fileId: string };  

        const { reason, initiatedBy } = request.body as { reason: string; initiatedBy: string };  

         

        const requestId = await stageMapper.retryExtraction(  

            fileId,  

            request.clientApplicationId,  

            reason,  

            initiatedBy  

        );  

         

        return { requestId, fileId, status: 'queued' };  

    });  

  

    // Batch file retry  

    fastify.post('/retry/batch', async (request, reply) => {  

        const { fileIds, reason, initiatedBy } = request.body as {  

            fileIds: string[];  

            reason: string;  

            initiatedBy: string;  

        };  

         

        const requestId = await stageMapper.retryExtractionBatch(  

            fileIds,  

            request.clientApplicationId,  

            reason,  

            initiatedBy  

        );  

         

        return { requestId, fileIds, count: fileIds.length, status: 'queued' };  

    });  

}  

```

---

## Testing Strategy

### Unit Tests

#### Test 1: Message Formatting

**File**: `test/unit/services/messaging/formatRetryMessage.test.ts`

```typescript

import { expect, it } from 'vitest';  

import { formatRetryMessage } from '../../../../src/services/messaging/formatRetryMessage.js';  

import { ExtractionRetryEvent } from '../../../../src/types/messaging.js';  

  

it('must format timestamp as ISO string', () => {  

    const event: ExtractionRetryEvent = {  

        requestId: 'test-request-id',  

        fileIds: ['file-1'],  

        reason: 'test_retry',  

        retryAttempt: 1,  

        maxRetries: 3,  

        clientApplicationId: 'test-client',  

        initiatedBy: 'test-user',  

        timestamp: new Date('2024-01-01T00:00:00.000Z'),  

    };  

  

    const formatted = formatRetryMessage(event);  

  

    expect(formatted.timestamp).toBe('2024-01-01T00:00:00.000Z');  

    expect(formatted.requestId).toBe('test-request-id');  

    expect(formatted.fileIds).toEqual(['file-1']);  

});  

```

#### Test 2: Send Retry Event

**File**: `test/unit/services/messaging/sendExtractionRetryEvent.test.ts`

```typescript

import { expect, it, vi, beforeEach } from 'vitest';  

import { sendExtractionRetryEvent } from '../../../../src/services/messaging/sendExtractionRetryEvent.js';  

import { createSendToExtractionRetryTopic } from '../../../../src/services/messaging/createSendToExtractionRetryTopic.js';  

import { ExtractionRetryEvent } from '../../../../src/types/messaging.js';  

import { app } from '../../../../src/internal/app.js';  

  

vi.mock('../../../../src/services/messaging/createSendToExtractionRetryTopic.js');  

vi.mock('../../../../src/internal/app.js', () => ({  

    app: {  

        log: {  

            info: vi.fn(),  

        },  

    },  

}));  

  

beforeEach(() => {  

    vi.clearAllMocks();  

});  

  

it('must send formatted message to topic', async () => {  

    const mockSender = vi.fn().mockResolvedValue(undefined);  

    vi.mocked(createSendToExtractionRetryTopic).mockResolvedValue(mockSender);  

  

    const event: ExtractionRetryEvent = {  

        requestId: 'test-request',  

        fileIds: ['file-1', 'file-2'],  

        reason: 'extraction_failed',  

        retryAttempt: 1,  

        maxRetries: 3,  

        clientApplicationId: 'test-client',  

        initiatedBy: 'system',  

        timestamp: new Date(),  

    };  

  

    await sendExtractionRetryEvent(event);  

  

    expect(mockSender).toHaveBeenCalledWith(  

        expect.objectContaining({  

            requestId: 'test-request',  

            fileIds: ['file-1', 'file-2'],  

            timestamp: expect.any(String),  

        })  

    );  

  

    expect(app.log.info).toHaveBeenCalledWith(  

        expect.stringContaining('Extraction retry event sent'),  

        expect.objectContaining({  

            requestId: 'test-request',  

            fileIds: ['file-1', 'file-2'],  

        })  

    );  

});  

```

#### Test 3: Extraction Retry Handler

**File**: `test/unit/subscriptions/extractionRetryHandler.test.ts`

```typescript

import { it, expect, vi, beforeEach } from 'vitest';  

import { mock } from 'vitest-mock-extended';  

import { ServiceBusReceivedMessage } from '@azure/service-bus';  

import { ExtractionRetryHandler } from '../../../src/subscriptions/extractionRetryHandler.js';  

import { ExtractionService } from '../../../src/services/extractionService.js';  

import { FileService } from '../../../src/services/fileService.js';  

import { sendExtractionRetryResult } from '../../../src/services/messaging/sendExtractionRetryResult.js';  

  

vi.mock('../../../src/services/messaging/sendExtractionRetryResult.js');  

vi.mock('../../../src/internal/app.js', () => ({  

    app: {  

        log: {  

            info: vi.fn(),  

            error: vi.fn(),  

        },  

    },  

}));  

  

const extractionService = mock<ExtractionService>();  

const fileService = mock<FileService>();  

  

beforeEach(() => {  

    vi.clearAllMocks();  

});  

  

it('must process retry message for single file', async () => {  

    const handler = new ExtractionRetryHandler(extractionService, fileService);  

     

    const message = {  

        body: {  

            requestId: 'test-request',  

            fileIds: ['file-1'],  

            reason: 'extraction_failed',  

            retryAttempt: 1,  

            maxRetries: 3,  

            clientApplicationId: 'test-client',  

            initiatedBy: 'system',  

            timestamp: new Date().toISOString(),  

        },  

    } as ServiceBusReceivedMessage;  

  

    fileService.getFileById.mockResolvedValue({  

        fileId: 'file-1',  

        clientApplicationId: 'test-client',  

    });  

  

    extractionService.extractDocument.mockResolvedValue(undefined);  

  

    await handler.messageHandler.processMessage(message);  

  

    expect(fileService.getFileById).toHaveBeenCalledWith('file-1');  

    expect(extractionService.extractDocument).toHaveBeenCalledWith('file-1');  

    expect(sendExtractionRetryResult).toHaveBeenCalledWith(  

        expect.objectContaining({  

            requestId: 'test-request',  

            successfulIds: ['file-1'],  

            failedIds: [],  

        })  

    );  

});  

  

it('must handle multiple files in batch', async () => {  

    const handler = new ExtractionRetryHandler(extractionService, fileService);  

     

    const message = {  

        body: {  

            requestId: 'batch-request',  

            fileIds: ['file-1', 'file-2', 'file-3'],  

            reason: 'manual_retry',  

            retryAttempt: 1,  

            maxRetries: 3,  

            clientApplicationId: 'test-client',  

            initiatedBy: 'admin@example.com',  

            timestamp: new Date().toISOString(),  

        },  

    } as ServiceBusReceivedMessage;  

  

    fileService.getFileById.mockResolvedValue({ fileId: 'file-1' });  

    extractionService.extractDocument.mockResolvedValue(undefined);  

  

    await handler.messageHandler.processMessage(message);  

  

    expect(extractionService.extractDocument).toHaveBeenCalledTimes(3);  

    expect(sendExtractionRetryResult).toHaveBeenCalledWith(  

        expect.objectContaining({  

            requestId: 'batch-request',  

            fileIds: ['file-1', 'file-2', 'file-3'],  

            successfulIds: ['file-1', 'file-2', 'file-3'],  

        })  

    );  

});  

  

it('must handle partial failures in batch', async () => {  

    const handler = new ExtractionRetryHandler(extractionService, fileService);  

     

    const message = {  

        body: {  

            requestId: 'batch-request',  

            fileIds: ['file-1', 'file-2'],  

            reason: 'manual_retry',  

            retryAttempt: 1,  

            maxRetries: 3,  

            clientApplicationId: 'test-client',  

            initiatedBy: 'admin',  

            timestamp: new Date().toISOString(),  

        },  

    } as ServiceBusReceivedMessage;  

  

    fileService.getFileById.mockResolvedValue({ fileId: 'file-1' });  

     

    // First file succeeds, second fails  

    extractionService.extractDocument  

        .mockResolvedValueOnce(undefined)  

        .mockRejectedValueOnce(new Error('Extraction failed'));  

  

    await handler.messageHandler.processMessage(message);  

  

    expect(sendExtractionRetryResult).toHaveBeenCalledWith(  

        expect.objectContaining({  

            requestId: 'batch-request',  

            successfulIds: ['file-1'],  

            failedIds: ['file-2'],  

            errors: {  

                'file-2': 'Extraction failed',  

            },  

        })  

    );  

});  

  

it('must throw if fileIds is missing', async () => {  

    const handler = new ExtractionRetryHandler(extractionService, fileService);  

     

    const message = {  

        body: {  

            requestId: 'test-request',  

            fileIds: [],  

        },  

    } as ServiceBusReceivedMessage;  

  

    await expect(handler.messageHandler.processMessage(message)).rejects.toThrow(  

        /must contain at least one fileId/  

    );  

});  

  

it('must throw if requestId is missing', async () => {  

    const handler = new ExtractionRetryHandler(extractionService, fileService);  

     

    const message = {  

        body: {  

            fileIds: ['file-1'],  

        },  

    } as ServiceBusReceivedMessage;  

  

    await expect(handler.messageHandler.processMessage(message)).rejects.toThrow(  

        /must contain a requestId/  

    );  

});  

  

it('must enforce max retry limit', async () => {  

    const handler = new ExtractionRetryHandler(extractionService, fileService);  

     

    const message = {  

        body: {  

            requestId: 'test-request',  

            fileIds: ['file-1'],  

            retryAttempt: 4,  

            maxRetries: 3,  

            clientApplicationId: 'test-client',  

            initiatedBy: 'system',  

        },  

    } as ServiceBusReceivedMessage;  

  

    fileService.getFileById.mockResolvedValue({ fileId: 'file-1' });  

  

    await handler.messageHandler.processMessage(message);  

  

    expect(extractionService.extractDocument).not.toHaveBeenCalled();  

    expect(sendExtractionRetryResult).toHaveBeenCalledWith(  

        expect.objectContaining({  

            failedIds: ['file-1'],  

            errors: {  

                'file-1': expect.stringContaining('Maximum retry attempts'),  

            },  

        })  

    );  

});  

```

#### Test 4: Stage Mapper Retry Methods

**File**: `test/unit/services/stageMapper.test.ts`

```typescript

import { expect, it, vi, beforeEach } from 'vitest';  

import { StageMapper } from '../../../src/services/stageMapper.js';  

import { sendExtractionRetryEvent } from '../../../src/services/messaging/sendExtractionRetryEvent.js';  

  

vi.mock('../../../src/services/messaging/sendExtractionRetryEvent.js');  

  

beforeEach(() => {  

    vi.clearAllMocks();  

});  

  

it('must send retry event for single file', async () => {  

    vi.mocked(sendExtractionRetryEvent).mockResolvedValue(undefined);  

     

    const stageMapper = new StageMapper();  

     

    const requestId = await stageMapper.retryExtraction(  

        'file-123',  

        'client-abc',  

        'extraction_failed',  

        'system'  

    );  

  

    expect(requestId).toBeDefined();  

    expect(sendExtractionRetryEvent).toHaveBeenCalledWith(  

        expect.objectContaining({  

            requestId,  

            fileIds: ['file-123'],  

            reason: 'extraction_failed',  

            clientApplicationId: 'client-abc',  

            initiatedBy: 'system',  

            retryAttempt: 1,  

            maxRetries: 3,  

        })  

    );  

});  

  

it('must send retry event for batch of files', async () => {  

    vi.mocked(sendExtractionRetryEvent).mockResolvedValue(undefined);  

     

    const stageMapper = new StageMapper();  

     

    const fileIds = ['file-1', 'file-2', 'file-3'];  

    const requestId = await stageMapper.retryExtractionBatch(  

        fileIds,  

        'client-abc',  

        'manual_retry',  

        'admin@example.com'  

    );  

  

    expect(requestId).toBeDefined();  

    expect(sendExtractionRetryEvent).toHaveBeenCalledWith(  

        expect.objectContaining({  

            requestId,  

            fileIds,  

            reason: 'manual_retry',  

            initiatedBy: 'admin@example.com',  

        })  

    );  

});  

  

it('must throw if batch is empty', async () => {  

    const stageMapper = new StageMapper();  

     

    await expect(  

        stageMapper.retryExtractionBatch(  

            [],  

            'client-abc',  

            'test',  

            'system'  

        )  

    ).rejects.toThrow(/empty file list/);  

});  

```

---

### Contract Tests

#### Test 5: Service Bus Publish

**File**: `test/contract/servicebus-publish.test.ts`

```typescript

import { ServiceBusClient } from '@azure/service-bus';  

import dotenv from 'dotenv';  

import { DevFramework } from 'framework-nodejs-core';  

import path from 'node:path';  

import { expect, it } from 'vitest';  

import { ENV_KEYS } from '../../src/constants/configuration.js';  

import { app } from '../../src/internal/app.js';  

import { ExtractionRetryEvent } from '../../src/types/messaging.js';  

  

dotenv.config({ path: path.resolve(__dirname, '../../.env.local.example') });  

  

const frameworkApp = await DevFramework.create();  

await frameworkApp.initAppConfig({ useEnvironmentVariables: true });  

await app.prepareLookup(frameworkApp);  

app.registerLogger(frameworkApp);  

  

const serviceBusConnStr = await app.lookup(ENV_KEYS.AZURE_SERVICE_BUS_CONNECTION_STRING);  

  

it('can publish extraction retry message to topic', async () => {  

    const sbClient = new ServiceBusClient(serviceBusConnStr);  

    const sender = sbClient.createSender(await app.lookup(ENV_KEYS.TOPIC_EXTRACTION_RETRY));  

     

    const testMessage: ExtractionRetryEvent = {  

        requestId: 'test-request-123',  

        fileIds: ['file-abc-123'],  

        reason: 'test_publish',  

        retryAttempt: 1,  

        maxRetries: 3,  

        clientApplicationId: 'test-client',  

        initiatedBy: 'test-system',  

        timestamp: new Date('2024-01-01T00:00:00.000Z'),  

    };  

  

    await expect(sender.sendMessages([{ body: testMessage }])).resolves.not.toThrow();  

     

    await sender.close();  

    await sbClient.close();  

});  

```

#### Test 6: Service Bus Subscribe

**File**: `test/contract/servicebus-subscribe.test.ts`

```typescript

import { ServiceBusClient } from '@azure/service-bus';  

import dotenv from 'dotenv';  

import { DevFramework } from 'framework-nodejs-core';  

import Long from 'long';  

import path from 'node:path';  

import { expect, it } from 'vitest';  

import { ENV_KEYS } from '../../src/constants/configuration.js';  

import { app } from '../../src/internal/app.js';  

  

dotenv.config({ path: path.resolve(__dirname, '../../.env.local.example') });  

  

const frameworkApp = await DevFramework.create();  

await frameworkApp.initAppConfig({ useEnvironmentVariables: true });  

await app.prepareLookup(frameworkApp);  

app.registerLogger(frameworkApp);  

  

const serviceBusConnStr = await app.lookup(ENV_KEYS.AZURE_SERVICE_BUS_CONNECTION_STRING);  

const topicName = await app.lookup(ENV_KEYS.TOPIC_EXTRACTION_RETRY);  

const subscriptionName = await app.lookup(ENV_KEYS.SUBSCRIPTION_EXTRACTION_RETRY);  

  

it('can peek messages from extraction-retry subscription', async () => {  

    const sbClient = new ServiceBusClient(serviceBusConnStr);  

    const receiver = sbClient.createReceiver(topicName, subscriptionName, {  

        receiveMode: 'peekLock',  

    });  

  

    const messages = await receiver.peekMessages(10, {  

        fromSequenceNumber: Long.fromNumber(1),  

    });  

  

    await receiver.close();  

    await sbClient.close();  

  

    expect(Array.isArray(messages)).toBe(true);  

});  

```

---

### Integration Tests

#### Test 7: End-to-End Message Flow

**File**: `test/integration/extraction-retry-flow.test.ts`

```typescript

import { expect, it, beforeAll, afterAll, vi } from 'vitest';  

import { DevFramework } from 'framework-nodejs-core';  

import { app } from '../../src/internal/app.js';  

import { StageMapper } from '../../src/services/stageMapper.js';  

import { MessageReceiver } from '../../src/subscriptions/messageReceiver.js';  

import { TopicService } from '../../src/services/topicService.js';  

import { ENV_KEYS } from '../../src/constants/configuration.js';  

  

let messageReceiver: MessageReceiver;  

let frameworkApp: DevFramework;  

  

beforeAll(async () => {  

    frameworkApp = await DevFramework.create();  

    await frameworkApp.initAppConfig({ useEnvironmentVariables: true });  

    await app.prepareLookup(frameworkApp);  

    app.registerLogger(frameworkApp);  

  

    const topicService = frameworkApp.diContainer.resolve<TopicService>('topicService');  

    const serviceBusConnectionString = await app.lookup(ENV_KEYS.AZURE_SERVICE_BUS_CONNECTION_STRING);  

     

    messageReceiver = new MessageReceiver(topicService, serviceBusConnectionString);  

    await messageReceiver.startListening();  

});  

  

afterAll(async () => {  

    await messageReceiver.stopListening();  

});  

  

it('should successfully process extraction retry end-to-end', async () => {  

    const stageMapper = new StageMapper();  

     

    // Publish retry message  

    const requestId = await stageMapper.retryExtraction(  

        'test-file-123',  

        'test-client',  

        'integration_test',  

        'test-system'  

    );  

  

    expect(requestId).toBeDefined();  

  

    // Wait for message processing (in real scenario, would use proper async wait)  

    await new Promise(resolve => setTimeout(resolve, 2000));  

  

    // Verify message was processed (check logs or database state)  

    // This is environment-specific  

});  

```

---

## Acceptance Criteria

### Functional Requirements - Restart Endpoint

* [ ] REST API endpoint `POST /v1/extracted_entities/restart` is implemented

* [ ] Endpoint accepts array of extracted entity IDs (1-100 items)

* [ ] Endpoint accepts optional reason parameter for restart

* [ ] Endpoint validates all entity IDs exist in database

* [ ] Endpoint retrieves document hash for each entity via Documents table join (read-only)

* [ ] Endpoint sends extraction retry message to Fluxx Pub/Sub with file hashes

* [ ] Endpoint returns unique request ID for tracking

* [ ] Endpoint returns list of file hashes sent to Fluxx

* [ ] Endpoint returns partial success status when some entities fail validation

* [ ] Endpoint does not modify database (read-only operation)

* [ ] Endpoint prevents restart of entities currently in 'processing' status

* [ ] Endpoint requires authentication and authorization (EXTRACTED\_ENTITIES\_WRITE scope)

* [ ] Endpoint returns 400 for invalid input (empty array, >100 items)

* [ ] Endpoint returns 404 when entities not found

* [ ] Endpoint returns 500 for internal errors with proper logging

### Functional Requirements - Message Publishing

* [ ] Stage mapper can publish retry message for single file

* [ ] Stage mapper can publish retry message for batch of files

* [ ] Extraction retry handler receives and processes messages from topic

* [ ] Retry handler validates message structure (fileIds, requestId required)

* [ ] Retry handler enforces maximum retry attempt limit

* [ ] Retry handler processes each file in batch independently

* [ ] Partial failures in batch are handled gracefully

* [ ] Success and failure results are published to result topic

* [ ] All retry operations are logged with request context

* [ ] Message receiver can be started and stopped cleanly

### Technical Requirements - Restart Endpoint

* [ ] Schema defined in `extractedEntities.schema.ts` with proper validation

* [ ] Types defined in `extractedEntities.types.ts` (RestartExtractionRequest, RestartExtractionResponse, ExtractedEntityWithDocument)

* [ ] Model methods added: `getEntitiesWithDocumentsByIds()` (read-only)

* [ ] Service method added: `restartExtraction()` with full business logic

* [ ] Controller method added: `restartExtraction()` with proper error handling

* [ ] Route added with authentication and authorization middleware

* [ ] No database writes performed (read-only operation)

* [ ] Proper SQL joins for read queries between ExtractedEntities and Documents tables

* [ ] Parameterized SQL queries to prevent SQL injection

* [ ] Error handling for message sending

### Technical Requirements - Message Infrastructure

* [ ] Environment variables defined for Service Bus connection and topics

* [ ] `ExtractionRetryEvent` message type defined

* [ ] `ExtractionRetryResult` message type defined

* [ ] Publisher infrastructure created (createServiceBusPublisher, etc.)

* [ ] Subscriber infrastructure created (ExtractionRetryHandler, MessageReceiver)

* [ ] TopicService configured with retry handler

* [ ] Message receiver initialized in application startup

* [ ] Error handling for missing or malformed messages

* [ ] TypeScript type safety for all message operations

### Code Quality Requirements

* [ ] All message publishing uses framework logger

* [ ] Structured logging includes request context (requestId, fileIds, etc.)

* [ ] No sensitive data logged in messages

* [ ] Consistent error messages for validation failures

* [ ] Proper async/await usage throughout

* [ ] Message formatting consistent with existing patterns

* [ ] Code follows existing architectural patterns from fluxx-file-intake

* [ ] No use of `any` type without documentation

* [ ] SOLID principles followed in service and controller design

* [ ] Repository pattern used for data access

### Testing Requirements - Restart Endpoint

* [ ] Unit test: Model.getEntitiesWithDocumentsByIds() returns correct data (read-only)

* [ ] Unit test: Service.restartExtraction() successfully sends messages for valid entities

* [ ] Unit test: Service.restartExtraction() handles entities not found

* [ ] Unit test: Service.restartExtraction() rejects currently processing entities

* [ ] Unit test: Service.restartExtraction() handles partial failures

* [ ] Unit test: Service.restartExtraction() handles message send failure

* [ ] Unit test: Service.restartExtraction() rejects >100 entities

* [ ] Unit test: Controller.restartExtraction() calls service correctly

* [ ] Integration test: POST /restart returns 200 for valid request

* [ ] Integration test: POST /restart returns 400 for invalid request

* [ ] Integration test: POST /restart returns proper response structure

* [ ] Integration test: POST /restart requires authentication

* [ ] Integration test: POST /restart requires correct scope

### Testing Requirements - Message Infrastructure

* [ ] Unit test: formatRetryMessage correctly formats timestamps

* [ ] Unit test: sendExtractionRetryEvent publishes to correct topic

* [ ] Unit test: ExtractionRetryHandler processes single file

* [ ] Unit test: ExtractionRetryHandler processes batch of files

* [ ] Unit test: ExtractionRetryHandler handles partial failures

* [ ] Unit test: ExtractionRetryHandler validates required fields

* [ ] Unit test: ExtractionRetryHandler enforces retry limit

* [ ] Unit test: StageMapper.retryExtraction sends correct message

* [ ] Unit test: StageMapper.retryExtractionBatch handles batch

* [ ] Unit test: StageMapper.retryExtractionBatch rejects empty batch

* [ ] Contract test: Can publish to extraction-retry topic

* [ ] Contract test: Can subscribe to extraction-retry subscription

* [ ] Integration test: End-to-end message flow works

* [ ] All tests pass in CI/CD pipeline

* [ ] Test coverage > 85% for new messaging modules and restart endpoint

### Documentation Requirements

* [ ] Environment variables documented in .env.local.example

* [ ] README updated with Service Bus configuration instructions

* [ ] Message types documented with JSDoc comments

* [ ] Retry flow diagram included in documentation

* [ ] API endpoints documented (if implemented)

* [ ] Team training materials created

### Deployment Requirements

* [ ] Azure Service Bus namespace created in all environments

* [ ] Topics created: extraction-retry, extraction-result

* [ ] Subscriptions created with correct configuration

* [ ] Environment variables configured in all environments

* [ ] Connection strings stored securely in Azure Key Vault

* [ ] Application successfully starts with message receiver

* [ ] Messages flow through system in production

* [ ] Monitoring and alerting configured for Service Bus

* [ ] Rollback plan documented

---

## Environment Variables Reference

### Required Variables

\| Variable | Description | Example | Required |

\|----------|-------------|---------|----------|

\| `AZURE_SERVICE_BUS_CONNECTION_STRING` | Connection string for Azure Service Bus namespace | `Endpoint=sb://myapp.servicebus.windows.net/;...` | Yes |

\| `TOPIC_EXTRACTION_RETRY` | Name of topic for extraction retry events | `extraction-retry` | Yes |

\| `TOPIC_EXTRACTION_RESULT` | Name of topic for extraction result events | `extraction-result` | Yes |

\| `SUBSCRIPTION_EXTRACTION_RETRY` | Name of subscription for retry handler | `extractor-retry-subscription` | Yes |

\| `SUBSCRIPTION_EXTRACTION_RESULT` | Name of subscription for result handler | `mapper-result-subscription` | Optional |

### Development Environment

```bash

# Local development with emulator  

AZURE_SERVICE_BUS_CONNECTION_STRING='Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;'  

TOPIC_EXTRACTION_RETRY=extraction-retry  

TOPIC_EXTRACTION_RESULT=extraction-result  

SUBSCRIPTION_EXTRACTION_RETRY=extractor-retry-subscription  

SUBSCRIPTION_EXTRACTION_RESULT=mapper-result-subscription  

```

### Production Environment

```bash

# Production with real Azure Service Bus  

AZURE_SERVICE_BUS_CONNECTION_STRING='Endpoint=sb://prod-docai.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=<from-key-vault>;'  

TOPIC_EXTRACTION_RETRY=extraction-retry  

TOPIC_EXTRACTION_RESULT=extraction-result  

SUBSCRIPTION_EXTRACTION_RETRY=extractor-retry-subscription  

SUBSCRIPTION_EXTRACTION_RESULT=mapper-result-subscription  

```

### Azure Service Bus Configuration

* **Topic TTL**: 7 days (configurable)

* **Max Delivery Count**: 10 (configurable)

* **Lock Duration**: 5 minutes

* **Duplicate Detection**: Enabled (1 minute window)

* **Dead Letter Queue**: Enabled

---

## Non-Functional Requirements

### Performance

* Message publishing should complete in \< 100ms

* Message processing should handle 100+ messages/second

* Batch operations should support up to 100 files per request

* No blocking operations during message handling

### Reliability

* Messages should be delivered at-least-once

* Failed messages should be retried automatically

* Dead letter queue for messages exceeding max retries

* Graceful degradation if Service Bus unavailable

### Observability

* All message operations logged with structured context

* Request IDs preserved across async operations

* Azure Application Insights integration

* Service Bus metrics monitored (queue length, processing time)

### Security

* Connection strings stored in Azure Key Vault

* Messages do not contain sensitive data (PII, credentials)

* RBAC configured for Service Bus access

* Network security rules configured

---

## Risks and Mitigation

### Risk 1: Message Processing Failures

**Impact**: High - Messages lost or stuck in queue

**Mitigation**:

* Implement dead letter queue handling

* Configure max delivery count

* Monitor dead letter queue size

* Alert on processing failures

### Risk 2: Service Bus Unavailability

**Impact**: High - Cannot publish or process messages

**Mitigation**:

* Implement retry logic with exponential backoff

* Cache messages locally if Service Bus down

* Graceful degradation for non-critical operations

* Monitor Service Bus health

### Risk 3: Duplicate Message Processing

**Impact**: Medium - Files extracted multiple times

**Mitigation**:

* Enable duplicate detection on Service Bus

* Implement idempotent extraction logic

* Check file state before processing

* Use request ID for deduplication

### Risk 4: Batch Processing Failures

**Impact**: Medium - Partial batch failures

**Mitigation**:

* Process files independently in batch

* Track success/failure per file

* Publish detailed result messages

* Support retry of failed files only

---

## Rollout Plan

### Phase 1: Infrastructure Setup (Week 1)

1. Create Azure Service Bus namespace (Dev environment)

2. Create topics and subscriptions

3. Configure connection strings

4. Implement publisher infrastructure

5. Implement subscriber infrastructure

6. Update database schema to support 'restarting' status

7. Deploy to Dev environment

### Phase 2: Message Integration (Week 2)

1. Integrate with StageMapper

2. Integrate with ExtractionService

3. Add unit tests for messaging

4. Add contract tests

5. Deploy to QA environment

### Phase 3: REST API Implementation (Week 2-3)

1. Implement POST /extracted\_entities/restart endpoint

\- Add schema, types, model methods

\- Add service and controller methods

\- Add route with authentication

2. Add comprehensive unit tests for endpoint

3. Add integration tests for endpoint

4. Test end-to-end flow: API → Status Update → Message → Fluxx

5. Deploy to QA environment

### Phase 4: Testing (Week 3)

1. End-to-end testing in QA

2. Performance testing (message publishing)

3. Load testing (batch operations with 100 entities)

4. Chaos testing (Service Bus failures, rollback scenarios)

5. User acceptance testing of restart endpoint

6. Deploy to UAT environment

### Phase 5: Production Deployment (Week 4)

1. Create production Service Bus resources

2. Configure production environment variables

3. Deploy database schema changes to production

4. Deploy to production with feature flag

5. Monitor message flow and endpoint usage

6. Enable for all users

7. Post-deployment validation

---

## API Usage Examples

### Example 1: Restart Single Extraction

```bash

POST /v1/extracted_entities/restart  

Authorization: Bearer <token>  

Content-Type: application/json  

  

{  

  "extractedEntityIds": [12345],  

  "reason": "manual_retry - extraction failed due to network timeout"  

}  

```

**Response (200 OK)**:

```json

{  

  "requestId": "a7f8e3d2-4c5b-4e6f-8a9b-c1d2e3f4g5h6",  

  "extractedEntityIds": [12345],  

  "fileHashes": ["abc123def456"],  

  "status": "accepted",  

  "message": "Successfully initiated restart for 1 extraction(s)"  

}  

```

### Example 2: Restart Multiple Extractions (Batch)

```bash

POST /v1/extracted_entities/restart  

Authorization: Bearer <token>  

Content-Type: application/json  

  

{  

  "extractedEntityIds": [100, 101, 102, 103],  

  "reason": "manual_restart"  

}  

```

**Response (200 OK)**:

```json

{  

  "requestId": "b8g9h4e3-5d6c-5f7g-9b0c-d2e3f4g5h6i7",  

  "extractedEntityIds": [100, 101, 102, 103],  

  "fileHashes": ["hash1", "hash2", "hash3", "hash4"],  

  "status": "accepted",  

  "message": "Successfully initiated restart for 4 extraction(s)"  

}  

```

### Example 3: Partial Success (Some Entities Not Found)

```bash

POST /v1/extracted_entities/restart  

Authorization: Bearer <token>  

Content-Type: application/json  

  

{  

  "extractedEntityIds": [200, 999, 201],  

  "reason": "retry_failed_extractions"  

}  

```

**Response (200 OK)**:

```json

{  

  "requestId": "c9h0i5f4-6e7d-6g8h-0c1d-e3f4g5h6i7j8",  

  "extractedEntityIds": [200, 201],  

  "fileHashes": ["hash200", "hash201"],  

  "status": "partial",  

  "message": "Initiated restart for 2 extraction(s) with 1 error(s)",  

  "errors": [  

    {  

      "extractedEntityId": 999,  

      "error": "Extracted entity with ID 999 not found"  

    }  

  ]  

}  

```

### Example 4: Cannot Restart Processing Entity

```bash

POST /v1/extracted_entities/restart  

Authorization: Bearer <token>  

Content-Type: application/json  

  

{  

  "extractedEntityIds": [300],  

  "reason": "manual_restart"  

}  

```

**Response (200 OK)**:

```json

{  

  "requestId": "d0i1j6g5-7f8e-7h9i-1d2e-f4g5h6i7j8k9",  

  "extractedEntityIds": [],  

  "fileHashes": [],  

  "status": "failed",  

  "message": "No valid entities to restart",  

  "errors": [  

    {  

      "extractedEntityId": 300,  

      "error": "Cannot restart entity that is currently processing"  

    }  

  ]  

}  

```

### Example 5: Invalid Request - Empty Array

```bash

POST /v1/extracted_entities/restart  

Authorization: Bearer <token>  

Content-Type: application/json  

  

{  

  "extractedEntityIds": []  

}  

```

**Response (400 Bad Request)**:

```json

{  

  "error": "Bad Request",  

  "message": "extractedEntityIds is required and must contain at least one ID"  

}  

```

### Example 6: Invalid Request - Too Many Entities

```bash

POST /v1/extracted_entities/restart  

Authorization: Bearer <token>  

Content-Type: application/json  

  

{  

  "extractedEntityIds": [1, 2, 3, ..., 101]  // 101 items  

}  

```

**Response (400 Bad Request)**:

```json

{  

  "error": "Bad Request",  

  "message": "Cannot restart more than 100 entities at once"  

}  

```

---

## Success Metrics

### Technical Metrics - Messaging

* 99.9% message delivery success rate

* \< 100ms average message publishing latency

* \< 5 seconds average message processing time

* Zero message loss

* \< 0.1% messages in dead letter queue

### Technical Metrics - REST API

* \< 500ms average response time for restart endpoint

* 99.5% success rate for valid requests

* Proper status code usage (200, 400, 404, 500)

* Zero data inconsistencies (status updates + messages)

* Database transaction rollback works correctly on failures

### Business Metrics

* Automated retry reduces manual intervention by 80%

* Failed extractions resolved within 5 minutes

* Batch operations support up to 100 entities efficiently

* Mean time to retry reduced from 30min to 5min

* Operators can restart extractions via API without database access

### Operational Metrics

* Service Bus uptime > 99.9%

* All retry requests tracked and logged

* Zero production incidents from messaging or restart endpoint

* Monitoring dashboards fully configured

* Clear audit trail for all restart operations

---

## Related Documentation

* [fluxx-file-intake Repository](https://github.com/CLADevOps/fluxx-file-intake)

* [Azure Service Bus Documentation](https://learn.microsoft.com/en-us/azure/service-bus-messaging/)

* [Azure Service Bus Node.js SDK](https://www.npmjs.com/package/@azure/service-bus)

* [Message-based Architecture Patterns](https://learn.microsoft.com/en-us/azure/architecture/patterns/messaging)

---

## Story Points: 21

**Complexity**: High

**Effort**: 5 weeks

**Dependencies**: Azure Service Bus, framework-nodejs-extensions, fluxx-file-intake patterns

**Additional Scope**: REST API endpoint for extraction restart with batch support

---

## Definition of Done

* [ ] All acceptance criteria met

* [ ] All functional requirements implemented (messaging + REST API)

* [ ] All technical requirements completed

* [ ] All tests passing (unit, contract, integration) with >85% coverage

* [ ] POST /extracted\_entities/restart endpoint fully implemented

* [ ] No database writes performed (read-only operation)

* [ ] Extracted entities correctly linked to document hashes via JOIN

* [ ] Message sending handled with proper error handling

* [ ] Code reviewed and approved

* [ ] Environment variables configured in all environments

* [ ] Azure Service Bus resources created

* [ ] Deployed to all environments (Dev, QA, UAT, Prod)

* [ ] API documentation complete with examples

* [ ] OpenAPI spec updated for restart endpoint

* [ ] Team training completed

* [ ] Messages flowing correctly to Fluxx in production

* [ ] Restart endpoint functioning in production

* [ ] Monitoring configured for both messaging and endpoint

* [ ] Post-deployment validation successful

* [ ] Stakeholder sign-off received