feat: Implement Distributed & Reactive CacheFlow Strategy

build.gradle.kts

@@ -58,7 +58,9 @@ dependencies {
     implementation("org.jetbrains.kotlin:kotlin-reflect")
     implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core")
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core")
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor")
+    implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
 
     implementation("software.amazon.awssdk:cloudfront:2.21.29")
 

src/main/kotlin/io/cacheflow/spring/annotation/CacheFlowUpdate.kt

@@ -0,0 +1,23 @@
+package io.cacheflow.spring.annotation
+
+import java.lang.annotation.Inherited
+
+/**
+ * Annotation to trigger an update (touch) on a parent entity when a method is executed.
+ *
+ * This is useful for "Russian Doll" caching where updating a child entity should invalidate
+ * or update the parent entity's cache key (e.g. by updating its updatedAt timestamp).
+ *
+ * @property parent SpEL expression to evaluate the parent ID (e.g., "#entity.parentId" or "#args[0]").
+ * @property entityType The type of the parent entity (e.g., "user", "organization").
+ * @property condition SpEL expression to verify if the update should proceed.
+ */
+@Target(AnnotationTarget.FUNCTION)
+@Retention(AnnotationRetention.RUNTIME)
+@Inherited
+@MustBeDocumented
+annotation class CacheFlowUpdate(
+    val parent: String,
+    val entityType: String,
+    val condition: String = "",
+)

src/main/kotlin/io/cacheflow/spring/aspect/FragmentCacheAspect.kt

@@ -82,12 +82,15 @@ class FragmentCacheAspect(
         val result = joinPoint.proceed()
         if (result is String) {
             val ttl = if (fragment.ttl > 0) fragment.ttl else defaultTtlSeconds
-            
+
             // Evaluate tags
-            val evaluatedTags = fragment.tags.map { tag ->
-                evaluateFragmentKeyExpression(tag, joinPoint)
-            }.filter { it.isNotBlank() }.toSet()
-
+            val evaluatedTags =
+                fragment.tags
+                    .map { tag ->
+                        evaluateFragmentKeyExpression(tag, joinPoint)
+                    }.filter { it.isNotBlank() }
+                    .toSet()
+
             fragmentCacheService.cacheFragment(key, result, ttl, evaluatedTags)
 
             // Add tags to local tag manager for local tracking
@@ -137,12 +140,15 @@ class FragmentCacheAspect(
 
         return if (composedResult.isNotBlank()) {
             val ttl = if (composition.ttl > 0) composition.ttl else defaultTtlSeconds
-            
+
             // Evaluate tags for composition
-            val evaluatedTags = composition.tags.map { tag ->
-                evaluateFragmentKeyExpression(tag, joinPoint)
-            }.filter { it.isNotBlank() }.toSet()
-
+            val evaluatedTags =
+                composition.tags
+                    .map { tag ->
+                        evaluateFragmentKeyExpression(tag, joinPoint)
+                    }.filter { it.isNotBlank() }
+                    .toSet()
+
             fragmentCacheService.cacheFragment(key, composedResult, ttl, evaluatedTags)
             composedResult
         } else {

src/main/kotlin/io/cacheflow/spring/aspect/ParentToucher.kt

@@ -0,0 +1,21 @@
+package io.cacheflow.spring.aspect
+
+/**
+ * Interface to define how to "touch" a parent entity to update its timestamp.
+ *
+ * Implementations should update the 'updatedAt' (or equivalent) timestamp of the
+ * specified entity, triggering a cache invalidation or refresh for any Russian Doll
+ * caches that depend on that parent.
+ */
+interface ParentToucher {
+    /**
+     * Touches the specified parent entity.
+     *
+     * @param entityType The type string from @CacheFlowUpdate
+     * @param parentId The ID of the parent entity
+     */
+    fun touch(
+        entityType: String,
+        parentId: String,
+    )
+}

src/main/kotlin/io/cacheflow/spring/aspect/TouchPropagationAspect.kt

@@ -0,0 +1,82 @@
+package io.cacheflow.spring.aspect
+
+import io.cacheflow.spring.annotation.CacheFlowUpdate
+import org.aspectj.lang.JoinPoint
+import org.aspectj.lang.annotation.AfterReturning
+import org.aspectj.lang.annotation.Aspect
+import org.aspectj.lang.reflect.MethodSignature
+import org.slf4j.LoggerFactory
+import org.springframework.context.expression.MethodBasedEvaluationContext
+import org.springframework.core.DefaultParameterNameDiscoverer
+import org.springframework.expression.ExpressionParser
+import org.springframework.expression.spel.standard.SpelExpressionParser
+import org.springframework.stereotype.Component
+
+/**
+ * Aspect to handle [CacheFlowUpdate] annotations.
+ *
+ * This aspect intercepts methods annotated with @CacheFlowUpdate and executes the
+ * [ParentToucher.touch] method for the resolved parent entity.
+ */
+@Aspect
+@Component
+class TouchPropagationAspect(
+    private val parentToucher: ParentToucher?,
+) {
+    private val logger = LoggerFactory.getLogger(TouchPropagationAspect::class.java)
+    private val parser: ExpressionParser = SpelExpressionParser()
+    private val parameterNameDiscoverer = DefaultParameterNameDiscoverer()
+
+    @AfterReturning("@annotation(io.cacheflow.spring.annotation.CacheFlowUpdate)")
+    fun handleUpdate(joinPoint: JoinPoint) {
+        if (parentToucher == null) {
+            logger.debug("No ParentToucher bean found. Skipping @CacheFlowUpdate processing.")
+            return
+        }
+
+        val signature = joinPoint.signature as MethodSignature
+        var method = signature.method
+        var annotation = method.getAnnotation(CacheFlowUpdate::class.java)
+
+        // If annotation is not on the interface method, check the implementation class
+        if (annotation == null && joinPoint.target != null) {
+            try {
+                val targetMethod =
+                    joinPoint.target.javaClass.getMethod(method.name, *method.parameterTypes)
+                annotation = targetMethod.getAnnotation(CacheFlowUpdate::class.java)
+                method = targetMethod // Use the target method for context evaluation
+            } catch (e: NoSuchMethodException) {
+                // Ignore, keep original method
+            }
+        }
+
+        if (annotation == null) return
+
+        try {
+            val context =
+                MethodBasedEvaluationContext(
+                    joinPoint.target,
+                    method,
+                    joinPoint.args,
+                    parameterNameDiscoverer,
+                )
+
+            // Check condition if present
+            if (annotation.condition.isNotBlank()) {
+                val conditionMet =
+                    parser.parseExpression(annotation.condition).getValue(context, Boolean::class.java)
+                if (conditionMet != true) return
+            }
+
+            // Resolve parent ID
+            val parentId =
+                parser.parseExpression(annotation.parent).getValue(context, String::class.java)
+
+            if (!parentId.isNullOrBlank()) {
+                parentToucher.touch(annotation.entityType, parentId)
+            }
+        } catch (e: Exception) {
+            logger.error("Error processing @CacheFlowUpdate", e)
+        }
+    }
+}

src/main/kotlin/io/cacheflow/spring/autoconfigure/CacheFlowAspectConfiguration.kt

@@ -75,4 +75,18 @@ class CacheFlowAspectConfiguration {
         dependencyResolver: DependencyResolver,
         tagManager: FragmentTagManager,
     ): FragmentCacheAspect = FragmentCacheAspect(fragmentCacheService, dependencyResolver, tagManager)
+
+    /**
+     * Creates the touch propagation aspect bean.
+     *
+     * @param parentToucher The parent toucher (optional)
+     * @return The touch propagation aspect
+     */
+    @Bean
+    @ConditionalOnMissingBean
+    fun touchPropagationAspect(
+        @org.springframework.beans.factory.annotation.Autowired(required = false) parentToucher: io.cacheflow.spring.aspect.ParentToucher?,
+    ): io.cacheflow.spring.aspect.TouchPropagationAspect =
+        io.cacheflow.spring.aspect
+            .TouchPropagationAspect(parentToucher)
 }

src/main/kotlin/io/cacheflow/spring/autoconfigure/CacheFlowAutoConfiguration.kt

@@ -1,9 +1,10 @@
 package io.cacheflow.spring.autoconfigure
 
+import io.cacheflow.spring.autoconfigure.CacheFlowWarmingConfiguration
 import io.cacheflow.spring.config.CacheFlowProperties
+import org.springframework.boot.autoconfigure.AutoConfiguration
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.boot.context.properties.EnableConfigurationProperties
-import org.springframework.context.annotation.Configuration
 import org.springframework.context.annotation.Import
 
 /**
@@ -13,19 +14,15 @@ import org.springframework.context.annotation.Import
  * configuration properties.
  */
 
-@Configuration
-@ConditionalOnProperty(
-    prefix = "cacheflow",
-    name = ["enabled"],
-    havingValue = "true",
-    matchIfMissing = true,
-)
+@AutoConfiguration
+@ConditionalOnProperty(prefix = "cacheflow", name = ["enabled"], havingValue = "true", matchIfMissing = true)
 @EnableConfigurationProperties(CacheFlowProperties::class)
 @Import(
     CacheFlowCoreConfiguration::class,
     CacheFlowFragmentConfiguration::class,
+    CacheFlowRedisConfiguration::class,
     CacheFlowAspectConfiguration::class,
     CacheFlowManagementConfiguration::class,
-    CacheFlowRedisConfiguration::class,
+    CacheFlowWarmingConfiguration::class,
 )
 class CacheFlowAutoConfiguration

src/main/kotlin/io/cacheflow/spring/autoconfigure/CacheFlowCoreConfiguration.kt

@@ -42,7 +42,8 @@ class CacheFlowCoreConfiguration {
         @Autowired(required = false) @Qualifier("cacheFlowRedisTemplate") redisTemplate: RedisTemplate<String, Any>?,
         @Autowired(required = false) edgeCacheService: EdgeCacheIntegrationService?,
         @Autowired(required = false) meterRegistry: MeterRegistry?,
-    ): CacheFlowService = CacheFlowServiceImpl(properties, redisTemplate, edgeCacheService, meterRegistry)
+        @Autowired(required = false) redisCacheInvalidator: io.cacheflow.spring.messaging.RedisCacheInvalidator?,
+    ): CacheFlowService = CacheFlowServiceImpl(properties, redisTemplate, edgeCacheService, meterRegistry, redisCacheInvalidator)
 
     /**
      * Creates the dependency resolver bean.
@@ -51,7 +52,10 @@ class CacheFlowCoreConfiguration {
      */
     @Bean
     @ConditionalOnMissingBean
-    fun dependencyResolver(): DependencyResolver = CacheDependencyTracker()
+    fun dependencyResolver(
+        properties: CacheFlowProperties,
+        @Autowired(required = false) redisTemplate: org.springframework.data.redis.core.StringRedisTemplate?,
+    ): DependencyResolver = CacheDependencyTracker(properties, redisTemplate)
 
     /**
      * Creates the timestamp extractor bean.

src/main/kotlin/io/cacheflow/spring/autoconfigure/CacheFlowRedisConfiguration.kt

@@ -1,5 +1,6 @@
 package io.cacheflow.spring.autoconfigure
 
+import com.fasterxml.jackson.databind.ObjectMapper
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
@@ -9,13 +10,11 @@ import org.springframework.data.redis.connection.RedisConnectionFactory
 import org.springframework.data.redis.core.RedisTemplate
 import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer
 import org.springframework.data.redis.serializer.StringRedisSerializer
-import com.fasterxml.jackson.databind.ObjectMapper
 
 @Configuration
 @ConditionalOnClass(RedisTemplate::class, ObjectMapper::class)
 @ConditionalOnProperty(prefix = "cacheflow", name = ["storage"], havingValue = "REDIS")
 class CacheFlowRedisConfiguration {
-
     @Bean
     @ConditionalOnMissingBean(name = ["cacheFlowRedisTemplate"])
     fun cacheFlowRedisTemplate(connectionFactory: RedisConnectionFactory): RedisTemplate<String, Any> {
@@ -28,4 +27,47 @@ class CacheFlowRedisConfiguration {
         template.afterPropertiesSet()
         return template
     }
+
+    @Bean
+    @ConditionalOnMissingBean
+    fun redisCacheInvalidator(
+        properties: io.cacheflow.spring.config.CacheFlowProperties,
+        redisTemplate: org.springframework.data.redis.core.StringRedisTemplate,
+        @org.springframework.context.annotation.Lazy cacheFlowService: io.cacheflow.spring.service.CacheFlowService,
+        objectMapper: ObjectMapper,
+    ): io.cacheflow.spring.messaging.RedisCacheInvalidator =
+        io.cacheflow.spring.messaging.RedisCacheInvalidator(
+            properties,
+            redisTemplate,
+            cacheFlowService,
+            objectMapper,
+        )
+
+    @Bean
+    @ConditionalOnMissingBean
+    fun cacheInvalidationListenerAdapter(
+        redisCacheInvalidator: io.cacheflow.spring.messaging.RedisCacheInvalidator,
+    ): org.springframework.data.redis.listener.adapter.MessageListenerAdapter =
+        org.springframework.data.redis.listener.adapter.MessageListenerAdapter(
+            redisCacheInvalidator,
+            "handleMessage",
+        )
+
+    @Bean
+    @ConditionalOnMissingBean
+    fun redisMessageListenerContainer(
+        connectionFactory: RedisConnectionFactory,
+        cacheInvalidationListenerAdapter: org.springframework.data.redis.listener.adapter.MessageListenerAdapter,
+    ): org.springframework.data.redis.listener.RedisMessageListenerContainer {
+        val container =
+            org.springframework.data.redis.listener
+                .RedisMessageListenerContainer()
+        container.setConnectionFactory(connectionFactory)
+        container.addMessageListener(
+            cacheInvalidationListenerAdapter,
+            org.springframework.data.redis.listener
+                .ChannelTopic("cacheflow:invalidation"),
+        )
+        return container
+    }
 }

src/main/kotlin/io/cacheflow/spring/autoconfigure/CacheFlowWarmingConfiguration.kt

@@ -0,0 +1,20 @@
+package io.cacheflow.spring.autoconfigure
+
+import io.cacheflow.spring.config.CacheFlowProperties
+import io.cacheflow.spring.warming.CacheWarmer
+import io.cacheflow.spring.warming.CacheWarmupProvider
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+
+@Configuration
+@ConditionalOnProperty(prefix = "cacheflow.warming", name = ["enabled"], havingValue = "true", matchIfMissing = true)
+class CacheFlowWarmingConfiguration {
+    @Bean
+    @ConditionalOnMissingBean
+    fun cacheWarmer(
+        properties: CacheFlowProperties,
+        warmupProviders: List<CacheWarmupProvider>,
+    ): CacheWarmer = CacheWarmer(properties, warmupProviders)
+}

src/main/kotlin/io/cacheflow/spring/config/CacheFlowProperties.kt

@@ -29,6 +29,7 @@ data class CacheFlowProperties(
     val awsCloudFront: AwsCloudFrontProperties = AwsCloudFrontProperties(),
     val fastly: FastlyProperties = FastlyProperties(),
     val metrics: MetricsProperties = MetricsProperties(),
+    val warming: WarmingProperties = WarmingProperties(),
     val baseUrl: String = "https://yourdomain.com",
 ) {
     /**
@@ -163,4 +164,13 @@ data class CacheFlowProperties(
         val enabled: Boolean = true,
         val exportInterval: Long = 60,
     )
+
+    /**
+     * Cache warming configuration.
+     *
+     * @property enabled Whether cache warming is enabled
+     */
+    data class WarmingProperties(
+        val enabled: Boolean = true,
+    )
 }