warpspeedopen-source · lb1192176991-lab · Jun 27, 2026
diff --git a/packages/attachment-summarizer/.env.example 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
@@ -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
@@ -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('<html><body><h1>Title</h1><p>Content</p></body></html>');
+    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('<html>');
+  });
+});
diff --git a/packages/attachment-summarizer/__tests__/sqs-consumer.test.ts 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
@@ -0,0 +1,21 @@
+/** @type {import('ts-jest').JestConfigWithTsJest} */
+module.exports = {
+  preset: 'ts-jest',
+  testEnvironment: 'node',
+  roots: ['<rootDir>/__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
@@ -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
@@ -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
@@ -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<void> {
+  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);
+});