Handle massive read traffic and high-velocity engagement without database bottlenecks

plugins {
    java
    id("org.springframework.boot") version "4.1.1"
    id("io.spring.dependency-management") version "1.1.7"
}

java {
    toolchain {
        languageVersion = JavaLanguageVersion.of(25)
    }
}

dependencies {
    // Web & Validation
    implementation("org.springframework.boot:spring-boot-starter-web")
    implementation("org.springframework.boot:spring-boot-starter-validation")

    // Polyglot Persistence: PostgreSQL + MongoDB
    implementation("org.springframework.boot:spring-boot-starter-data-jpa")
    runtimeOnly("org.postgresql:postgresql")
    implementation("org.springframework.boot:spring-boot-starter-data-mongodb")

    // High-Volume Caching: Redis
    implementation("org.springframework.boot:spring-boot-starter-data-redis")
    implementation("org.springframework.boot:spring-boot-starter-cache")

    // Event Streaming & Batch Processing
    implementation("org.springframework.kafka:spring-kafka")
    implementation("org.springframework.boot:spring-boot-starter-batch")

    // Observability
    implementation("org.springframework.boot:spring-boot-starter-actuator")
}

spring:
  application:
    name: high-scale-blog
  threads:
    virtual:
      enabled: true # Java 21/25 Project Loom virtual threads everywhere

  # 1. PostgreSQL Configuration (HikariCP tuned for Virtual Threads)
  datasource:
    url: jdbc:postgresql://localhost:5432/blog_rel_db
    username: postgres
    password: secret
    hikari:
      maximum-pool-size: 50
      minimum-idle: 10
      connection-timeout: 20000
  jpa:
    open-in-view: false # Crucial for high-volume performance; prevents holding DB connections during view rendering
    hibernate:
      ddl-auto: validate
    properties:
      hibernate:
        jdbc.batch_size: 50
        order_inserts: true
        order_updates: true

  # 2. MongoDB Configuration
  data:
    mongodb:
      uri: mongodb://localhost:27017/blog_doc_db?maxPoolSize=100
    redis:
      host: localhost
      port: 6379
      timeout: 2000ms

  # 3. Redis Cache Configuration
  cache:
    type: redis
    redis:
      time-to-live: 3600000 # 1 hour TTL for cached blog posts
      cache-null-values: false

  # 4. Kafka Configuration (High-Throughput Batching)
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: 1
      batch-size: 32768
      properties:
        linger.ms: 10 # Batch messages for 10ms to maximize throughput
    consumer:
      group-id: blog-analytics-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      max-poll-records: 500 # Pull up to 500 events per batch
      properties:
        spring.json.trusted.packages: "com.example.blog.*"
    listener:
      type: batch # Process high-volume view events in batches

package com.example.blog.domain.postgres;

import jakarta.persistence.*;
import java.time.Instant;
import java.util.UUID;

@Entity
@Table(name = "post_metadata", indexes = {
    @Index(name = "idx_post_slug", columnList = "slug", unique = true),
    @Index(name = "idx_post_status_pub", columnList = "status, publishedAt")
})
public class PostMetadata {

    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    @Column(nullable = false, unique = true, length = 160)
    private String slug;

    @Column(nullable = false, length = 255)
    private String title;

    @Column(nullable = false)
    private UUID authorId;

    @Enumerated(EnumType.STRING)
    @Column(nullable = false, length = 20)
    private PostStatus status = PostStatus.DRAFT;

    @Column(nullable = false)
    private long viewCount = 0L;

    @Column(nullable = false, updatable = false)
    private Instant createdAt = Instant.now();

    private Instant publishedAt;

    public enum PostStatus { DRAFT, PUBLISHED, ARCHIVED }

    // Standard getters and setters
    public UUID getId() { return id; }
    public String getSlug() { return slug; }
    public void setSlug(String slug) { this.slug = slug; }
    public String getTitle() { return title; }
    public void setTitle(String title) { this.title = title; }
    public UUID getAuthorId() { return authorId; }
    public void setAuthorId(UUID authorId) { this.authorId = authorId; }
    public PostStatus getStatus() { return status; }
    public void setStatus(PostStatus status) { this.status = status; }
    public long getViewCount() { return viewCount; }
    public void setPublishedAt(Instant publishedAt) { this.publishedAt = publishedAt; }
}

package com.example.blog.repository.postgres;

import com.example.blog.domain.postgres.PostMetadata;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Modifying;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;

import java.util.Optional;
import java.util.UUID;

public interface PostMetadataRepository extends JpaRepository<PostMetadata, UUID> {

    Optional<PostMetadata> findBySlug(String slug);

    Page<PostMetadata> findByStatus(PostMetadata.PostStatus status, Pageable pageable);

    // Used by Spring Batch to increment views without entity locking
    @Modifying
    @Query("UPDATE PostMetadata p SET p.viewCount = p.viewCount + :delta WHERE p.id = :id")
    void incrementViewCount(@Param("id") UUID id, @Param("delta") long delta);
}

package com.example.blog.domain.mongo;

import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.Indexed;
import org.springframework.data.mongodb.core.mapping.Document;
import java.time.Instant;
import java.util.List;
import java.util.UUID;

@Document(collection = "post_contents")
public record PostContent(
    @Id String id, // Matches PostgreSQL PostMetadata UUID as String
    @Indexed(unique = true) String slug,
    String markdownBody,
    List<CodeSnippet> codeSnippets,
    List<String> tags,
    Instant lastModified
) {
    public record CodeSnippet(String language, String filename, String code) {}
}

package com.example.blog.domain.mongo;

import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.Indexed;
import org.springframework.data.mongodb.core.mapping.Document;
import java.time.Instant;
import java.util.UUID;

@Document(collection = "raw_view_logs")
public record RawViewLog(
    @Id String id,
    @Indexed UUID postId,
    String userAgent,
    Instant viewedAt,
    @Indexed boolean processedByBatch
) {}

package com.example.blog.repository.mongo;

import com.example.blog.domain.mongo.PostContent;
import org.springframework.data.mongodb.repository.MongoRepository;
import java.util.List;

public interface PostContentRepository extends MongoRepository<PostContent, String> {
    List<PostContent> findByTagsContaining(String tag);
}

package com.example.blog.dto;

import com.example.blog.domain.mongo.PostContent.CodeSnippet;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import java.io.Serializable;
import java.time.Instant;
import java.util.List;
import java.util.UUID;

public record CreatePostRequest(
    @NotBlank String title,
    @NotBlank String slug,
    @NotNull UUID authorId,
    @NotBlank String markdownBody,
    List<CodeSnippet> codeSnippets,
    List<String> tags
) {}

// Implements Serializable (or use Jackson RedisSerializer) for Redis caching
public record FullPostResponse(
    UUID id,
    String slug,
    String title,
    UUID authorId,
    long viewCount,
    String markdownBody,
    List<CodeSnippet> codeSnippets,
    List<String> tags,
    Instant publishedAt
) implements Serializable {}

public record PostViewedEvent(UUID postId, String slug, String userAgent, Instant timestamp) {}

package com.example.blog.service;

import com.example.blog.domain.mongo.PostContent;
import com.example.blog.domain.postgres.PostMetadata;
import com.example.blog.dto.*;
import com.example.blog.repository.mongo.PostContentRepository;
import com.example.blog.repository.postgres.PostMetadataRepository;
import org.springframework.cache.annotation.CacheEvict;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.UUID;

@Service
public class BlogService {

    private final PostMetadataRepository metadataRepo;
    private final PostContentRepository contentRepo;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public BlogService(PostMetadataRepository metadataRepo,
                       PostContentRepository contentRepo,
                       KafkaTemplate<String, Object> kafkaTemplate) {
        this.metadataRepo = metadataRepo;
        this.contentRepo = contentRepo;
        this.kafkaTemplate = kafkaTemplate;
    }

    // 1. CREATE: Saves relational metadata in Postgres & rich document in MongoDB
    @Transactional
    public FullPostResponse createPost(CreatePostRequest req) {
        PostMetadata meta = new PostMetadata();
        meta.setSlug(req.slug());
        meta.setTitle(req.title());
        meta.setAuthorId(req.authorId());
        meta.setStatus(PostMetadata.PostStatus.PUBLISHED);
        meta.setPublishedAt(Instant.now());
        PostMetadata savedMeta = metadataRepo.save(meta);

        PostContent content = new PostContent(
            savedMeta.getId().toString(),
            req.slug(),
            req.markdownBody(),
            req.codeSnippets(),
            req.tags(),
            Instant.now()
        );
        contentRepo.save(content);

        return mapToResponse(savedMeta, content);
    }

    // 2. READ (Cached): Hits Redis first. On cache miss, fetches from Postgres + MongoDB.
    @Cacheable(value = "posts", key = "#slug")
    @Transactional(readOnly = true)
    public FullPostResponse getPostBySlug(String slug) {
        PostMetadata meta = metadataRepo.findBySlug(slug)
            .orElseThrow(() -> new IllegalArgumentException("Post not found: " + slug));

        PostContent content = contentRepo.findById(meta.getId().toString())
            .orElseThrow(() -> new IllegalStateException("Content missing for post: " + slug));

        return mapToResponse(meta, content);
    }

    // 3. ASYNC VIEW TRACKING: Fired on every read without blocking or evicting the cache
    public void recordViewAsync(UUID postId, String slug, String userAgent) {
        PostViewedEvent event = new PostViewedEvent(postId, slug, userAgent, Instant.now());
        // Partition by postId so events for the same post stay ordered
        kafkaTemplate.send("blog.post.views", postId.toString(), event);
    }

    // 4. UPDATE: Updates both databases and evicts the stale Redis cache entry
    @CacheEvict(value = "posts", key = "#slug")
    @Transactional
    public FullPostResponse updatePost(String slug, CreatePostRequest req) {
        PostMetadata meta = metadataRepo.findBySlug(slug)
            .orElseThrow(() -> new IllegalArgumentException("Post not found: " + slug));
        meta.setTitle(req.title());
        metadataRepo.save(meta);

        PostContent updatedContent = new PostContent(
            meta.getId().toString(),
            slug,
            req.markdownBody(),
            req.codeSnippets(),
            req.tags(),
            Instant.now()
        );
        contentRepo.save(updatedContent);

        return mapToResponse(meta, updatedContent);
    }

    // 5. DELETE: Cleans up both stores and evicts cache
    @CacheEvict(value = "posts", key = "#slug")
    @Transactional
    public void deletePost(String slug) {
        PostMetadata meta = metadataRepo.findBySlug(slug)
            .orElseThrow(() -> new IllegalArgumentException("Post not found: " + slug));
        contentRepo.deleteById(meta.getId().toString());
        metadataRepo.delete(meta);
    }

    private FullPostResponse mapToResponse(PostMetadata meta, PostContent content) {
        return new FullPostResponse(
            meta.getId(), meta.getSlug(), meta.getTitle(), meta.getAuthorId(),
            meta.getViewCount(), content.markdownBody(), content.codeSnippets(),
            content.tags(), meta.getPublishedAt()
        );
    }
}

package com.example.blog.controller;

import com.example.blog.dto.CreatePostRequest;
import com.example.blog.dto.FullPostResponse;
import com.example.blog.service.BlogService;
import jakarta.validation.Valid;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/v1/posts")
public class BlogController {

    private final BlogService blogService;

    public BlogController(BlogService blogService) {
        this.blogService = blogService;
    }

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public FullPostResponse create(@Valid @RequestBody CreatePostRequest request) {
        return blogService.createPost(request);
    }

    @GetMapping("/{slug}")
    public FullPostResponse getBySlug(
            @PathVariable String slug,
            @RequestHeader(value = "User-Agent", defaultValue = "unknown") String userAgent) {
        // 1. Served from Redis Cache in ~1ms
        FullPostResponse post = blogService.getPostBySlug(slug);
        // 2. Non-blocking Kafka fire-and-forget for high-volume view counting
        blogService.recordViewAsync(post.id(), post.slug(), userAgent);
        return post;
    }

    @PutMapping("/{slug}")
    public FullPostResponse update(
            @PathVariable String slug,
            @Valid @RequestBody CreatePostRequest request) {
        return blogService.updatePost(slug, request);
    }

    @DeleteMapping("/{slug}")
    @ResponseStatus(HttpStatus.NO_CONTENT)
    public void delete(@PathVariable String slug) {
        blogService.deletePost(slug);
    }
}

package com.example.blog.messaging;

import com.example.blog.domain.mongo.RawViewLog;
import com.example.blog.dto.PostViewedEvent;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.UUID;

@Component
public class ViewEventBatchConsumer {

    private final MongoTemplate mongoTemplate;

    public ViewEventBatchConsumer(MongoTemplate mongoTemplate) {
        this.mongoTemplate = mongoTemplate;
    }

    // Receives up to 500 events per poll (configured via max-poll-records)
    @KafkaListener(topics = "blog.post.views", groupId = "blog-analytics-group")
    public void consumeViewEventsBatch(List<PostViewedEvent> events) {
        List<RawViewLog> logs = events.stream()
            .map(e -> new RawViewLog(
                UUID.randomUUID().toString(),
                e.postId(),
                e.userAgent(),
                e.timestamp(),
                false
            ))
            .toList();

        // Single bulk network round-trip to MongoDB
        mongoTemplate.insertAll(logs);
    }
}

package com.example.blog.batch;

import com.example.blog.domain.mongo.RawViewLog;
import com.example.blog.repository.postgres.PostMetadataRepository;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.data.MongoPagingItemReader;
import org.springframework.batch.item.data.builder.MongoPagingItemReaderBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.PlatformTransactionManager;

import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.stream.Collectors;

@Configuration
@EnableScheduling
public class ViewAggregationBatchConfig {

    // 1. READER: Streams unprocessed view logs from MongoDB in pages
    @Bean
    public MongoPagingItemReader<RawViewLog> viewLogReader(MongoTemplate mongoTemplate) {
        return new MongoPagingItemReaderBuilder<RawViewLog>()
            .name("viewLogReader")
            .template(mongoTemplate)
            .targetType(RawViewLog.class)
            .jsonQuery("{ 'processedByBatch': false }")
            .sorts(Map.of("viewedAt", Sort.Direction.ASC))
            .pageSize(1000)
            .build();
    }

    // 2. WRITER: Aggregates views by postId in memory per chunk and flushes to Postgres
    @Bean
    public ItemWriter<RawViewLog> viewLogAggregateWriter(
            PostMetadataRepository metadataRepo,
            MongoTemplate mongoTemplate) {
        return (Chunk<? extends RawViewLog> chunk) -> {
            List<? extends RawViewLog> items = chunk.getItems();
            if (items.isEmpty()) return;

            // Group by postId and count views in the current 1,000-record chunk
            Map<UUID, Long> viewCountsByPost = items.stream()
                .collect(Collectors.groupingBy(RawViewLog::postId, Collectors.counting()));

            // Update PostgreSQL totals (only 1 UPDATE per unique post in the chunk)
            viewCountsByPost.forEach(metadataRepo::incrementViewCount);

            // Bulk mark raw logs as processed in MongoDB
            List<String> processedIds = items.stream().map(RawViewLog::id).toList();
            mongoTemplate.updateMulti(
                Query.query(Criteria.where("_id").in(processedIds)),
                Update.update("processedByBatch", true),
                RawViewLog.class
            );
        };
    }

    // 3. STEP: Processes 1,000 logs per transaction
    @Bean
    public Step aggregateViewsStep(
            JobRepository jobRepository,
            PlatformTransactionManager transactionManager,
            MongoPagingItemReader<RawViewLog> reader,
            ItemWriter<RawViewLog> writer) {
        return new StepBuilder("aggregateViewsStep", jobRepository)
            .<RawViewLog, RawViewLog>chunk(1000, transactionManager)
            .reader(reader)
            .writer(writer)
            .build();
    }

    // 4. JOB: Wraps the step
    @Bean
    public Job viewSyncJob(JobRepository jobRepository, Step aggregateViewsStep) {
        return new JobBuilder("viewSyncJob", jobRepository)
            .start(aggregateViewsStep)
            .build();
    }

    // 5. SCHEDULER: Triggers the Batch Job every 5 minutes on a Virtual Thread
    @Scheduled(fixedDelay = 300_000)
    public void runViewSyncJob(JobLauncher jobLauncher, Job viewSyncJob) throws Exception {
        jobLauncher.run(
            viewSyncJob,
            new JobParametersBuilder()
                .addLong("runAt", System.currentTimeMillis())
                .toJobParameters()
        );
    }
}