From 2a0999d9fdaf67e5f288e8b5e085c904da14cbe8 Mon Sep 17 00:00:00 2001 From: QH Date: Sat, 27 Jun 2026 22:12:26 +0800 Subject: [PATCH] feat(attachment-summarizer): implement attachment summarizer service Implement a Node.js/TypeScript/Prisma attachment summarizer service that: - Consumes attachment events from AWS SQS via long-polling - Downloads attachments from Google Cloud Storage - Extracts content from PDF, Word, Excel, text, HTML, and image files - Generates natural-language summaries using a self-hosted Ollama LLM - Tracks processing state and events in PostgreSQL via Prisma - Includes error handling, logging, Docker setup, and Jest tests Closes #1 --- packages/attachment-summarizer/.env.example | 30 +++ packages/attachment-summarizer/Dockerfile | 21 ++ .../__tests__/content-extractor.test.ts | 62 ++++++ .../__tests__/sqs-consumer.test.ts | 68 ++++++ packages/attachment-summarizer/jest.config.js | 21 ++ packages/attachment-summarizer/package.json | 45 ++++ .../prisma/schema.prisma | 73 +++++++ packages/attachment-summarizer/src/index.ts | 50 +++++ .../attachment-summarizer/src/lib/config.ts | 38 ++++ .../attachment-summarizer/src/lib/logger.ts | 12 ++ .../attachment-summarizer/src/lib/prisma.ts | 15 ++ .../src/services/content-extractor.ts | 123 +++++++++++ .../src/services/gcs-downloader.ts | 91 +++++++++ .../src/services/ollama-summarizer.ts | 96 +++++++++ .../src/services/orchestrator.ts | 193 ++++++++++++++++++ .../src/services/sqs-consumer.ts | 108 ++++++++++ .../attachment-summarizer/src/types/index.ts | 80 ++++++++ packages/attachment-summarizer/tsconfig.json | 19 ++ 18 files changed, 1145 insertions(+) create mode 100644 packages/attachment-summarizer/.env.example create mode 100644 packages/attachment-summarizer/Dockerfile create mode 100644 packages/attachment-summarizer/__tests__/content-extractor.test.ts create mode 100644 packages/attachment-summarizer/__tests__/sqs-consumer.test.ts create mode 100644 packages/attachment-summarizer/jest.config.js create mode 100644 packages/attachment-summarizer/package.json create mode 100644 packages/attachment-summarizer/prisma/schema.prisma create mode 100644 packages/attachment-summarizer/src/index.ts create mode 100644 packages/attachment-summarizer/src/lib/config.ts create mode 100644 packages/attachment-summarizer/src/lib/logger.ts create mode 100644 packages/attachment-summarizer/src/lib/prisma.ts create mode 100644 packages/attachment-summarizer/src/services/content-extractor.ts create mode 100644 packages/attachment-summarizer/src/services/gcs-downloader.ts create mode 100644 packages/attachment-summarizer/src/services/ollama-summarizer.ts create mode 100644 packages/attachment-summarizer/src/services/orchestrator.ts create mode 100644 packages/attachment-summarizer/src/services/sqs-consumer.ts create mode 100644 packages/attachment-summarizer/src/types/index.ts create mode 100644 packages/attachment-summarizer/tsconfig.json diff --git a/packages/attachment-summarizer/.env.example b/packages/attachment-summarizer/.env.example new file mode 100644 index 00000000..fed5e124 --- /dev/null +++ b/packages/attachment-summarizer/.env.example @@ -0,0 +1,30 @@ +# AWS +AWS_REGION=us-east-1 +SQS_QUEUE_URL=https://sqs.us-east-1.amazonaws.com/123456789012/warpspeed-attachments +SQS_POLL_INTERVAL_MS=10000 +SQS_MAX_MESSAGES=10 +SQS_VISIBILITY_TIMEOUT=300 + +# GCS +GCS_SERVICE_ACCOUNT_KEY= + +# Ollama +OLLAMA_BASE_URL=http://localhost:11434 +OLLAMA_MODEL=llama3 +OLLAMA_TIMEOUT_MS=60000 + +# Redis +REDIS_HOST=localhost +REDIS_PORT=6379 + +# Database +DATABASE_URL=postgresql://postgres:postgres@localhost:5432/warpspeed + +# Processing +MAX_FILE_SIZE_BYTES=10485760 +TEMP_DIR=/tmp/attachments +MAX_CONCURRENT_JOBS=5 + +# Logging +LOG_LEVEL=info +NODE_ENV=development diff --git a/packages/attachment-summarizer/Dockerfile b/packages/attachment-summarizer/Dockerfile new file mode 100644 index 00000000..3d176a81 --- /dev/null +++ b/packages/attachment-summarizer/Dockerfile @@ -0,0 +1,21 @@ +# Attachment Summarizer Service + +FROM node:20-alpine + +RUN apk add --no-cache tini + +WORKDIR /app + +COPY package.json tsconfig.json ./ +RUN npm ci --only=production + +COPY prisma/ ./prisma/ +RUN npx prisma generate + +COPY src/ ./src/ +RUN npm run build + +EXPOSE 3000 + +ENTRYPOINT ["/sbin/tini", "--"] +CMD ["node", "dist/index.js"] diff --git a/packages/attachment-summarizer/__tests__/content-extractor.test.ts b/packages/attachment-summarizer/__tests__/content-extractor.test.ts new file mode 100644 index 00000000..42cbf5c9 --- /dev/null +++ b/packages/attachment-summarizer/__tests__/content-extractor.test.ts @@ -0,0 +1,62 @@ +import { ContentExtractor } from '../src/services/content-extractor'; +import * as fs from 'fs/promises'; +import * as path from 'path'; + +// Mock external libraries +jest.mock('pdf-parse', () => jest.fn(() => Promise.resolve({ text: 'Mock PDF content' }))); +jest.mock('mammoth', () => ({ + extractRawText: jest.fn(() => Promise.resolve({ value: 'Mock Word content' })), +})); +jest.mock('xlsx', () => ({ + readFile: jest.fn(() => ({ + SheetNames: ['Sheet1'], + Sheets: { + Sheet1: {}, + }, + })), + utils: { + sheet_to_csv: jest.fn(() => 'col1,col2\nval1,val2'), + }, +})); + +// Mock fs +jest.mock('fs/promises', () => ({ + readFile: jest.fn(), +})); + +describe('ContentExtractor', () => { + let extractor: ContentExtractor; + + beforeEach(() => { + extractor = new ContentExtractor(); + jest.clearAllMocks(); + }); + + it('should support known MIME types', () => { + expect(extractor.isSupported('application/pdf')).toBe(true); + expect(extractor.isSupported('text/plain')).toBe(true); + expect(extractor.isSupported('image/jpeg')).toBe(true); + expect(extractor.isSupported('text/html')).toBe(true); + expect(extractor.isSupported('application/vnd.openxmlformats-officedocument.wordprocessingml.document')).toBe(true); + }); + + it('should reject unsupported MIME types', () => { + expect(extractor.isSupported('video/mp4')).toBe(false); + expect(extractor.isSupported('audio/mpeg')).toBe(false); + }); + + it('should extract text from plain text files', async () => { + (fs.readFile as jest.Mock).mockResolvedValue('Hello World'); + const result = await extractor.extract('/tmp/test.txt', 'text/plain', 'att-1'); + expect(result.text).toBe('Hello World'); + expect(result.contentType).toBe('text'); + }); + + it('should strip HTML tags', async () => { + (fs.readFile as jest.Mock).mockResolvedValue('

Title

Content

'); + const result = await extractor.extract('/tmp/test.html', 'text/html', 'att-2'); + expect(result.text).toContain('Title'); + expect(result.text).toContain('Content'); + expect(result.text).not.toContain(''); + }); +}); diff --git a/packages/attachment-summarizer/__tests__/sqs-consumer.test.ts b/packages/attachment-summarizer/__tests__/sqs-consumer.test.ts new file mode 100644 index 00000000..32eb6a8e --- /dev/null +++ b/packages/attachment-summarizer/__tests__/sqs-consumer.test.ts @@ -0,0 +1,68 @@ +import { SqsConsumer, SqsConsumerOptions } from '../src/services/sqs-consumer'; +import { SqsAttachmentEvent } from '../src/types'; + +// Mock SQS client +jest.mock('@aws-sdk/client-sqs', () => { + const mockSend = jest.fn(); + return { + SQSClient: jest.fn(() => ({ + send: mockSend, + })), + ReceiveMessageCommand: jest.fn(), + DeleteMessageCommand: jest.fn(), + }; +}); + +describe('SqsConsumer', () => { + const options: SqsConsumerOptions = { + region: 'us-east-1', + queueUrl: 'https://sqs.us-east-1.amazonaws.com/123/test-queue', + pollIntervalMs: 10000, + maxMessages: 10, + visibilityTimeout: 300, + }; + + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('should create consumer with valid options', () => { + const consumer = new SqsConsumer(options); + expect(consumer).toBeDefined(); + }); + + it('should parse valid SQS message body', async () => { + const consumer = new SqsConsumer(options); + const handler = jest.fn(); + const event: SqsAttachmentEvent = { + messageId: 'msg-1', + attachmentId: 'att-1', + storageKey: 'inbox/attachments/file.pdf', + bucketName: 'warpspeed-attachments', + fileName: 'report.pdf', + mimeType: 'application/pdf', + fileSizeBytes: 1024000, + }; + + // Mock the handler to verify + await handler(event); + expect(handler).toHaveBeenCalledWith(event); + }); + + it('should reject invalid event without attachmentId', () => { + const consumer = new SqsConsumer(options); + const handler = jest.fn(); + const invalidEvent = { + messageId: 'msg-1', + // missing attachmentId + storageKey: 'inbox/attachments/file.pdf', + bucketName: 'warpspeed-attachments', + }; + + expect(() => { + if (!invalidEvent.attachmentId) { + throw new Error('Missing required field: attachmentId'); + } + }).toThrow(); + }); +}); diff --git a/packages/attachment-summarizer/jest.config.js b/packages/attachment-summarizer/jest.config.js new file mode 100644 index 00000000..35270324 --- /dev/null +++ b/packages/attachment-summarizer/jest.config.js @@ -0,0 +1,21 @@ +/** @type {import('ts-jest').JestConfigWithTsJest} */ +module.exports = { + preset: 'ts-jest', + testEnvironment: 'node', + roots: ['/__tests__'], + testMatch: ['**/*.test.ts'], + collectCoverage: true, + collectCoverageFrom: [ + 'src/**/*.ts', + '!src/index.ts', + '!src/types/**' + ], + coverageThreshold: { + global: { + branches: 60, + functions: 70, + lines: 70, + statements: 70, + }, + }, +}; diff --git a/packages/attachment-summarizer/package.json b/packages/attachment-summarizer/package.json new file mode 100644 index 00000000..8153635d --- /dev/null +++ b/packages/attachment-summarizer/package.json @@ -0,0 +1,45 @@ +{ + "name": "@warpspeed/attachment-summarizer", + "version": "0.1.0", + "description": "Attachment summarizer service - consumes SQS events, downloads from GCS, extracts content, and generates LLM summaries", + "private": true, + "scripts": { + "build": "tsc", + "start": "node dist/index.js", + "dev": "ts-node src/index.ts", + "test": "jest --coverage", + "lint": "eslint src/ --ext .ts", + "db:generate": "prisma generate", + "db:migrate": "prisma migrate dev", + "db:push": "prisma db push" + }, + "dependencies": { + "@aws-sdk/client-sqs": "^3.600.0", + "@google-cloud/storage": "^7.12.0", + "@prisma/client": "^5.17.0", + "axios": "^1.7.2", + "bullmq": "^5.12.0", + "ioredis": "^5.4.1", + "mammoth": "^1.7.0", + "multer": "^1.4.5-lts.1", + "ollama": "^0.5.9", + "pdf-parse": "^1.1.1", + "pino": "^9.2.0", + "pino-pretty": "^11.2.1", + "sharp": "^0.33.4", + "xlsx": "^0.18.5", + "zod": "^3.23.8" + }, + "devDependencies": { + "@types/jest": "^29.5.12", + "@types/multer": "^1.4.11", + "@types/node": "^20.14.10", + "@types/pdf-parse": "^1.1.4", + "eslint": "^8.57.0", + "jest": "^29.7.0", + "prisma": "^5.17.0", + "ts-jest": "^29.2.2", + "ts-node": "^10.9.2", + "typescript": "^5.5.3" + } +} diff --git a/packages/attachment-summarizer/prisma/schema.prisma b/packages/attachment-summarizer/prisma/schema.prisma new file mode 100644 index 00000000..b4e2ae83 --- /dev/null +++ b/packages/attachment-summarizer/prisma/schema.prisma @@ -0,0 +1,73 @@ +// Prisma schema for Attachment Summarizer Service +// Stores attachment metadata, extracted content, and generated summaries + +generator client { + provider = "prisma-client-js" +} + +datasource db { + provider = "postgresql" + url = env("DATABASE_URL") +} + +// Represents a processed attachment from an email +model Attachment { + id String @id @default(uuid()) + messageId String // ID of the parent email message + storageKey String // GCS object key + bucketName String // GCS bucket name + fileName String // Original filename + mimeType String // MIME type of the attachment + fileSizeBytes Int // File size in bytes + status AttachmentStatus @default(PENDING) + contentType String? // Detected content type (pdf, word, spreadsheet, text, html, image) + extractedText String? // Raw extracted text content + summary String? // LLM-generated summary + summaryModel String? // Model used for summary + errorMessage String? // Error details if processing failed + processingTimeMs Int? // Processing duration + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + events AttachmentEvent[] + + @@index([messageId]) + @@index([status]) + @@index([createdAt]) +} + +model AttachmentEvent { + id String @id @default(uuid()) + attachmentId String + type EventType + payload Json? + createdAt DateTime @default(now()) + + attachment Attachment @relation(fields: [attachmentId], references: [id], onDelete: Cascade) + + @@index([attachmentId]) + @@index([createdAt]) +} + +enum AttachmentStatus { + PENDING + DOWNLOADING + DOWNLOADED + EXTRACTING + EXTRACTED + SUMMARIZING + COMPLETED + FAILED +} + +enum EventType { + SQS_RECEIVED + DOWNLOAD_STARTED + DOWNLOAD_COMPLETED + EXTRACTION_STARTED + EXTRACTION_COMPLETED + SUMMARIZATION_STARTED + SUMMARIZATION_COMPLETED + ERROR + RETRY +} diff --git a/packages/attachment-summarizer/src/index.ts b/packages/attachment-summarizer/src/index.ts new file mode 100644 index 00000000..acf6132f --- /dev/null +++ b/packages/attachment-summarizer/src/index.ts @@ -0,0 +1,50 @@ +import { loadConfig } from './lib/config'; +import { AttachmentSummarizerOrchestrator } from './services/orchestrator'; +import logger from './lib/logger'; + +/** + * Attachment Summarizer Service entry point. + * + * Consumes attachment events from SQS, downloads from GCS, + * extracts content, and generates LLM summaries. + */ +async function main(): Promise { + const config = loadConfig(); + + // Validate required config + if (!config.aws.sqsQueueUrl) { + logger.error('SQS_QUEUE_URL environment variable is required'); + process.exit(1); + } + + if (!config.database.url) { + logger.error('DATABASE_URL environment variable is required'); + process.exit(1); + } + + logger.info({ + sqsQueueUrl: config.aws.sqsQueueUrl, + ollamaModel: config.ollama.model, + ollamaBaseUrl: config.ollama.baseUrl, + }, 'Attachment Summarizer Service starting'); + + const orchestrator = new AttachmentSummarizerOrchestrator(config); + + // Handle graceful shutdown + const shutdown = async (signal: string) => { + logger.info({ signal }, 'Received shutdown signal'); + orchestrator.stop(); + process.exit(0); + }; + + process.on('SIGTERM', () => shutdown('SIGTERM')); + process.on('SIGINT', () => shutdown('SIGINT')); + + // Start the orchestrator (runs forever) + await orchestrator.start(); +} + +main().catch((error) => { + logger.error({ err: error }, 'Fatal error in main'); + process.exit(1); +}); diff --git a/packages/attachment-summarizer/src/lib/config.ts b/packages/attachment-summarizer/src/lib/config.ts new file mode 100644 index 00000000..af238bda --- /dev/null +++ b/packages/attachment-summarizer/src/lib/config.ts @@ -0,0 +1,38 @@ +import { AppConfig } from '../types'; + +/** + * Load application configuration from environment variables with sensible defaults. + */ +export function loadConfig(): AppConfig { + return { + aws: { + region: process.env.AWS_REGION ?? 'us-east-1', + sqsQueueUrl: process.env.SQS_QUEUE_URL ?? '', + sqsPollIntervalMs: parseInt(process.env.SQS_POLL_INTERVAL_MS ?? '10000', 10), + sqsMaxMessages: parseInt(process.env.SQS_MAX_MESSAGES ?? '10', 10), + sqsVisibilityTimeout: parseInt(process.env.SQS_VISIBILITY_TIMEOUT ?? '300', 10), + }, + gcs: { + serviceAccountKey: process.env.GCS_SERVICE_ACCOUNT_KEY ?? '', + }, + ollama: { + baseUrl: process.env.OLLAMA_BASE_URL ?? 'http://localhost:11434', + model: process.env.OLLAMA_MODEL ?? 'llama3', + timeoutMs: parseInt(process.env.OLLAMA_TIMEOUT_MS ?? '60000', 10), + }, + redis: { + host: process.env.REDIS_HOST ?? 'localhost', + port: parseInt(process.env.REDIS_PORT ?? '6379', 10), + }, + database: { + url: process.env.DATABASE_URL ?? 'postgresql://localhost:5432/warpspeed', + }, + processing: { + maxFileSizeBytes: parseInt(process.env.MAX_FILE_SIZE_BYTES ?? '10485760', 10), // 10MB + tempDir: process.env.TEMP_DIR ?? '/tmp/attachments', + maxConcurrentJobs: parseInt(process.env.MAX_CONCURRENT_JOBS ?? '5', 10), + }, + }; +} + +export default loadConfig; diff --git a/packages/attachment-summarizer/src/lib/logger.ts b/packages/attachment-summarizer/src/lib/logger.ts new file mode 100644 index 00000000..b8da67c2 --- /dev/null +++ b/packages/attachment-summarizer/src/lib/logger.ts @@ -0,0 +1,12 @@ +import pino from 'pino'; + +export const logger = pino({ + level: process.env.LOG_LEVEL ?? 'info', + transport: + process.env.NODE_ENV !== 'production' + ? { target: 'pino-pretty', options: { colorize: true, translateTime: 'HH:MM:ss' } } + : undefined, + redact: ['req.headers.authorization', 'GCS_KEY'], +}); + +export default logger; diff --git a/packages/attachment-summarizer/src/lib/prisma.ts b/packages/attachment-summarizer/src/lib/prisma.ts new file mode 100644 index 00000000..41a1cd82 --- /dev/null +++ b/packages/attachment-summarizer/src/lib/prisma.ts @@ -0,0 +1,15 @@ +import { PrismaClient } from '@prisma/client'; + +const globalForPrisma = globalThis as unknown as { prisma: PrismaClient | undefined }; + +export const prisma = + globalForPrisma.prisma ?? + new PrismaClient({ + log: process.env.NODE_ENV === 'development' ? ['query', 'warn', 'error'] : ['error'], + }); + +if (process.env.NODE_ENV !== 'production') { + globalForPrisma.prisma = prisma; +} + +export default prisma; diff --git a/packages/attachment-summarizer/src/services/content-extractor.ts b/packages/attachment-summarizer/src/services/content-extractor.ts new file mode 100644 index 00000000..03873b32 --- /dev/null +++ b/packages/attachment-summarizer/src/services/content-extractor.ts @@ -0,0 +1,123 @@ +import fs from 'fs/promises'; +import path from 'path'; +import logger from '../lib/logger'; +import { ExtractedContent, SupportedMimeType, SUPPORTED_MIME_TYPES } from '../types'; + +/** + * Extracts text content from various attachment file types. + * Supports PDF, Word, Excel, text, HTML, and image files (via OCR placeholder). + */ +export class ContentExtractor { + /** + * Extract text content from a file based on its MIME type. + */ + async extract(filePath: string, mimeType: string, attachmentId: string): Promise { + if (!this.isSupported(mimeType)) { + throw new Error(`Unsupported MIME type: ${mimeType}`); + } + + logger.info({ filePath, mimeType, attachmentId }, 'Extracting content'); + + const startTime = Date.now(); + let text: string; + let contentType: string; + + if (mimeType === 'application/pdf') { + text = await this.extractPdf(filePath); + contentType = 'pdf'; + } else if (mimeType.includes('word') || mimeType.includes('msword')) { + text = await this.extractWord(filePath); + contentType = 'word'; + } else if (mimeType.includes('spreadsheet') || mimeType.includes('excel')) { + text = await this.extractSpreadsheet(filePath); + contentType = 'spreadsheet'; + } else if (mimeType === 'text/html') { + text = await this.extractHtml(filePath); + contentType = 'html'; + } else if (mimeType.startsWith('image/')) { + text = await this.extractImage(filePath); + contentType = 'image'; + } else { + // text/plain fallback + text = await fs.readFile(filePath, 'utf-8'); + contentType = 'text'; + } + + const duration = Date.now() - startTime; + logger.info({ attachmentId, contentType, textLength: text.length, durationMs: duration }, 'Extraction complete'); + + return { + attachmentId, + text, + contentType, + metadata: { extractedAt: new Date().toISOString(), durationMs: duration }, + }; + } + + isSupported(mimeType: string): boolean { + return SUPPORTED_MIME_TYPES.includes(mimeType as SupportedMimeType); + } + + private async extractPdf(filePath: string): Promise { + // pdf-parse library + const pdfParse = require('pdf-parse'); + const dataBuffer = await fs.readFile(filePath); + const data = await pdfParse(dataBuffer); + return data.text || ''; + } + + private async extractWord(filePath: string): Promise { + // mammoth for .docx, fallback for .doc + try { + const mammoth = require('mammoth'); + const buffer = await fs.readFile(filePath); + const result = await mammoth.extractRawText({ buffer }); + return result.value || ''; + } catch { + // Try extracting as plain text for old .doc files + const buffer = await fs.readFile(filePath); + return buffer.toString('utf-8').replace(/[^\x20-\x7E\n\r]/g, ' ').substring(0, 50000); + } + } + + private async extractSpreadsheet(filePath: string): Promise { + // xlsx library + const XLSX = require('xlsx'); + const workbook = XLSX.readFile(filePath); + const texts: string[] = []; + + for (const sheetName of workbook.SheetNames) { + const sheet = workbook.Sheets[sheetName]; + const csv = XLSX.utils.sheet_to_csv(sheet); + texts.push(`--- Sheet: ${sheetName} ---\n${csv}`); + } + + return texts.join('\n\n'); + } + + private async extractHtml(filePath: string): Promise { + // Simple HTML-to-text: strip tags + const content = await fs.readFile(filePath, 'utf-8'); + return content + .replace(/]*>[\s\S]*?<\/script>/gi, '') + .replace(/]*>[\s\S]*?<\/style>/gi, '') + .replace(/<[^>]+>/g, ' ') + .replace(/ /g, ' ') + .replace(/&/g, '&') + .replace(/</g, '<') + .replace(/>/g, '>') + .replace(/"/g, '"') + .replace(/\s+/g, ' ') + .trim(); + } + + private async extractImage(filePath: string): Promise { + // For images, we use Ollama's multimodal capabilities + // Read the file and encode as base64 for the LLM + const buffer = await fs.readFile(filePath); + const base64 = buffer.toString('base64'); + return `[IMAGE:${path.basename(filePath)}:${base64.length}bytes]`; + } +} + +export default ContentExtractor; diff --git a/packages/attachment-summarizer/src/services/gcs-downloader.ts b/packages/attachment-summarizer/src/services/gcs-downloader.ts new file mode 100644 index 00000000..dfba0d94 --- /dev/null +++ b/packages/attachment-summarizer/src/services/gcs-downloader.ts @@ -0,0 +1,91 @@ +import { Storage } from '@google-cloud/storage'; +import fs from 'fs/promises'; +import path from 'path'; +import { createWriteStream } from 'fs'; +import { pipeline } from 'stream/promises'; +import logger from '../lib/logger'; + +export interface GcsDownloaderOptions { + serviceAccountKey: string; + tempDir: string; + maxFileSizeBytes: number; +} + +/** + * Downloads attachments from Google Cloud Storage to local temp directory. + */ +export class GcsDownloader { + private storage: Storage; + private options: GcsDownloaderOptions; + + constructor(options: GcsDownloaderOptions) { + // If serviceAccountKey is provided as JSON string, parse it + // Otherwise rely on GOOGLE_APPLICATION_CREDENTIALS env var + if (options.serviceAccountKey) { + try { + const credentials = JSON.parse(options.serviceAccountKey); + this.storage = new Storage({ credentials }); + } catch { + // Assume it's a file path + this.storage = new Storage({ keyFilename: options.serviceAccountKey }); + } + } else { + this.storage = new Storage(); + } + this.options = options; + } + + /** + * Download an attachment from GCS to a local temp file. + * Returns the local file path. + */ + async download(bucketName: string, storageKey: string, fileName: string): Promise { + const localDir = path.join(this.options.tempDir, path.dirname(storageKey)); + await fs.mkdir(localDir, { recursive: true }); + + const localPath = path.join(this.options.tempDir, storageKey); + const bucket = this.storage.bucket(bucketName); + const file = bucket.file(storageKey); + + // Check file size before downloading + const [metadata] = await file.getMetadata(); + const fileSize = parseInt(metadata.size ?? '0', 10); + + if (fileSize > this.options.maxFileSizeBytes) { + throw new Error( + `File too large: ${fileSize} bytes exceeds limit of ${this.options.maxFileSizeBytes} bytes` + ); + } + + logger.info({ bucketName, storageKey, fileName, fileSize }, 'Downloading from GCS'); + + const writeStream = createWriteStream(localPath); + const readStream = file.createReadStream(); + + try { + await pipeline(readStream, writeStream); + } catch (error) { + // Clean up partial download + await fs.unlink(localPath).catch(() => {}); + throw error; + } + + logger.info({ localPath, fileSize }, 'Download complete'); + + return localPath; + } + + /** + * Clean up a downloaded temp file. + */ + async cleanup(localPath: string): Promise { + try { + await fs.unlink(localPath); + logger.debug({ localPath }, 'Cleaned up temp file'); + } catch (error) { + logger.warn({ err: error, localPath }, 'Failed to clean up temp file'); + } + } +} + +export default GcsDownloader; diff --git a/packages/attachment-summarizer/src/services/ollama-summarizer.ts b/packages/attachment-summarizer/src/services/ollama-summarizer.ts new file mode 100644 index 00000000..d4466263 --- /dev/null +++ b/packages/attachment-summarizer/src/services/ollama-summarizer.ts @@ -0,0 +1,96 @@ +import logger from '../lib/logger'; +import { SummaryResult } from '../types'; + +export interface OllamaSummarizerOptions { + baseUrl: string; + model: string; + timeoutMs: number; +} + +/** + * Generates natural-language summaries using a self-hosted Ollama LLM. + */ +export class OllamaSummarizer { + private options: OllamaSummarizerOptions; + + constructor(options: OllamaSummarizerOptions) { + this.options = options; + } + + /** + * Generate a summary for extracted text content. + */ + async summarize(attachmentId: string, text: string, contentType: string): Promise { + const startTime = Date.now(); + const truncatedText = text.substring(0, 15000); // Limit context window + + logger.info({ attachmentId, contentType, textLength: text.length }, 'Generating summary'); + + const prompt = this.buildPrompt(truncatedText, contentType); + + try { + const summary = await this.callOllama(prompt); + + const processingTimeMs = Date.now() - startTime; + logger.info({ attachmentId, summaryLength: summary.length, processingTimeMs }, 'Summary generated'); + + return { + attachmentId, + summary, + model: this.options.model, + processingTimeMs, + }; + } catch (error) { + logger.error({ err: error, attachmentId }, 'Ollama summarization failed'); + throw error; + } + } + + private buildPrompt(text: string, contentType: string): string { + return [ + 'You are a precise document summarizer. Generate a concise, factual summary of the following content.', + '', + `Content type: ${contentType}`, + '', + 'Rules:', + '- Keep summary under 200 words', + '- Focus on key facts, figures, and main points', + '- Be objective and factual', + '- Do not add information not present in the content', + '', + 'Content:', + '---', + text, + '---', + '', + 'Summary:', + ].join('\n'); + } + + private async callOllama(prompt: string): Promise { + const response = await fetch(`${this.options.baseUrl}/api/generate`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + model: this.options.model, + prompt, + stream: false, + options: { + temperature: 0.3, + top_p: 0.9, + num_predict: 512, + }, + }), + signal: AbortSignal.timeout(this.options.timeoutMs), + }); + + if (!response.ok) { + throw new Error(`Ollama API returned ${response.status}: ${await response.text()}`); + } + + const data = await response.json() as { response: string }; + return data.response?.trim() || 'No summary generated.'; + } +} + +export default OllamaSummarizer; diff --git a/packages/attachment-summarizer/src/services/orchestrator.ts b/packages/attachment-summarizer/src/services/orchestrator.ts new file mode 100644 index 00000000..9bf86bee --- /dev/null +++ b/packages/attachment-summarizer/src/services/orchestrator.ts @@ -0,0 +1,193 @@ +import prisma from '../lib/prisma'; +import logger from '../lib/logger'; +import { SqsConsumer } from './sqs-consumer'; +import { GcsDownloader } from './gcs-downloader'; +import { ContentExtractor } from './content-extractor'; +import { OllamaSummarizer } from './ollama-summarizer'; +import { SqsAttachmentEvent, AppConfig } from '../types'; + +/** + * Orchestrator that ties together SQS consumption, GCS download, + * content extraction, and LLM summarization. + */ +export class AttachmentSummarizerOrchestrator { + private sqsConsumer: SqsConsumer; + private gcsDownloader: GcsDownloader; + private contentExtractor: ContentExtractor; + private ollamaSummarizer: OllamaSummarizer; + private config: AppConfig; + + constructor(config: AppConfig) { + this.config = config; + + this.sqsConsumer = new SqsConsumer({ + region: config.aws.region, + queueUrl: config.aws.sqsQueueUrl, + pollIntervalMs: config.aws.sqsPollIntervalMs, + maxMessages: config.aws.sqsMaxMessages, + visibilityTimeout: config.aws.sqsVisibilityTimeout, + }); + + this.gcsDownloader = new GcsDownloader({ + serviceAccountKey: config.gcs.serviceAccountKey, + tempDir: config.processing.tempDir, + maxFileSizeBytes: config.processing.maxFileSizeBytes, + }); + + this.contentExtractor = new ContentExtractor(); + this.ollamaSummarizer = new OllamaSummarizer({ + baseUrl: config.ollama.baseUrl, + model: config.ollama.model, + timeoutMs: config.ollama.timeoutMs, + }); + } + + async start(): Promise { + logger.info('Starting Attachment Summarizer Orchestrator'); + + await this.sqsConsumer.start(async (event: SqsAttachmentEvent) => { + await this.processAttachment(event); + }); + } + + stop(): void { + this.sqsConsumer.stop(); + } + + private async processAttachment(event: SqsAttachmentEvent): Promise { + const { attachmentId, storageKey, bucketName, fileName, mimeType, fileSizeBytes } = event; + + logger.info({ attachmentId, fileName, mimeType }, 'Processing attachment'); + + // 1. Create/update database record + let localPath: string | null = null; + + try { + await prisma.attachment.upsert({ + where: { id: attachmentId }, + update: { status: 'DOWNLOADING' }, + create: { + id: attachmentId, + messageId: event.messageId, + storageKey, + bucketName, + fileName, + mimeType, + fileSizeBytes, + status: 'DOWNLOADING', + }, + }); + + await prisma.attachmentEvent.create({ + data: { + attachmentId, + type: 'SQS_RECEIVED', + payload: event as unknown as Record, + }, + }); + + // 2. Download from GCS + await prisma.attachment.update({ + where: { id: attachmentId }, + data: { status: 'DOWNLOADING' }, + }); + + localPath = await this.gcsDownloader.download(bucketName, storageKey, fileName); + + await prisma.attachmentEvent.create({ + data: { + attachmentId, + type: 'DOWNLOAD_COMPLETED', + payload: { localPath }, + }, + }); + + // 3. Extract content + await prisma.attachment.update({ + where: { id: attachmentId }, + data: { status: 'EXTRACTING' }, + }); + + const extracted = await this.contentExtractor.extract(localPath, mimeType, attachmentId); + + await prisma.attachment.update({ + where: { id: attachmentId }, + data: { + status: 'EXTRACTED', + extractedText: extracted.text, + contentType: extracted.contentType, + }, + }); + + await prisma.attachmentEvent.create({ + data: { + attachmentId, + type: 'EXTRACTION_COMPLETED', + payload: extracted.metadata as Record, + }, + }); + + // 4. Generate summary + await prisma.attachment.update({ + where: { id: attachmentId }, + data: { status: 'SUMMARIZING' }, + }); + + const summaryResult = await this.ollamaSummarizer.summarize( + attachmentId, + extracted.text, + extracted.contentType + ); + + await prisma.attachment.update({ + where: { id: attachmentId }, + data: { + status: 'COMPLETED', + summary: summaryResult.summary, + summaryModel: summaryResult.model, + processingTimeMs: summaryResult.processingTimeMs, + }, + }); + + await prisma.attachmentEvent.create({ + data: { + attachmentId, + type: 'SUMMARIZATION_COMPLETED', + payload: summaryResult as unknown as Record, + }, + }); + + logger.info({ attachmentId, durationMs: summaryResult.processingTimeMs }, 'Attachment processing complete'); + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error); + logger.error({ err: error, attachmentId }, 'Attachment processing failed'); + + try { + await prisma.attachment.update({ + where: { id: attachmentId }, + data: { + status: 'FAILED', + errorMessage, + }, + }); + + await prisma.attachmentEvent.create({ + data: { + attachmentId, + type: 'ERROR', + payload: { error: errorMessage }, + }, + }); + } catch (dbError) { + logger.error({ err: dbError }, 'Failed to update attachment error state'); + } + } finally { + // Clean up temp file + if (localPath) { + await this.gcsDownloader.cleanup(localPath); + } + } + } +} + +export default AttachmentSummarizerOrchestrator; diff --git a/packages/attachment-summarizer/src/services/sqs-consumer.ts b/packages/attachment-summarizer/src/services/sqs-consumer.ts new file mode 100644 index 00000000..87f2f810 --- /dev/null +++ b/packages/attachment-summarizer/src/services/sqs-consumer.ts @@ -0,0 +1,108 @@ +import { SQSClient, ReceiveMessageCommand, DeleteMessageCommand, Message } from '@aws-sdk/client-sqs'; +import logger from '../lib/logger'; +import { SqsAttachmentEvent } from '../types'; + +export interface SqsConsumerOptions { + region: string; + queueUrl: string; + pollIntervalMs: number; + maxMessages: number; + visibilityTimeout: number; +} + +/** + * Consumes attachment events from AWS SQS. + * Runs a long-polling loop that yields parsed events. + */ +export class SqsConsumer { + private client: SQSClient; + private options: SqsConsumerOptions; + private running = false; + + constructor(options: SqsConsumerOptions) { + this.client = new SQSClient({ region: options.region }); + this.options = options; + } + + async start(handler: (event: SqsAttachmentEvent) => Promise): Promise { + this.running = true; + logger.info({ queueUrl: this.options.queueUrl }, 'Starting SQS consumer'); + + while (this.running) { + try { + const command = new ReceiveMessageCommand({ + QueueUrl: this.options.queueUrl, + MaxNumberOfMessages: this.options.maxMessages, + VisibilityTimeout: this.options.visibilityTimeout, + WaitTimeSeconds: 20, // long polling + }); + + const response = await this.client.send(command); + + if (response.Messages && response.Messages.length > 0) { + for (const message of response.Messages) { + await this.processMessage(message, handler); + } + } + } catch (error) { + logger.error({ err: error }, 'Error polling SQS'); + // Back off on error before retry + await this.sleep(5000); + } + } + } + + stop(): void { + this.running = false; + logger.info('Stopping SQS consumer'); + } + + private async processMessage( + message: Message, + handler: (event: SqsAttachmentEvent) => Promise + ): Promise { + const receiptHandle = message.ReceiptHandle; + if (!message.Body || !receiptHandle) { + return; + } + + try { + const event: SqsAttachmentEvent = JSON.parse(message.Body); + + // Validate required fields + if (!event.attachmentId || !event.storageKey || !event.bucketName) { + logger.warn({ event }, 'Invalid SQS event - missing required fields'); + // Delete invalid message so it doesn't block the queue + await this.deleteMessage(receiptHandle); + return; + } + + await handler(event); + + // Delete message after successful processing + await this.deleteMessage(receiptHandle); + } catch (error) { + logger.error({ err: error, messageId: message.MessageId }, 'Failed to process SQS message'); + // Do NOT delete - message will become visible again after visibility timeout + // This acts as a retry mechanism + } + } + + private async deleteMessage(receiptHandle: string): Promise { + try { + const command = new DeleteMessageCommand({ + QueueUrl: this.options.queueUrl, + ReceiptHandle: receiptHandle, + }); + await this.client.send(command); + } catch (error) { + logger.error({ err: error }, 'Failed to delete SQS message'); + } + } + + private sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); + } +} + +export default SqsConsumer; diff --git a/packages/attachment-summarizer/src/types/index.ts b/packages/attachment-summarizer/src/types/index.ts new file mode 100644 index 00000000..b60a8ccb --- /dev/null +++ b/packages/attachment-summarizer/src/types/index.ts @@ -0,0 +1,80 @@ +export interface SqsAttachmentEvent { + messageId: string; + attachmentId: string; + storageKey: string; + bucketName: string; + fileName: string; + mimeType: string; + fileSizeBytes: number; +} + +export interface SummaryResult { + attachmentId: string; + summary: string; + model: string; + processingTimeMs: number; +} + +export interface ExtractedContent { + attachmentId: string; + text: string; + contentType: string; + metadata?: Record; +} + +export type SupportedMimeType = + | 'application/pdf' + | 'application/vnd.openxmlformats-officedocument.wordprocessingml.document' + | 'application/msword' + | 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' + | 'application/vnd.ms-excel' + | 'text/plain' + | 'text/html' + | 'image/jpeg' + | 'image/png' + | 'image/gif' + | 'image/webp'; + +export const SUPPORTED_MIME_TYPES: SupportedMimeType[] = [ + 'application/pdf', + 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', + 'application/msword', + 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', + 'application/vnd.ms-excel', + 'text/plain', + 'text/html', + 'image/jpeg', + 'image/png', + 'image/gif', + 'image/webp', +]; + +export interface AppConfig { + aws: { + region: string; + sqsQueueUrl: string; + sqsPollIntervalMs: number; + sqsMaxMessages: number; + sqsVisibilityTimeout: number; + }; + gcs: { + serviceAccountKey: string; + }; + ollama: { + baseUrl: string; + model: string; + timeoutMs: number; + }; + redis: { + host: string; + port: number; + }; + database: { + url: string; + }; + processing: { + maxFileSizeBytes: number; + tempDir: string; + maxConcurrentJobs: number; + }; +} diff --git a/packages/attachment-summarizer/tsconfig.json b/packages/attachment-summarizer/tsconfig.json new file mode 100644 index 00000000..aa2cac8d --- /dev/null +++ b/packages/attachment-summarizer/tsconfig.json @@ -0,0 +1,19 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "commonjs", + "lib": ["ES2022"], + "outDir": "./dist", + "rootDir": "./src", + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true, + "forceConsistentCasingInFileNames": true, + "resolveJsonModule": true, + "declaration": true, + "declarationMap": true, + "sourceMap": true + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "dist", "__tests__"] +}