@message-in-the-middle/store-memory

⚠️ Work in Progress Is this library production-ready? No. Is this library safe? No. When will it be ready? Soon™ (maybe tomorrow, maybe never). Why is it public? Experiment

message-in-the-middle is to Express.js what your message queue processing is to HTTP request processing. Just as Express provides a middleware pattern for HTTP requests, this library provides a middleware pattern for processing queue messages.

Why This Exists

Processing queue messages usually means copy-pasting the same boilerplate: parse JSON, validate, log, retry, deduplicate, route to handlers. This library lets you compose that logic as middlewares.

In-memory message store implementation for message-middleware. Fast, zero-dependency storage for development and testing.

⚠️ For Development and Testing Only - Data is not persisted and will be lost on restart. Use @message-in-the-middle/store-mysql for production.

Features

• ⚡ Fast - In-memory storage with O(1) lookups
• 🚫 Zero Dependencies - No external dependencies
• 🎯 Bounded Size - Configurable max size with automatic eviction
• 🔄 Priority Eviction - Keeps important messages (FAILED > SUCCEEDED > ARCHIVED)
• 🎨 Full TypeScript Support - Complete type safety
• 💾 MessageStore Interface - Compatible with all persistence middleware

Installation

# npm
npm install @message-in-the-middle/store-memory @message-in-the-middle/persistence-core

# pnpm
pnpm add @message-in-the-middle/store-memory @message-in-the-middle/persistence-core

# yarn
yarn add @message-in-the-middle/store-memory @message-in-the-middle/persistence-core

Quick Start

import { MessageMiddlewareManager } from '@message-in-the-middle/core';
import { PersistenceInboundMiddleware } from '@message-in-the-middle/persistence-core';
import { InMemoryMessageStore } from '@message-in-the-middle/store-memory';

// Create in-memory store
const store = new InMemoryMessageStore();

// Use with persistence middleware
const manager = new MessageMiddlewareManager();
manager.addInboundMiddleware(
  new PersistenceInboundMiddleware(store, {
    storeOn: ['error']
  })
);

// Process messages
await manager.processInbound(message);

// Query stored messages
const failed = await store.findByStatus(MessageStatus.FAILED);

Configuration

Default Configuration

// Default: 10,000 messages max
const store = new InMemoryMessageStore();

Custom Max Size

const store = new InMemoryMessageStore({
  maxSize: 5000  // Store up to 5,000 messages
});

No Size Limit (Not Recommended)

const store = new InMemoryMessageStore({
  maxSize: Infinity  // Unlimited (can cause memory issues)
});

Size Management

Automatic Eviction

When the store reaches maxSize, it automatically evicts old messages using a priority-based strategy:

Eviction Priority (lowest to highest):

1. ARCHIVED - Evicted first (least important)
2. SUCCEEDED - Evicted second
3. FAILED - Kept as long as possible (most important for debugging)

Why This Priority?

• FAILED messages are most valuable for debugging production issues
• SUCCEEDED messages are useful for audit but less critical
• ARCHIVED messages are historical and can be removed first

Check Current Size

const currentSize = store.size();
console.log(`Store contains ${currentSize} messages`);

Manual Cleanup

// Clear all messages
await store.clear();

// Delete specific message
await store.delete('message-id-123');

Querying Messages

Find by Status

import { MessageStatus } from '@message-in-the-middle/persistence-core';

// Find all failed messages
const failed = await store.findByStatus(MessageStatus.FAILED);

// With options
const recent = await store.findByStatus(MessageStatus.FAILED, {
  limit: 20,
  offset: 0,
  sortBy: 'created',
  sortOrder: 'desc'
});

Find by Error Type

// Find validation errors
const errors = await store.findByError('ValidationError');

// With date range
const recentErrors = await store.findByError('ValidationError', {
  startDate: new Date(Date.now() - 24 * 60 * 60 * 1000)
});

Find by ID

const message = await store.findById('msg-123');

if (message) {
  console.log('Status:', message.status);
  console.log('Error:', message.errorMessage);
  console.log('Attempts:', message.retryCount);
}

Count Messages

// Total messages
const total = await store.count();

// Count by status
const failedCount = await store.count({
  status: MessageStatus.FAILED
});

// Count by error type
const validationErrors = await store.count({
  errorType: 'ValidationError'
});

Message Replay

Use with MessageReplayManager to replay stored messages:

import { MessageReplayManager } from '@message-in-the-middle/persistence-core';

const replayManager = new MessageReplayManager(store, pipeline);

// Replay failed messages
const result = await replayManager.replayFailed({ limit: 50 });
console.log(`Replayed ${result.succeeded} successfully`);

Complete Example

import {
  MessageMiddlewareManager,
  ParseJsonInboundMiddleware,
  ValidateInboundMiddleware
} from '@message-in-the-middle/core';
import {
  PersistenceInboundMiddleware,
  MessageReplayManager,
  MessageStatus
} from '@message-in-the-middle/persistence-core';
import { InMemoryMessageStore } from '@message-in-the-middle/store-memory';

// Create store with custom size
const store = new InMemoryMessageStore({ maxSize: 1000 });

// Create manager with persistence
const manager = new MessageMiddlewareManager();
manager
  .addInboundMiddleware(new PersistenceInboundMiddleware(store, {
    storeOn: ['error']  // Store only errors
  }))
  .addInboundMiddleware(new ParseJsonInboundMiddleware())
  .addInboundMiddleware(new ValidateInboundMiddleware(validator));

// Process messages
try {
  await manager.processInbound(messageBody);
} catch (error) {
  console.error('Processing failed, message stored for replay');
}

// Check store size
console.log(`Stored messages: ${store.size()}`);

// Query failed messages
const failed = await store.findByStatus(MessageStatus.FAILED);
console.log(`Failed messages: ${failed.length}`);

// Replay failures
const replayManager = new MessageReplayManager(store, manager);
const result = await replayManager.replayFailed({ limit: 10 });
console.log(`Replay result:`, result);

// Cleanup
await store.clear();

Performance Characteristics

Time Complexity

• save() - O(1)
• findById() - O(1)
• findByStatus() - O(n) where n = number of messages with that status
• findByError() - O(n) where n = total messages
• count() - O(n)
• delete() - O(1)
• clear() - O(1)

Memory Usage

Each message consumes approximately:

• Base overhead: ~500 bytes
• Message payload: Variable (depends on message size)
• Metadata: ~200 bytes

Example:

• 10,000 messages @ 1KB each ≈ ~17 MB total
• 1,000 messages @ 10KB each ≈ ~17 MB total

API Reference

Constructor

new InMemoryMessageStore(options?: InMemoryMessageStoreOptions)

Options:

interface InMemoryMessageStoreOptions {
  maxSize?: number;  // Default: 10000
}

Methods

All methods implement the MessageStore interface from @message-in-the-middle/persistence-core:

// Storage
save(message: StoredMessage): Promise<void>
updateStatus(id: string, updates: Partial<StoredMessage>): Promise<void>
delete(id: string): Promise<void>
clear(): Promise<void>

// Queries
findById(id: string): Promise<StoredMessage | null>
findByStatus(status: MessageStatus, options?: QueryOptions): Promise<StoredMessage[]>
findByError(errorType: string, options?: QueryOptions): Promise<StoredMessage[]>
count(filters?: QueryFilters): Promise<number>

// Utility
size(): number
destroy(): Promise<void>

When to Use

✅ Good For

• Local development - Fast feedback, no setup required
• Unit tests - Isolated, repeatable tests
• Integration tests - Test persistence logic without external dependencies
• Prototyping - Quick experimentation
• Demo applications - Showcase without infrastructure

❌ Not Good For

• Production - Data lost on restart
• Multiple processes - Not shared across processes
• Large datasets - Limited by available RAM
• Long-term storage - No persistence between restarts
• Distributed systems - Can't share state across nodes

Migration to Production Store

When moving to production, replace with a persistent store:

Before (Development)

import { InMemoryMessageStore } from '@message-in-the-middle/store-memory';

const store = new InMemoryMessageStore();

After (Production)

import { MySQLMessageStore } from '@message-in-the-middle/store-mysql';
import { createPool } from 'mysql2/promise';

const pool = createPool({
  host: 'localhost',
  user: 'user',
  password: 'password',
  database: 'messages'
});

const store = new MySQLMessageStore(pool);

Everything else stays the same! The MessageStore interface ensures compatibility.

Testing Example

import { describe, it, expect, beforeEach } from 'vitest';
import { InMemoryMessageStore } from '@message-in-the-middle/store-memory';
import { MessageStatus } from '@message-in-the-middle/persistence-core';

describe('Message Processing', () => {
  let store: InMemoryMessageStore;

  beforeEach(() => {
    store = new InMemoryMessageStore();
  });

  it('should store failed messages', async () => {
    await store.save({
      id: 'msg-1',
      status: MessageStatus.FAILED,
      message: { order: 123 },
      errorMessage: 'Validation failed',
      retryCount: 0,
      created: new Date(),
      updated: new Date()
    });

    const failed = await store.findByStatus(MessageStatus.FAILED);
    expect(failed).toHaveLength(1);
    expect(failed[0].id).toBe('msg-1');
  });
});

Related Packages

• @message-in-the-middle/persistence-core - Persistence interfaces (required)
• @message-in-the-middle/store-mysql - MySQL store for production
• @message-in-the-middle/core - Core library

Documentation

• Persistence Core - Complete persistence documentation
• Main README - Library documentation
• Architecture - Design patterns
• Contributing - How to contribute

License

MIT

Links

• GitHub Repository
• Report Issues