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__"]
+}