Integration Patterns

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/