Microservices e-commerce #6: MongoDB, MySQL và database-per-service

dependencies {
    implementation project(':util')
    implementation 'org.springframework.boot:spring-boot-starter-actuator'
    implementation 'org.springframework.boot:spring-boot-starter-webflux'
    implementation 'org.springframework.boot:spring-boot-starter-data-mongodb-reactive'
    implementation "org.mapstruct:mapstruct:${mapstructVersion}"
    annotationProcessor "org.mapstruct:mapstruct-processor:${mapstructVersion}"

    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'io.projectreactor:reactor-test'
    testImplementation 'org.springframework.boot:spring-boot-testcontainers'
    testImplementation 'org.testcontainers:junit-jupiter'
    testImplementation 'org.testcontainers:mongodb'
}

    implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
    runtimeOnly 'com.mysql:mysql-connector-j'
    // ...
    testImplementation 'org.testcontainers:mysql'

@Document(collection = "products")
public class ProductEntity {

    @Id
    private String id;

    @Version
    private Integer version;

    @Indexed(unique = true)
    private int productId;

    private String name;
    // Mặc định Spring Data lưu BigDecimal thành chuỗi; Decimal128 giữ đúng kiểu số thập phân của MongoDB
    @Field(targetType = FieldType.DECIMAL128)
    private BigDecimal price;

    // constructor, getter, setter
}

{
  _id: ObjectId('6abf52b48c9d6c63afc5ad01'),
  version: 0,
  productId: 1,
  name: 'Giay sneaker',
  price: '899000',
  _class: 'com.ecommerce.core.product.persistence.ProductEntity'
}

$ docker compose exec -T mongodb mongosh product-db --quiet --eval "db.products.find({productId: 1})"
[
  {
    _id: ObjectId('6abf53043b6bbc433455ceeb'),
    version: 0,
    productId: 1,
    name: 'Giay sneaker',
    price: Decimal128('899000'),
    _class: 'com.ecommerce.core.product.persistence.ProductEntity'
  }
]

@Document(collection = "recommendations")
@CompoundIndex(name = "prod-rec-id", unique = true, def = "{'productId': 1, 'recommendationId' : 1}")
public class RecommendationEntity {

    @Id
    private String id;

    @Version
    private Integer version;

    private int productId;
    private int recommendationId;
    private String author;
    private int rating;
    private String content;
    // ...
}

public interface ProductRepository extends ReactiveCrudRepository<ProductEntity, String> {

    Mono<ProductEntity> findByProductId(int productId);
}

spring.data.mongodb:
  host: localhost
  port: 27017
  database: product-db
  auto-index-creation: true

---
spring.config.activate.on-profile: docker

server.port: 8080
spring.data.mongodb.host: mongodb

@Entity
@Table(name = "reviews", indexes = {
    @Index(name = "reviews_unique_idx", unique = true, columnList = "productId,reviewId")
})
public class ReviewEntity {

    @Id
    @GeneratedValue
    private int id;

    @Version
    private int version;

    private int productId;
    private int reviewId;
    private String author;
    private String subject;
    private String content;
    // ...
}

public interface ReviewRepository extends CrudRepository<ReviewEntity, Integer> {

    @Transactional(readOnly = true)
    List<ReviewEntity> findByProductId(int productId);
}

# Chỉ dùng "update" khi phát triển; production nên dùng Flyway/Liquibase.
spring.jpa.hibernate.ddl-auto: update
spring.jpa.open-in-view: false

spring.datasource:
  url: jdbc:mysql://localhost/review_db
  username: ${MYSQL_USER:user}
  password: ${MYSQL_PASSWORD:pwd}
  hikari.initializationFailTimeout: 60000

@Mapper(componentModel = "spring")
public interface ProductMapper {

    @Mappings({
        @Mapping(target = "serviceAddress", ignore = true)
    })
    Product entityToApi(ProductEntity entity);

    @Mappings({
        @Mapping(target = "id", ignore = true),
        @Mapping(target = "version", ignore = true)
    })
    ProductEntity apiToEntity(Product api);
}

@Override
public Mono<Product> createProduct(Product body) {
    if (body.productId() < 1) {
        throw new InvalidInputException("Invalid productId: " + body.productId());
    }
    ProductEntity entity = mapper.apiToEntity(body);
    return repository.save(entity)
        .onErrorMap(DuplicateKeyException.class,
            ex -> new InvalidInputException("Duplicate key, Product Id: " + body.productId()))
        .map(mapper::entityToApi);
}

@Override
public Mono<Product> getProduct(int productId) {
    if (productId < 1) {
        throw new InvalidInputException("Invalid productId: " + productId);
    }
    LOG.debug("Will get product info for id={}", productId);
    return repository.findByProductId(productId)
        .switchIfEmpty(Mono.error(new NotFoundException("No product found for productId: " + productId)))
        .map(mapper::entityToApi)
        .map(this::setServiceAddress);
}

@Override
public Mono<Void> deleteProduct(int productId) {
    if (productId < 1) {
        throw new InvalidInputException("Invalid productId: " + productId);
    }
    LOG.debug("deleteProduct: tries to delete an entity with productId: {}", productId);
    return repository.findByProductId(productId).map(repository::delete).flatMap(e -> e);
}

/** JPA là blocking, nên chạy trên một thread pool riêng để không chặn event loop của WebFlux. */
@Bean
public Scheduler jdbcScheduler(
        @Value("${app.threadPoolSize:10}") Integer threadPoolSize,
        @Value("${app.taskQueueSize:100}") Integer taskQueueSize) {
    return Schedulers.newBoundedElastic(threadPoolSize, taskQueueSize, "jdbc-pool");
}

@Override
public Mono<Review> createReview(Review body) {
    if (body.productId() < 1) {
        throw new InvalidInputException("Invalid productId: " + body.productId());
    }
    return Mono.fromCallable(() -> internalCreateReview(body)).subscribeOn(jdbcScheduler);
}

private Review internalCreateReview(Review body) {
    try {
        ReviewEntity entity = mapper.apiToEntity(body);
        ReviewEntity newEntity = repository.save(entity);
        LOG.debug("createReview: created a review entity: {}/{}", body.productId(), body.reviewId());
        return mapper.entityToApi(newEntity);
    } catch (DataIntegrityViolationException dive) {
        throw new InvalidInputException(
            "Duplicate key, Product Id: " + body.productId() + ", Review Id:" + body.reviewId());
    }
}

@Override
public Flux<Review> getReviews(int productId) {
    if (productId < 1) {
        throw new InvalidInputException("Invalid productId: " + productId);
    }
    LOG.info("Will get reviews for product with id={}", productId);
    return Mono.fromCallable(() -> internalGetReviews(productId))
        .flatMapMany(Flux::fromIterable)
        .subscribeOn(jdbcScheduler);
}

public interface ProductService {

    @PostMapping(value = "/product", consumes = "application/json", produces = "application/json")
    Mono<Product> createProduct(@RequestBody Product body);

    @GetMapping(value = "/product/{productId}", produces = "application/json")
    Mono<Product> getProduct(@PathVariable int productId);

    @DeleteMapping(value = "/product/{productId}")
    Mono<Void> deleteProduct(@PathVariable int productId);
}

@Override
public Mono<Void> createProduct(ProductAggregate body) {
    LOG.info("Will create a new composite entity for product.id: {}", body.productId());

    List<Mono<?>> monoList = new ArrayList<>();
    monoList.add(integration.createProduct(new Product(body.productId(), body.name(), body.price(), null)));

    if (body.recommendations() != null) {
        body.recommendations().forEach(r -> monoList.add(integration.createRecommendation(
            new Recommendation(body.productId(), r.recommendationId(), r.author(), r.rate(), r.content(), null))));
    }

    if (body.reviews() != null) {
        body.reviews().forEach(r -> monoList.add(integration.createReview(
            new Review(body.productId(), r.reviewId(), r.author(), r.subject(), r.content(), null))));
    }

    return Mono.when(monoList)
        .doOnError(ex -> LOG.warn("createCompositeProduct failed: {}", ex.toString()));
}

@Override
public Mono<Void> deleteProduct(int productId) {
    LOG.info("Will delete a product aggregate for product.id: {}", productId);
    return Mono.when(
            integration.deleteProduct(productId),
            integration.deleteRecommendations(productId),
            integration.deleteReviews(productId))
        .doOnError(ex -> LOG.warn("delete failed: {}", ex.toString()));
}

@Override
public Mono<Product> createProduct(Product body) {
    return webClient.post().uri(productServiceUrl).bodyValue(body).retrieve()
        .bodyToMono(Product.class)
        .onErrorMap(WebClientResponseException.class, this::handleException);
}

@Testcontainers
@DataMongoTest
class PersistenceTests {

    @Container
    @ServiceConnection
    static MongoDBContainer database = new MongoDBContainer("mongo:7.0");

    @Autowired
    private ProductRepository repository;

    private ProductEntity savedEntity;

    @BeforeEach
    void setupDb() {
        StepVerifier.create(repository.deleteAll()).verifyComplete();

        ProductEntity entity = new ProductEntity(1, "n", new BigDecimal("199000"));
        StepVerifier.create(repository.save(entity))
            .consumeNextWith(created -> savedEntity = created)
            .verifyComplete();
    }

    @Test
    void duplicateError() {
        ProductEntity entity = new ProductEntity(savedEntity.getProductId(), "n", new BigDecimal("1"));
        StepVerifier.create(repository.save(entity)).expectError(DuplicateKeyException.class).verify();
    }

    @Test
    void optimisticLockError() {
        // Đọc cùng một entity vào hai biến khác nhau
        ProductEntity entity1 = repository.findById(savedEntity.getId()).block();
        ProductEntity entity2 = repository.findById(savedEntity.getId()).block();

        // Cập nhật bằng biến thứ nhất, version tăng lên 1
        entity1.setName("n1");
        repository.save(entity1).block();

        // Biến thứ hai vẫn giữ version cũ nên phải bị từ chối
        entity2.setName("n2");
        StepVerifier.create(repository.save(entity2)).expectError(OptimisticLockingFailureException.class).verify();

        StepVerifier.create(repository.findById(savedEntity.getId()))
            .expectNextMatches(found -> found.getVersion() == 1 && found.getName().equals("n1"))
            .verifyComplete();
    }

    // create, update, delete, getByProductId ...
}

@Testcontainers
@DataJpaTest
@AutoConfigureTestDatabase(replace = AutoConfigureTestDatabase.Replace.NONE)
@Transactional(propagation = Propagation.NOT_SUPPORTED)
class PersistenceTests {

    @Container
    @ServiceConnection
    static MySQLContainer<?> database = new MySQLContainer<>("mysql:8.4");
    // ...
}

./gradlew build

  review:
    build: microservices/review-service
    mem_limit: 512m
    environment:
      - SPRING_PROFILES_ACTIVE=docker
      - MYSQL_USER=${MYSQL_USER}
      - MYSQL_PASSWORD=${MYSQL_PASSWORD}
    depends_on:
      mysql:
        condition: service_healthy

  mongodb:
    image: mongo:7.0
    mem_limit: 512m
    ports:
      - "27017:27017"
    command: mongod
    healthcheck:
      test: ["CMD", "mongosh", "--quiet", "--eval", "db.runCommand('ping').ok"]
      interval: 5s
      timeout: 2s
      retries: 60

  mysql:
    image: mysql:8.4
    mem_limit: 512m
    ports:
      - "3306:3306"
    environment:
      - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
      - MYSQL_DATABASE=review_db
      - MYSQL_USER=${MYSQL_USER}
      - MYSQL_PASSWORD=${MYSQL_PASSWORD}
    healthcheck:
      test: ["CMD", "mysqladmin", "ping", "-uroot", "-p${MYSQL_ROOT_PASSWORD}", "-h", "localhost"]
      interval: 5s
      timeout: 2s
      retries: 60

# Chỉ dùng khi phát triển trên máy. Không dùng mật khẩu này ở môi trường thật.
MYSQL_ROOT_PASSWORD=rootpwd
MYSQL_USER=user
MYSQL_PASSWORD=pwd

function recreateComposite() {
  local productId=$1
  local composite=$2

  assertCurl 200 "curl -X DELETE http://$HOST:$PORT/product-composite/${productId} -s"
  assertEqual 200 $(curl -X POST -s http://$HOST:$PORT/product-composite -H "Content-Type: application/json" \
    --data "$composite" -w "%{http_code}")
}

# Gọi DELETE tới composite: chỉ thành công khi composite và cả ba service lõi đều đã sẵn sàng
waitForService curl -X DELETE http://$HOST:$PORT/product-composite/$PROD_ID_NOT_FOUND

# Xoá là idempotent: xoá hai lần đều trả về 200, sau đó sản phẩm không còn
assertCurl 200 "curl -X DELETE http://$HOST:$PORT/product-composite/$PROD_ID_NO_REVS -s"
assertCurl 200 "curl -X DELETE http://$HOST:$PORT/product-composite/$PROD_ID_NO_REVS -s"
assertCurl 404 "curl http://$HOST:$PORT/product-composite/$PROD_ID_NO_REVS -s"

$ ./gradlew build && docker compose build
$ ./test-em-all.bash start
...
Wait for: curl -X DELETE http://localhost:8080/product-composite/13... , retry #1 , retry #2 , retry #3 , retry #4 DONE, continues...
...
End, all tests OK: Fri Oct 2 06:45:26 UTC 2026

$ docker compose exec -T mongodb mongosh product-db --quiet --eval "db.products.getIndexes()"
...
  { v: 2, key: { _id: 1 }, name: '_id_' },
  { v: 2, key: { productId: 1 }, name: 'productId', unique: true }

$ docker compose exec -T mysql mysql -uuser -ppwd \
    -e "select id, version, product_id, review_id, author from review_db.reviews"
id	version	product_id	review_id	author
1	0	113	3	author 3
2	0	113	1	author 1
3	0	113	2	author 2
4	0	1	1	author 1
5	0	1	3	author 3
6	0	1	2	author 2

git add .
git commit -m "Bài 6: lưu dữ liệu với MongoDB và MySQL"
git tag blog-06