majiayu000 Integration Patterns Skill
Overview
Design and implement robust integration patterns for connecting services, databases, and external systems. This skill encompasses RESTful and GraphQL API design, database connection management, event-driven architecture, webhook processing, retry strategies, and resilient service communication patterns.
Core Competencies
REST API Design
Design Resource-Oriented APIs:
Structure REST APIs following best practices with proper HTTP semantics:
// src/api/users/users.controller.ts
import { Router, Request, Response, NextFunction } from 'express';
import { UserService } from './user.service';
import { validateRequest } from '../../middleware/validation';
import { authenticate } from '../../middleware/auth';
import { rateLimit } from '../../middleware/rate-limit';
import { CreateUserDto, UpdateUserDto, UserQueryDto } from './user.dto';
export class UsersController {
public router = Router();
constructor(private userService: UserService) {
this.setupRoutes();
}
private setupRoutes(): void {
// List users with pagination and filtering
this.router.get(
'/',
authenticate,
rateLimit({ windowMs: 60000, max: 100 }),
validateRequest(UserQueryDto, 'query'),
this.listUsers.bind(this)
);
// Get single user
this.router.get(
'/:id',
authenticate,
this.getUser.bind(this)
);
// Create user
this.router.post(
'/',
rateLimit({ windowMs: 60000, max: 10 }),
validateRequest(CreateUserDto, 'body'),
this.createUser.bind(this)
);
// Update user (full update)
this.router.put(
'/:id',
authenticate,
validateRequest(UpdateUserDto, 'body'),
this.updateUser.bind(this)
);
// Partial update
this.router.patch(
'/:id',
authenticate,
validateRequest(UpdateUserDto, 'body', { partial: true }),
this.patchUser.bind(this)
);
// Delete user
this.router.delete(
'/:id',
authenticate,
this.deleteUser.bind(this)
);
}
private async listUsers(req: Request, res: Response, next: NextFunction): Promise<void> {
try {
const { page = 1, limit = 20, sort = 'createdAt', order = 'desc', search } = req.query;
const result = await this.userService.findAll({
page: Number(page),
limit: Number(limit),
sort: sort as string,
order: order as 'asc' | 'desc',
search: search as string,
});
res.status(200).json({
data: result.users,
pagination: {
page: result.page,
limit: result.limit,
total: result.total,
pages: Math.ceil(result.total / result.limit),
},
links: {
self: `/api/users?page=${result.page}&limit=${result.limit}`,
next: result.page < Math.ceil(result.total / result.limit)
? `/api/users?page=${result.page + 1}&limit=${result.limit}`
: null,
prev: result.page > 1
? `/api/users?page=${result.page - 1}&limit=${result.limit}`
: null,
},
});
} catch (error) {
next(error);
}
}
private async getUser(req: Request, res: Response, next: NextFunction): Promise<void> {
try {
const user = await this.userService.findById(req.params.id);
if (!user) {
res.status(404).json({
error: 'Not Found',
message: 'User not found',
statusCode: 404,
});
return;
}
res.status(200).json({
data: user,
links: {
self: `/api/users/${user.id}`,
posts: `/api/users/${user.id}/posts`,
comments: `/api/users/${user.id}/comments`,
},
});
} catch (error) {
next(error);
}
}
private async createUser(req: Request, res: Response, next: NextFunction): Promise<void> {
try {
const user = await this.userService.create(req.body);
res.status(201)
.location(`/api/users/${user.id}`)
.json({
data: user,
links: {
self: `/api/users/${user.id}`,
},
});
} catch (error) {
next(error);
}
}
private async updateUser(req: Request, res: Response, next: NextFunction): Promise<void> {
try {
const user = await this.userService.update(req.params.id, req.body);
if (!user) {
res.status(404).json({
error: 'Not Found',
message: 'User not found',
statusCode: 404,
});
return;
}
res.status(200).json({
data: user,
});
} catch (error) {
next(error);
}
}
private async patchUser(req: Request, res: Response, next: NextFunction): Promise<void> {
try {
const user = await this.userService.patch(req.params.id, req.body);
if (!user) {
res.status(404).json({
error: 'Not Found',
message: 'User not found',
statusCode: 404,
});
return;
}
res.status(200).json({
data: user,
});
} catch (error) {
next(error);
}
}
private async deleteUser(req: Request, res: Response, next: NextFunction): Promise<void> {
try {
const deleted = await this.userService.delete(req.params.id);
if (!deleted) {
res.status(404).json({
error: 'Not Found',
message: 'User not found',
statusCode: 404,
});
return;
}
res.status(204).send();
} catch (error) {
next(error);
}
}
}
Implement Consistent Error Handling:
Create standardized error responses across all endpoints:
// src/middleware/error-handler.ts
import { Request, Response, NextFunction } from 'express';
export class AppError extends Error {
constructor(
public statusCode: number,
public message: string,
public isOperational = true,
public errors?: Record<string, string[]>
) {
super(message);
Object.setPrototypeOf(this, AppError.prototype);
}
}
export function errorHandler(
err: Error | AppError,
req: Request,
res: Response,
next: NextFunction
): void {
if (err instanceof AppError) {
res.status(err.statusCode).json({
error: err.message,
statusCode: err.statusCode,
errors: err.errors,
path: req.path,
method: req.method,
timestamp: new Date().toISOString(),
});
return;
}
// Unhandled errors
console.error('Unhandled error:', err);
res.status(500).json({
error: 'Internal Server Error',
statusCode: 500,
message: process.env.NODE_ENV === 'development' ? err.message : 'An unexpected error occurred',
path: req.path,
method: req.method,
timestamp: new Date().toISOString(),
});
}
GraphQL API Design
Design Type-Safe GraphQL Schema:
Implement GraphQL APIs with proper type definitions and resolvers:
// src/graphql/schema.ts
import { gql } from 'apollo-server-express';
export const typeDefs = gql`
type User {
id: ID!
name: String!
email: String!
posts: [Post!]!
createdAt: DateTime!
updatedAt: DateTime!
}
type Post {
id: ID!
title: String!
content: String!
author: User!
comments: [Comment!]!
published: Boolean!
createdAt: DateTime!
updatedAt: DateTime!
}
type Comment {
id: ID!
content: String!
author: User!
post: Post!
createdAt: DateTime!
}
type Query {
user(id: ID!): User
users(
page: Int = 1
limit: Int = 20
search: String
): UserConnection!
post(id: ID!): Post
posts(
page: Int = 1
limit: Int = 20
published: Boolean
): PostConnection!
}
type Mutation {
createUser(input: CreateUserInput!): User!
updateUser(id: ID!, input: UpdateUserInput!): User!
deleteUser(id: ID!): Boolean!
createPost(input: CreatePostInput!): Post!
publishPost(id: ID!): Post!
createComment(input: CreateCommentInput!): Comment!
}
type Subscription {
postCreated: Post!
commentAdded(postId: ID!): Comment!
}
input CreateUserInput {
name: String!
email: String!
password: String!
}
input UpdateUserInput {
name: String
email: String
}
input CreatePostInput {
title: String!
content: String!
published: Boolean = false
}
input CreateCommentInput {
postId: ID!
content: String!
}
type UserConnection {
edges: [UserEdge!]!
pageInfo: PageInfo!
totalCount: Int!
}
type UserEdge {
node: User!
cursor: String!
}
type PostConnection {
edges: [PostEdge!]!
pageInfo: PageInfo!
totalCount: Int!
}
type PostEdge {
node: Post!
cursor: String!
}
type PageInfo {
hasNextPage: Boolean!
hasPreviousPage: Boolean!
startCursor: String
endCursor: String
}
scalar DateTime
`;
// src/graphql/resolvers.ts
import { UserService } from '../services/user.service';
import { PostService } from '../services/post.service';
import { CommentService } from '../services/comment.service';
import { PubSub } from 'graphql-subscriptions';
const pubsub = new PubSub();
export const resolvers = {
Query: {
user: async (_parent: any, { id }: { id: string }, { services }: any) => {
return services.user.findById(id);
},
users: async (_parent: any, args: any, { services }: any) => {
const result = await services.user.findAll(args);
return {
edges: result.users.map((user: any) => ({
node: user,
cursor: Buffer.from(user.id).toString('base64'),
})),
pageInfo: {
hasNextPage: result.page * result.limit < result.total,
hasPreviousPage: result.page > 1,
startCursor: result.users.length > 0
? Buffer.from(result.users[0].id).toString('base64')
: null,
endCursor: result.users.length > 0
? Buffer.from(result.users[result.users.length - 1].id).toString('base64')
: null,
},
totalCount: result.total,
};
},
post: async (_parent: any, { id }: { id: string }, { services }: any) => {
return services.post.findById(id);
},
},
Mutation: {
createUser: async (_parent: any, { input }: any, { services }: any) => {
return services.user.create(input);
},
createPost: async (_parent: any, { input }: any, { services, user }: any) => {
if (!user) {
throw new Error('Authentication required');
}
const post = await services.post.create({
...input,
authorId: user.id,
});
pubsub.publish('POST_CREATED', { postCreated: post });
return post;
},
createComment: async (_parent: any, { input }: any, { services, user }: any) => {
if (!user) {
throw new Error('Authentication required');
}
const comment = await services.comment.create({
...input,
authorId: user.id,
});
pubsub.publish(`COMMENT_ADDED_${input.postId}`, { commentAdded: comment });
return comment;
},
},
Subscription: {
postCreated: {
subscribe: () => pubsub.asyncIterator(['POST_CREATED']),
},
commentAdded: {
subscribe: (_parent: any, { postId }: { postId: string }) =>
pubsub.asyncIterator([`COMMENT_ADDED_${postId}`]),
},
},
User: {
posts: async (user: any, _args: any, { services }: any) => {
return services.post.findByAuthorId(user.id);
},
},
Post: {
author: async (post: any, _args: any, { services }: any) => {
return services.user.findById(post.authorId);
},
comments: async (post: any, _args: any, { services }: any) => {
return services.comment.findByPostId(post.id);
},
},
Comment: {
author: async (comment: any, _args: any, { services }: any) => {
return services.user.findById(comment.authorId);
},
post: async (comment: any, _args: any, { services }: any) => {
return services.post.findById(comment.postId);
},
},
};
Database Integration
Implement Connection Pooling:
Configure efficient database connection management:
// src/database/connection.ts
import { Pool, PoolConfig } from 'pg';
import { createPool as createMySQLPool, PoolOptions } from 'mysql2/promise';
export class DatabaseConnection {
private static pgPool: Pool;
private static mysqlPool: any;
static initializePostgres(config: PoolConfig): Pool {
this.pgPool = new Pool({
host: config.host || process.env.DB_HOST,
port: config.port || Number(process.env.DB_PORT) || 5432,
database: config.database || process.env.DB_NAME,
user: config.user || process.env.DB_USER,
password: config.password || process.env.DB_PASSWORD,
max: 20, // Maximum pool size
min: 5, // Minimum pool size
idleTimeoutMillis: 30000, // Close idle clients after 30s
connectionTimeoutMillis: 2000, // Timeout connection attempts after 2s
maxUses: 7500, // Close connections after 7500 uses
});
// Handle pool errors
this.pgPool.on('error', (err) => {
console.error('Unexpected database error:', err);
});
return this.pgPool;
}
static initializeMySQL(config: PoolOptions): any {
this.mysqlPool = createMySQLPool({
host: config.host || process.env.DB_HOST,
port: config.port || Number(process.env.DB_PORT) || 3306,
database: config.database || process.env.DB_NAME,
user: config.user || process.env.DB_USER,
password: config.password || process.env.DB_PASSWORD,
connectionLimit: 20,
queueLimit: 0,
waitForConnections: true,
enableKeepAlive: true,
keepAliveInitialDelay: 0,
});
return this.mysqlPool;
}
static getPostgresPool(): Pool {
if (!this.pgPool) {
throw new Error('PostgreSQL pool not initialized');
}
return this.pgPool;
}
static async closeAll(): Promise<void> {
if (this.pgPool) {
await this.pgPool.end();
}
if (this.mysqlPool) {
await this.mysqlPool.end();
}
}
}
// src/repositories/user.repository.ts
import { Pool } from 'pg';
import { DatabaseConnection } from '../database/connection';
export class UserRepository {
private pool: Pool;
constructor() {
this.pool = DatabaseConnection.getPostgresPool();
}
async findById(id: string): Promise<any> {
const client = await this.pool.connect();
try {
const result = await client.query(
'SELECT * FROM users WHERE id = $1',
[id]
);
return result.rows[0];
} finally {
client.release();
}
}
async findAll(options: any): Promise<any> {
const client = await this.pool.connect();
try {
const offset = (options.page - 1) * options.limit;
let query = 'SELECT * FROM users';
let countQuery = 'SELECT COUNT(*) FROM users';
const params: any[] = [];
if (options.search) {
const searchCondition = ' WHERE name ILIKE $1 OR email ILIKE $1';
query += searchCondition;
countQuery += searchCondition;
params.push(`%${options.search}%`);
}
query += ` ORDER BY ${options.sort} ${options.order}`;
query += ` LIMIT $${params.length + 1} OFFSET $${params.length + 2}`;
params.push(options.limit, offset);
const [users, count] = await Promise.all([
client.query(query, params),
client.query(countQuery, params.length > 0 ? [params[0]] : []),
]);
return {
users: users.rows,
total: Number(count.rows[0].count),
page: options.page,
limit: options.limit,
};
} finally {
client.release();
}
}
async create(user: any): Promise<any> {
const client = await this.pool.connect();
try {
await client.query('BEGIN');
const result = await client.query(
'INSERT INTO users (name, email, password_hash) VALUES ($1, $2, $3) RETURNING *',
[user.name, user.email, user.passwordHash]
);
await client.query('COMMIT');
return result.rows[0];
} catch (error) {
await client.query('ROLLBACK');
throw error;
} finally {
client.release();
}
}
}
Implement Query Optimization:
Use prepared statements and indexes for performance:
// src/repositories/optimized-queries.ts
export class OptimizedQueryRepository {
async batchFindByIds(ids: string[]): Promise<any[]> {
// Use IN clause for batch retrieval
const client = await this.pool.connect();
try {
const result = await client.query(
'SELECT * FROM users WHERE id = ANY($1::uuid[])',
[ids]
);
return result.rows;
} finally {
client.release();
}
}
async findWithJoins(userId: string): Promise<any> {
// Optimize joins with indexes
const client = await this.pool.connect();
try {
const result = await client.query(`
SELECT
u.*,
json_agg(
json_build_object(
'id', p.id,
'title', p.title,
'createdAt', p.created_at
)
) FILTER (WHERE p.id IS NOT NULL) as posts
FROM users u
LEFT JOIN posts p ON p.author_id = u.id
WHERE u.id = $1
GROUP BY u.id
`, [userId]);
return result.rows[0];
} finally {
client.release();
}
}
}
// Create indexes for performance
// migrations/001_add_indexes.sql
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_posts_author_id ON posts(author_id);
CREATE INDEX idx_posts_created_at ON posts(created_at DESC);
CREATE INDEX idx_comments_post_id ON comments(post_id);
Event-Driven Architecture
Implement Message Queue Integration:
Use message queues for asynchronous processing:
// src/queues/job.queue.ts
import Bull, { Queue, Job } from 'bull';
import Redis from 'ioredis';
export interface JobData {
type: string;
payload: any;
}
export class JobQueue {
private queue: Queue<JobData>;
private redis: Redis;
constructor(queueName: string) {
this.redis = new Redis({
host: process.env.REDIS_HOST || 'localhost',
port: Number(process.env.REDIS_PORT) || 6379,
maxRetriesPerRequest: null,
enableReadyCheck: false,
});
this.queue = new Bull<JobData>(queueName, {
redis: this.redis,
defaultJobOptions: {
attempts: 3,
backoff: {
type: 'exponential',
delay: 2000,
},
removeOnComplete: 100,
removeOnFail: 500,
},
});
this.setupEventHandlers();
}
private setupEventHandlers(): void {
this.queue.on('error', (error) => {
console.error('Queue error:', error);
});
this.queue.on('failed', (job, error) => {
console.error(`Job ${job.id} failed:`, error);
});
this.queue.on('completed', (job) => {
console.log(`Job ${job.id} completed`);
});
}
async addJob(data: JobData, options?: Bull.JobOptions): Promise<Job<JobData>> {
return this.queue.add(data, options);
}
async process(concurrency: number, processor: (job: Job<JobData>) => Promise<any>): Promise<void> {
this.queue.process(concurrency, processor);
}
async close(): Promise<void> {
await this.queue.close();
this.redis.disconnect();
}
}
// src/workers/email.worker.ts
import { JobQueue } from '../queues/job.queue';
import { EmailService } from '../services/email.service';
export class EmailWorker {
private queue: JobQueue;
private emailService: EmailService;
constructor() {
this.queue = new JobQueue('email');
this.emailService = new EmailService();
this.start();
}
private start(): void {
this.queue.process(5, async (job) => {
const { type, payload } = job.data;
switch (type) {
case 'welcome':
await this.emailService.sendWelcomeEmail(payload);
break;
case 'password-reset':
await this.emailService.sendPasswordResetEmail(payload);
break;
case 'notification':
await this.emailService.sendNotification(payload);
break;
default:
throw new Error(`Unknown email type: ${type}`);
}
});
}
}
Implement Webhook Processing:
Handle incoming webhooks with retry logic:
// src/webhooks/webhook.handler.ts
import { Request, Response } from 'express';
import crypto from 'crypto';
export class WebhookHandler {
async handleStripeWebhook(req: Request, res: Response): Promise<void> {
const signature = req.headers['stripe-signature'] as string;
try {
// Verify webhook signature
this.verifyStripeSignature(req.rawBody, signature);
const event = req.body;
// Process event asynchronously
await this.processStripeEvent(event);
res.status(200).json({ received: true });
} catch (error) {
console.error('Webhook error:', error);
res.status(400).json({ error: 'Webhook validation failed' });
}
}
private verifyStripeSignature(payload: string, signature: string): void {
const secret = process.env.STRIPE_WEBHOOK_SECRET!;
const expectedSignature = crypto
.createHmac('sha256', secret)
.update(payload)
.digest('hex');
if (signature !== expectedSignature) {
throw new Error('Invalid signature');
}
}
private async processStripeEvent(event: any): Promise<void> {
switch (event.type) {
case 'payment_intent.succeeded':
await this.handlePaymentSuccess(event.data.object);
break;
case 'payment_intent.failed':
await this.handlePaymentFailure(event.data.object);
break;
case 'customer.subscription.updated':
await this.handleSubscriptionUpdate(event.data.object);
break;
default:
console.log(`Unhandled event type: ${event.type}`);
}
}
}
Retry Strategies
Implement Exponential Backoff:
Handle transient failures with intelligent retry logic:
// src/utils/retry.ts
export interface RetryOptions {
maxAttempts: number;
initialDelay: number;
maxDelay: number;
backoffMultiplier: number;
retryableErrors?: string[];
}
export async function retryWithBackoff<T>(
operation: () => Promise<T>,
options: RetryOptions
): Promise<T> {
let lastError: Error;
let delay = options.initialDelay;
for (let attempt = 1; attempt <= options.maxAttempts; attempt++) {
try {
return await operation();
} catch (error: any) {
lastError = error;
// Check if error is retryable
if (options.retryableErrors &&
!options.retryableErrors.includes(error.code)) {
throw error;
}
// Don't retry on last attempt
if (attempt === options.maxAttempts) {
break;
}
console.log(`Attempt ${attempt} failed, retrying in ${delay}ms...`);
await new Promise(resolve => setTimeout(resolve, delay));
// Calculate next delay with exponential backoff
delay = Math.min(
delay * options.backoffMultiplier,
options.maxDelay
);
}
}
throw lastError!;
}
// Usage example
async function fetchUserWithRetry(userId: string): Promise<User> {
return retryWithBackoff(
() => userService.findById(userId),
{
maxAttempts: 3,
initialDelay: 1000,
maxDelay: 10000,
backoffMultiplier: 2,
retryableErrors: ['ECONNRESET', 'ETIMEDOUT', 'ENOTFOUND'],
}
);
}
Related Resources
- DevOps Practices Skill - For deploying and managing integrated services
- Performance Optimization Skill - For monitoring API and database performance
- Documentation Patterns Skill - For API documentation with OpenAPI/Swagger
- REST API Best Practices - https://restfulapi.net
- GraphQL Documentation - https://graphql.org
- Database Design Patterns - https://www.postgresql.org/docs/