Idempotency Store
```typescript
// idempotency-store.ts
import { Redis } from 'ioredis';
interface IdempotencyRecord {
status: 'processing' | 'completed';
response?: {
statusCode: number;
body: unknown;
headers?: Record;
};
createdAt: number;
completedAt?: number;
}
interface IdempotencyConfig {
redis: Redis;
keyPrefix?: string;
lockTtlMs?: number; // How long to hold processing lock
responseTtlMs?: number; // How long to cache completed responses
}
class IdempotencyStore {
private redis: Redis;
private keyPrefix: string;
private lockTtl: number;
private responseTtl: number;
constructor(config: IdempotencyConfig) {
this.redis = config.redis;
this.keyPrefix = config.keyPrefix || 'idempotency:';
this.lockTtl = config.lockTtlMs || 60000; // 1 minute
this.responseTtl = config.responseTtlMs || 86400000; // 24 hours
}
async get(key: string): Promise {
const data = await this.redis.get(this.keyPrefix + key);
return data ? JSON.parse(data) : null;
}
async acquireLock(key: string): Promise {
const record: IdempotencyRecord = {
status: 'processing',
createdAt: Date.now(),
};
// SET NX = only set if not exists
const result = await this.redis.set(
this.keyPrefix + key,
JSON.stringify(record),
'PX',
this.lockTtl,
'NX'
);
return result === 'OK';
}
async complete(
key: string,
response: IdempotencyRecord['response']
): Promise {
const record: IdempotencyRecord = {
status: 'completed',
response,
createdAt: Date.now(),
completedAt: Date.now(),
};
await this.redis.set(
this.keyPrefix + key,
JSON.stringify(record),
'PX',
this.responseTtl
);
}
async release(key: string): Promise {
await this.redis.del(this.keyPrefix + key);
}
}
export { IdempotencyStore, IdempotencyRecord, IdempotencyConfig };
```
Express Middleware
```typescript
// idempotency-middleware.ts
import { Request, Response, NextFunction } from 'express';
import { IdempotencyStore } from './idempotency-store';
interface IdempotencyOptions {
store: IdempotencyStore;
headerName?: string;
methods?: string[];
paths?: RegExp[];
}
function idempotencyMiddleware(options: IdempotencyOptions) {
const {
store,
headerName = 'Idempotency-Key',
methods = ['POST', 'PUT', 'PATCH'],
paths = [/.*/],
} = options;
return async (req: Request, res: Response, next: NextFunction) => {
// Only apply to specified methods
if (!methods.includes(req.method)) {
return next();
}
// Only apply to specified paths
if (!paths.some(p => p.test(req.path))) {
return next();
}
const idempotencyKey = req.headers[headerName.toLowerCase()] as string;
// No key provided - proceed without idempotency
if (!idempotencyKey) {
return next();
}
// Create a unique key combining the idempotency key with request details
const fullKey = ${req.method}:${req.path}:${idempotencyKey};
// Check for existing record
const existing = await store.get(fullKey);
if (existing) {
if (existing.status === 'processing') {
// Request is still being processed
return res.status(409).json({
error: 'Conflict',
message: 'A request with this idempotency key is already being processed',
});
}
if (existing.status === 'completed' && existing.response) {
// Return cached response
res.status(existing.response.statusCode);
if (existing.response.headers) {
for (const [key, value] of Object.entries(existing.response.headers)) {
res.setHeader(key, value);
}
}
res.setHeader('X-Idempotent-Replayed', 'true');
return res.json(existing.response.body);
}
}
// Try to acquire lock
const acquired = await store.acquireLock(fullKey);
if (!acquired) {
// Another request just acquired the lock
return res.status(409).json({
error: 'Conflict',
message: 'A request with this idempotency key is already being processed',
});
}
// Capture the response
const originalJson = res.json.bind(res);
let responseBody: unknown;
res.json = (body: unknown) => {
responseBody = body;
return originalJson(body);
};
// Store response after it's sent
res.on('finish', async () => {
if (res.statusCode >= 200 && res.statusCode < 500) {
// Store successful responses and client errors (but not server errors)
await store.complete(fullKey, {
statusCode: res.statusCode,
body: responseBody,
});
} else {
// Release lock for server errors (allow retry)
await store.release(fullKey);
}
});
next();
};
}
export { idempotencyMiddleware, IdempotencyOptions };
```
Usage
```typescript
// app.ts
import express from 'express';
import { Redis } from 'ioredis';
import { IdempotencyStore } from './idempotency-store';
import { idempotencyMiddleware } from './idempotency-middleware';
const app = express();
const redis = new Redis();
const idempotencyStore = new IdempotencyStore({ redis });
// Apply to all POST/PUT/PATCH requests
app.use(idempotencyMiddleware({
store: idempotencyStore,
methods: ['POST', 'PUT', 'PATCH'],
}));
// Or apply to specific routes
app.post('/orders',
idempotencyMiddleware({
store: idempotencyStore,
paths: [/^\/orders$/],
}),
async (req, res) => {
const order = await createOrder(req.body);
res.status(201).json(order);
}
);
```
|