Database Patterns in Microservices: A Complete Guide to Data Architecture#

1
// Order Service - Creating an order
2
async function createOrder(orderData) {
3
  const transaction = await db.beginTransaction();
4
  try {
5
    // Create order
6
    const order = await orderRepository.create(orderData, transaction);
7

8
    // Publish event for other services
9
    await eventBus.publish("OrderCreated", {
10
      orderId: order.id,
11
      userId: order.userId,
12
      items: order.items,
13
      timestamp: new Date(),
14
    });
15

16
    await transaction.commit();
17
    return order;
18
  } catch (error) {
19
    await transaction.rollback();
20
    throw error;
21
  }
22
}

1
// Order Service - Getting order details with user info
2
async function getOrderDetails(orderId) {
3
  // Get order from local database
4
  const order = await orderRepository.findById(orderId);
5

6
  // Get user details from User Service
7
  const user = await userServiceClient.getUser(order.userId);
8

9
  // Get product details from Inventory Service
10
  const products = await Promise.all(
11
    order.items.map(item => inventoryServiceClient.getProduct(item.productId))
12
  );
13

14
  return {
15
    ...order,
16
    user,
17
    products,
18
  };
19
}

1
// Event-driven consistency example
2
class OrderService {
3
  async createOrder(orderData) {
4
    // 1. Create order in local database
5
    const order = await this.orderRepository.create(orderData);
6

7
    // 2. Publish event for inventory update
8
    await this.eventBus.publish("OrderCreated", {
9
      orderId: order.id,
10
      items: order.items,
11
      timestamp: new Date(),
12
    });
13

14
    return order;
15
  }
16
}
17

18
class InventoryService {
19
  async handleOrderCreated(event) {
20
    try {
21
      // Update inventory based on order
22
      for (const item of event.items) {
23
        await this.inventoryRepository.reduceStock(
24
          item.productId,
25
          item.quantity
26
        );
27
      }
28

29
      // Publish confirmation event
30
      await this.eventBus.publish("InventoryUpdated", {
31
        orderId: event.orderId,
32
        status: "success",
33
        timestamp: new Date(),
34
      });
35
    } catch (error) {
36
      // Publish failure event for compensation
37
      await this.eventBus.publish("InventoryUpdateFailed", {
38
        orderId: event.orderId,
39
        error: error.message,
40
        timestamp: new Date(),
41
      });
42
    }
43
  }
44
}

1
// Saga orchestrator for order processing
2
class OrderSaga {
3
  async processOrder(orderData) {
4
    const sagaId = generateSagaId();
5

6
    try {
7
      // Step 1: Validate payment
8
      const paymentResult = await this.paymentService.validatePayment(
9
        orderData.paymentInfo
10
      );
11

12
      if (!paymentResult.valid) {
13
        throw new Error("Payment validation failed");
14
      }
15

16
      // Step 2: Reserve inventory
17
      const inventoryResult = await this.inventoryService.reserveItems(
18
        orderData.items
19
      );
20

21
      if (!inventoryResult.success) {
22
        // Compensate: Release payment hold
23
        await this.paymentService.releasePaymentHold(paymentResult.holdId);
24
        throw new Error("Inventory reservation failed");
25
      }
26

27
      // Step 3: Create order
28
      const order = await this.orderService.createOrder(orderData);
29

30
      // Step 4: Process payment
31
      const finalPayment = await this.paymentService.processPayment(
32
        paymentResult.holdId
33
      );
34

35
      // Step 5: Confirm inventory
36
      await this.inventoryService.confirmReservation(
37
        inventoryResult.reservationId
38
      );
39

40
      return order;
41
    } catch (error) {
42
      // Execute compensation transactions
43
      await this.compensate(sagaId, error);
44
      throw error;
45
    }
46
  }
47

48
  async compensate(sagaId, error) {
49
    // Implement compensation logic
50
    console.log(`Compensating saga ${sagaId}: ${error.message}`);
51
  }
52
}

1
// Database synchronization during migration
2
class DatabaseMigrator {
3
  async synchronizeData(sourceTable, targetService) {
4
    // 1. Initial bulk copy
5
    await this.bulkCopyData(sourceTable, targetService);
6

7
    // 2. Set up real-time synchronization
8
    await this.setupChangeDataCapture(sourceTable, targetService);
9

10
    // 3. Implement dual-write pattern
11
    await this.enableDualWrite(sourceTable, targetService);
12
  }
13

14
  async bulkCopyData(sourceTable, targetService) {
15
    const batchSize = 1000;
16
    let offset = 0;
17

18
    while (true) {
19
      const batch = await this.sourceDb.query(`
20
                SELECT * FROM ${sourceTable}
21
                ORDER BY id
22
                LIMIT ${batchSize} OFFSET ${offset}
23
            `);
24

25
      if (batch.length === 0) break;
26

27
      // Transform and insert into new service
28
      await targetService.bulkInsert(batch);
29
      offset += batchSize;
30

31
      // Progress tracking
32
      console.log(`Migrated ${offset} records from ${sourceTable}`);
33
    }
34
  }
35

36
  async setupChangeDataCapture(sourceTable, targetService) {
37
    // PostgreSQL example using logical replication
38
    await this.sourceDb.query(`
39
            CREATE PUBLICATION ${sourceTable}_pub
40
            FOR TABLE ${sourceTable}
41
        `);
42

43
    // Set up subscriber in target service
44
    await targetService.setupReplicationSubscriber(sourceTable);
45
  }
46
}

1
// Zero-downtime migration strategy
2
class ZeroDowntimeMigrator {
3
  async migrateService(serviceName, migrationPlan) {
4
    // Phase 1: Prepare new infrastructure
5
    await this.prepareNewInfrastructure(serviceName);
6

7
    // Phase 2: Start dual-write
8
    await this.startDualWrite(serviceName);
9

10
    // Phase 3: Sync historical data
11
    await this.syncHistoricalData(serviceName);
12

13
    // Phase 4: Switch read traffic gradually
14
    await this.switchReadTrafficGradually(serviceName);
15

16
    // Phase 5: Switch write traffic
17
    await this.switchWriteTraffic(serviceName);
18

19
    // Phase 6: Cleanup old infrastructure
20
    await this.cleanupOldInfrastructure(serviceName);
21
  }
22

23
  async switchReadTrafficGradually(serviceName) {
24
    const trafficPercentages = [10, 25, 50, 75, 100];
25

26
    for (const percentage of trafficPercentages) {
27
      // Update load balancer configuration
28
      await this.updateLoadBalancer(serviceName, percentage);
29

30
      // Monitor for issues
31
      await this.monitorTraffic(serviceName, percentage);
32

33
      // Wait before next increment
34
      await this.sleep(300000); // 5 minutes
35
    }
36
  }
37
}

1
-- PostgreSQL example: User management with ACID compliance
2
CREATE TABLE users (
3
    id SERIAL PRIMARY KEY,
4
    email VARCHAR(255) UNIQUE NOT NULL,
5
    encrypted_password VARCHAR(255) NOT NULL,
6
    created_at TIMESTAMP DEFAULT NOW(),
7
    updated_at TIMESTAMP DEFAULT NOW()
8
);
9

10
CREATE TABLE user_profiles (
11
    id SERIAL PRIMARY KEY,
12
    user_id INTEGER REFERENCES users(id),
13
    first_name VARCHAR(100),
14
    last_name VARCHAR(100),
15
    date_of_birth DATE,
16
    CONSTRAINT unique_user_profile UNIQUE(user_id)
17
);
18

19
-- Complex query example
20
SELECT
21
    u.email,
22
    up.first_name,
23
    up.last_name,
24
    COUNT(o.id) as order_count
25
FROM users u
26
JOIN user_profiles up ON u.id = up.user_id
27
LEFT JOIN orders o ON u.id = o.user_id
28
WHERE u.created_at >= NOW() - INTERVAL '30 days'
29
GROUP BY u.id, u.email, up.first_name, up.last_name
30
HAVING COUNT(o.id) > 5;

1
// MongoDB example: Product catalog with flexible schema
2
const productSchema = {
3
  _id: ObjectId,
4
  name: String,
5
  description: String,
6
  category: String,
7
  price: {
8
    amount: Number,
9
    currency: String,
10
  },
11
  specifications: {
12
    // Flexible schema - different products have different specs
13
    weight: String,
14
    dimensions: {
15
      length: Number,
16
      width: Number,
17
      height: Number,
18
      unit: String,
19
    },
20
    // Electronics specific
21
    batteryLife: String,
22
    warranty: String,
23
    // Clothing specific
24
    size: String,
25
    material: String,
26
    color: String,
27
  },
28
  inventory: {
29
    quantity: Number,
30
    reserved: Number,
31
    available: Number,
32
  },
33
  reviews: [
34
    {
35
      userId: ObjectId,
36
      rating: Number,
37
      comment: String,
38
      date: Date,
39
    },
40
  ],
41
  tags:
42
  - String,
43
  createdAt: Date,
44
  updatedAt: Date,
45
};
46

47
// Complex aggregation query
48
db.products.aggregate([
49
  {
50
    $match: {
51
      category: "electronics",
52
      "price.amount": { $gte: 100, $lte: 1000 },
53
    },
54
  },
55
  {
56
    $addFields: {
57
      averageRating: { $avg: "$reviews.rating" },
58
      reviewCount: { $size: "$reviews" },
59
    },
60
  },
61
  {
62
    $match: {
63
      averageRating: { $gte: 4.0 },
64
      reviewCount: { $gte: 10 },
65
    },
66
  },
67
  {
68
    $sort: { averageRating: -1, reviewCount: -1 },
69
  },
70
  {
71
    $limit: 20,
72
  },
73
]);

1
// Redis example: Session management and caching
2
class SessionService {
3
  constructor(redisClient) {
4
    this.redis = redisClient;
5
  }
6

7
  async createSession(userId, sessionData) {
8
    const sessionId = generateSessionId();
9
    const sessionKey = `session:${sessionId}`;
10

11
    await this.redis.hset(sessionKey, {
12
      userId: userId,
13
      loginTime: Date.now(),
14
      lastActivity: Date.now(),
15
      ...sessionData,
16
    });
17

18
    // Set expiration (24 hours)
19
    await this.redis.expire(sessionKey, 24 * 60 * 60);
20

21
    return sessionId;
22
  }
23

24
  async getSession(sessionId) {
25
    const sessionKey = `session:${sessionId}`;
26
    const session = await this.redis.hgetall(sessionKey);
27

28
    if (Object.keys(session).length === 0) {
29
      return null;
30
    }
31

32
    // Update last activity
33
    await this.redis.hset(sessionKey, "lastActivity", Date.now());
34

35
    return session;
36
  }
37

38
  async cacheUserData(userId, userData) {
39
    const cacheKey = `user:${userId}`;
40

41
    await this.redis.setex(
42
      cacheKey,
43
      300, // 5 minutes TTL
44
      JSON.stringify(userData)
45
    );
46
  }
47
}

1
// Elasticsearch example: Product search service
2
class ProductSearchService {
3
  constructor(elasticsearchClient) {
4
    this.es = elasticsearchClient;
5
  }
6

7
  async indexProduct(product) {
8
    await this.es.index({
9
      index: "products",
10
      id: product.id,
11
      body: {
12
        name: product.name,
13
        description: product.description,
14
        category: product.category,
15
        price: product.price.amount,
16
        currency: product.price.currency,
17
        tags: product.tags,
18
        specifications: product.specifications,
19
        averageRating: this.calculateAverageRating(product.reviews),
20
        reviewCount: product.reviews.length,
21
        availability: product.inventory.available > 0,
22
        createdAt: product.createdAt,
23
      },
24
    });
25
  }
26

27
  async searchProducts(query, filters = {}) {
28
    const searchBody = {
29
      query: {
30
        bool: {
31
          must: [
32
            {
33
              multi_match: {
34
                query: query,
35
                fields: ["name^3", "description^2", "tags"],
36
                fuzziness: "AUTO",
37
              },
38
            },
39
          ],
40
          filter: [],
41
        },
42
      },
43
      aggregations: {
44
        categories: {
45
          terms: { field: "category.keyword" },
46
        },
47
        priceRanges: {
48
          range: {
49
            field: "price",
50
            ranges: [
51
              { to: 50 },
52
              { from: 50, to: 100 },
53
              { from: 100, to: 500 },
54
              { from: 500 },
55
            ],
56
          },
57
        },
58
      },
59
      sort: [
60
        { _score: { order: "desc" } },
61
        { averageRating: { order: "desc" } },
62
      ],
63
    };
64

65
    // Apply filters
66
    if (filters.category) {
67
      searchBody.query.bool.filter.push({
68
        term: { "category.keyword": filters.category },
69
      });
70
    }
71

72
    if (filters.minPrice || filters.maxPrice) {
73
      const priceFilter = { range: { price: {} } };
74
      if (filters.minPrice) priceFilter.range.price.gte = filters.minPrice;
75
      if (filters.maxPrice) priceFilter.range.price.lte = filters.maxPrice;
76
      searchBody.query.bool.filter.push(priceFilter);
77
    }
78

79
    const result = await this.es.search({
80
      index: "products",
81
      body: searchBody,
82
    });
83

84
    return {
85
      products: result.body.hits.hits.map(hit => hit._source),
86
      total: result.body.hits.total.value,
87
      aggregations: result.body.aggregations,
88
    };
89
  }
90
}

1
// Order processing saga with orchestration
2
class OrderProcessingSaga {
3
  constructor(services) {
4
    this.paymentService = services.payment;
5
    this.inventoryService = services.inventory;
6
    this.orderService = services.order;
7
    this.notificationService = services.notification;
8
    this.sagaRepository = services.sagaRepository;
9
  }
10

11
  async processOrder(orderData) {
12
    const sagaId = this.generateSagaId();
13
    const saga = new SagaInstance(sagaId, "OrderProcessing", orderData);
14

15
    try {
16
      // Step 1: Reserve inventory
17
      saga.addStep("reserveInventory");
18
      await this.sagaRepository.save(saga);
19

20
      const inventoryResult = await this.inventoryService.reserveItems({
21
        items: orderData.items,
22
        sagaId: sagaId,
23
      });
24

25
      if (!inventoryResult.success) {
26
        throw new SagaStepFailedException("Inventory reservation failed");
27
      }
28

29
      saga.completeStep("reserveInventory", inventoryResult);
30

31
      // Step 2: Process payment
32
      saga.addStep("processPayment");
33
      await this.sagaRepository.save(saga);
34

35
      const paymentResult = await this.paymentService.processPayment({
36
        amount: orderData.totalAmount,
37
        paymentMethod: orderData.paymentMethod,
38
        sagaId: sagaId,
39
      });
40

41
      if (!paymentResult.success) {
42
        throw new SagaStepFailedException("Payment processing failed");
43
      }
44

45
      saga.completeStep("processPayment", paymentResult);
46

47
      // Step 3: Create order
48
      saga.addStep("createOrder");
49
      await this.sagaRepository.save(saga);
50

51
      const order = await this.orderService.createOrder({
52
        ...orderData,
53
        inventoryReservationId: inventoryResult.reservationId,
54
        paymentTransactionId: paymentResult.transactionId,
55
        sagaId: sagaId,
56
      });
57

58
      saga.completeStep("createOrder", order);
59

60
      // Step 4: Send confirmation
61
      saga.addStep("sendConfirmation");
62
      await this.sagaRepository.save(saga);
63

64
      await this.notificationService.sendOrderConfirmation({
65
        orderId: order.id,
66
        userId: orderData.userId,
67
        sagaId: sagaId,
68
      });
69

70
      saga.completeStep("sendConfirmation");
71
      saga.markAsCompleted();
72
      await this.sagaRepository.save(saga);
73

74
      return order;
75
    } catch (error) {
76
      // Execute compensation
77
      await this.compensate(saga, error);
78
      throw error;
79
    }
80
  }
81

82
  async compensate(saga, error) {
83
    console.log(`Compensating saga ${saga.id}: ${error.message}`);
84

85
    const completedSteps = saga.getCompletedSteps().reverse();
86

87
    for (const step of completedSteps) {
88
      try {
89
        switch (step.name) {
90
          case "sendConfirmation":
91
            // No compensation needed for notification
92
            break;
93

94
          case "createOrder":
95
            await this.orderService.cancelOrder(step.result.id);
96
            break;
97

98
          case "processPayment":
99
            await this.paymentService.refundPayment(step.result.transactionId);
100
            break;
101

102
          case "reserveInventory":
103
            await this.inventoryService.releaseReservation(
104
              step.result.reservationId
105
            );
106
            break;
107
        }
108

109
        saga.addCompensation(step.name, "completed");
110
      } catch (compensationError) {
111
        console.error(
112
          `Compensation failed for step ${step.name}: ${compensationError.message}`
113
        );
114
        saga.addCompensation(step.name, "failed", compensationError.message);
115
      }
116
    }
117

118
    saga.markAsFailed(error.message);
119
    await this.sagaRepository.save(saga);
120
  }
121
}
122

123
// Saga instance model
124
class SagaInstance {
125
  constructor(id, type, data) {
126
    this.id = id;
127
    this.type = type;
128
    this.data = data;
129
    this.status = "started";
130
    this.steps = [];
131
    this.compensations = [];
132
    this.createdAt = new Date();
133
    this.updatedAt = new Date();
134
  }
135

136
  addStep(stepName) {
137
    this.steps.push({
138
      name: stepName,
139
      status: "started",
140
      startedAt: new Date(),
141
    });
142
    this.updatedAt = new Date();
143
  }
144

145
  completeStep(stepName, result) {
146
    const step = this.steps.find(
147
      s => s.name === stepName && s.status === "started"
148
    );
149
    if (step) {
150
      step.status = "completed";
151
      step.completedAt = new Date();
152
      step.result = result;
153
    }
154
    this.updatedAt = new Date();
155
  }
156

157
  getCompletedSteps() {
158
    return this.steps.filter(s => s.status === "completed");
159
  }
160

161
  addCompensation(stepName, status, error = null) {
162
    this.compensations.push({
163
      stepName,
164
      status,
165
      error,
166
      timestamp: new Date(),
167
    });
168
    this.updatedAt = new Date();
169
  }
170

171
  markAsCompleted() {
172
    this.status = "completed";
173
    this.completedAt = new Date();
174
    this.updatedAt = new Date();
175
  }
176

177
  markAsFailed(error) {
178
    this.status = "failed";
179
    this.error = error;
180
    this.failedAt = new Date();
181
    this.updatedAt = new Date();
182
  }
183
}

1
// Event-driven choreography saga
2
class OrderServiceEventHandler {
3
  constructor(eventBus, orderRepository) {
4
    this.eventBus = eventBus;
5
    this.orderRepository = orderRepository;
6
    this.setupEventHandlers();
7
  }
8

9
  setupEventHandlers() {
10
    this.eventBus.subscribe(
11
      "InventoryReserved",
12
      this.handleInventoryReserved.bind(this)
13
    );
14
    this.eventBus.subscribe(
15
      "InventoryReservationFailed",
16
      this.handleInventoryReservationFailed.bind(this)
17
    );
18
    this.eventBus.subscribe(
19
      "PaymentProcessed",
20
      this.handlePaymentProcessed.bind(this)
21
    );
22
    this.eventBus.subscribe(
23
      "PaymentFailed",
24
      this.handlePaymentFailed.bind(this)
25
    );
26
  }
27

28
  async createOrder(orderData) {
29
    // Create pending order
30
    const order = await this.orderRepository.create({
31
      ...orderData,
32
      status: "pending",
33
    });
34

35
    // Publish event to start the saga
36
    await this.eventBus.publish("OrderCreationStarted", {
37
      orderId: order.id,
38
      items: orderData.items,
39
      userId: orderData.userId,
40
      totalAmount: orderData.totalAmount,
41
      timestamp: new Date(),
42
    });
43

44
    return order;
45
  }
46

47
  async handleInventoryReserved(event) {
48
    // Update order status
49
    await this.orderRepository.updateStatus(
50
      event.orderId,
51
      "inventory_reserved"
52
    );
53

54
    // Trigger payment processing
55
    await this.eventBus.publish("ProcessPayment", {
56
      orderId: event.orderId,
57
      amount: event.totalAmount,
58
      paymentMethod: event.paymentMethod,
59
      timestamp: new Date(),
60
    });
61
  }
62

63
  async handleInventoryReservationFailed(event) {
64
    // Mark order as failed
65
    await this.orderRepository.updateStatus(event.orderId, "failed");
66

67
    // Publish failure event
68
    await this.eventBus.publish("OrderFailed", {
69
      orderId: event.orderId,
70
      reason: "Inventory reservation failed",
71
      timestamp: new Date(),
72
    });
73
  }
74

75
  async handlePaymentProcessed(event) {
76
    // Confirm order
77
    await this.orderRepository.updateStatus(event.orderId, "confirmed");
78

79
    // Publish order confirmed event
80
    await this.eventBus.publish("OrderConfirmed", {
81
      orderId: event.orderId,
82
      paymentTransactionId: event.transactionId,
83
      timestamp: new Date(),
84
    });
85
  }
86

87
  async handlePaymentFailed(event) {
88
    // Mark order as failed
89
    await this.orderRepository.updateStatus(event.orderId, "failed");
90

91
    // Trigger inventory compensation
92
    await this.eventBus.publish("ReleaseInventoryReservation", {
93
      orderId: event.orderId,
94
      reservationId: event.reservationId,
95
      timestamp: new Date(),
96
    });
97

98
    // Publish failure event
99
    await this.eventBus.publish("OrderFailed", {
100
      orderId: event.orderId,
101
      reason: "Payment processing failed",
102
      timestamp: new Date(),
103
    });
104
  }
105
}