diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt
new file mode 100644
index 0000000..8598c46
--- /dev/null
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessor.kt
@@ -0,0 +1,212 @@
+package dev.adamko.kafkatorio.processor.misc
+
+import dev.adamko.kotka.extensions.component1
+import dev.adamko.kotka.extensions.component2
+import dev.adamko.kotka.extensions.state.allAsSequence
+import kotlin.coroutines.CoroutineContext
+import kotlin.time.Duration
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancelChildren
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.emitAll
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.runningFold
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.streams.KeyValue
+import org.apache.kafka.streams.processor.api.Processor
+import org.apache.kafka.streams.processor.api.ProcessorContext
+import org.apache.kafka.streams.processor.api.ProcessorSupplier
+import org.apache.kafka.streams.processor.api.Record
+import org.apache.kafka.streams.state.StoreBuilder
+import org.apache.kafka.streams.state.Stores
+import org.apache.kafka.streams.state.TimestampedKeyValueStore
+import org.apache.kafka.streams.state.ValueAndTimestamp
+
+
+class DebounceProcessor<K, V>(
+  private val inactivityDuration: Duration,
+  private val recordsTimestampedStoreName: String,
+  coroutineDispatcher: CoroutineDispatcher = Dispatchers.Default,
+) :
+  Processor<K, V, K, V>,
+  CoroutineScope {
+
+  override var coroutineContext: CoroutineContext =
+    SupervisorJob() +
+        coroutineDispatcher +
+        CoroutineName("DebounceProcessor")
+
+  private lateinit var kafkaContext: ProcessorContext<K, V>
+  private lateinit var recordsTimestampedStore: TimestampedKeyValueStore<K, V?>
+
+  private val recordsFlow: MutableSharedFlow<DebounceTaskInitializer<K>> =
+    MutableSharedFlow(
+      replay = 0,
+      extraBufferCapacity = Int.MAX_VALUE,
+      onBufferOverflow = BufferOverflow.DROP_OLDEST,
+    )
+
+  override fun init(context: ProcessorContext<K, V>) {
+    this.kafkaContext = context
+
+    this.coroutineContext += CoroutineName("DebounceProcessor-${context.applicationId()}-${context.taskId()}")
+
+    recordsTimestampedStore = kafkaContext.getStateStore(recordsTimestampedStoreName)
+
+    // start collecting records - create a debounce-timer for each record
+    recordsFlow
+      .runningFold(mapOf<K, Job>()) { acc, (k, createTask) ->
+        require(coroutineContext.isActive)
+
+        // cancel any previously existing debounce task
+        acc[k]?.cancel()
+
+        // add a new debounce task
+        when (createTask) {
+          true -> acc.plus(k to createDebounceTask(k))
+          false -> acc.minus(k)
+        }
+
+      }.launchIn(this)
+
+    // re-initialise the flow from previously-stored records
+    launch {
+      recordsFlow.emitAll(
+        recordsTimestampedStore
+          .allAsSequence()
+          .map { keyValue ->
+            require(coroutineContext.isActive)
+            DebounceTaskInitializer<K>(keyValue)
+          }
+          .asFlow()
+      )
+    }
+  }
+
+  private fun createDebounceTask(key: K): Job {
+    return launch {
+      delay(inactivityDuration)
+      val (value, timestamp) = recordsTimestampedStore.get(key)
+      if (coroutineContext.isActive) {
+        kafkaContext.forward(Record(key, value, timestamp))
+      }
+    }
+  }
+
+  override fun process(record: Record<K, V?>) {
+    launch {
+      recordsTimestampedStore.put(
+        record.key(),
+        ValueAndTimestamp.make(record.value(), record.timestamp())
+      )
+      recordsFlow.emit(DebounceTaskInitializer(record))
+    }
+  }
+
+  override fun close() {
+    coroutineContext.cancelChildren(CancellationException("processor go sleepy bye bye"))
+  }
+
+}
+
+internal data class DebounceTaskInitializer<K>(
+  val key: K,
+  val createTask: Boolean,
+) {
+  constructor(keyValue: KeyValue<K, *>) : this(keyValue.key, keyValue.value != null)
+  constructor(record: Record<K, *>) : this(record.key(), record.value() != null)
+}
+
+
+fun <K, V> debounceProcessor(
+  inactivityDuration: Duration,
+  storeName: String,
+  keySerde: Serde<K>,
+  valueSerde: Serde<V>,
+): ProcessorSupplier<K, V, K, V> =
+  object : ProcessorSupplier<K, V, K, V> {
+
+    override fun get(): Processor<K, V, K, V> = DebounceProcessor(inactivityDuration, storeName)
+
+    override fun stores(): Set<StoreBuilder<*>> = setOf(
+      Stores.timestampedKeyValueStoreBuilder(
+        Stores.persistentTimestampedKeyValueStore(storeName),
+        keySerde,
+        valueSerde,
+      )
+    )
+
+  }
+
+
+//class SuppressProcessor<K, V>(
+//  private val storeName: String,
+//  private val tickDuration: Duration = 1.seconds,
+//  private val debounceDuration: Duration,
+//) : Processor<K, V, K, V> {
+//
+//  private lateinit var context: ProcessorContext<K, V>
+//
+//  private lateinit var stateStore: TimestampedKeyValueStore<K, V?>
+//
+//  override fun init(context: ProcessorContext<K, V>) {
+//
+//    this.context = context
+//
+//    stateStore = context.getStateStore(storeName) as TimestampedKeyValueStore<K, V?>
+//
+//    context.schedule(
+//      tickDuration.toJavaDuration(),
+//      PunctuationType.WALL_CLOCK_TIME,
+//      ::punctuate,
+//    )
+//  }
+//
+//  override fun process(record: Record<K, V?>) {
+//
+//    val (key, newVal) = record
+//
+//    val storedVal: V? = stateStore[key]?.value()
+//
+//    val cachedVal = ValueAndTimestamp.make(
+//      storedVal ?: newVal,
+//      System.currentTimeMillis()
+//    )
+//
+//    stateStore.put(key, cachedVal)
+//
+////    val newTimestamp = if (storedVal == null) System.currentTimeMillis() else storedTimestamp
+////    stateStore.put(key, ValueAndTimestamp.make(storedVal, newTimestamp))
+//  }
+//
+//  private fun punctuate(currentTimestamp: Long) {
+//
+//    stateStore.all().use { iterator ->
+//      iterator.forEach { record: KeyValue<K, ValueAndTimestamp<V?>> ->
+//
+//        val (k, v, storedTimestamp) = record
+//
+//        if (currentTimestamp - storedTimestamp > debounceDuration.inWholeMilliseconds) {
+//
+//          context.forward(Record(k, v, storedTimestamp))
+//
+//          stateStore.delete(k)
+//        }
+//
+//      }
+//    }
+//  }
+//
+//}
+//
diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/SuppressProcessor.java b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/SuppressProcessor.java
new file mode 100644
index 0000000..06d3ec8
--- /dev/null
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/misc/SuppressProcessor.java
@@ -0,0 +1,89 @@
+package dev.adamko.kafkatorio.processor.misc;
+
+import java.time.Duration;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.state.TimestampedKeyValueStore;
+import org.apache.kafka.streams.state.ValueAndTimestamp;
+
+/**
+ * THIS PROCESSOR IS UNTESTED
+ * <br>
+ * This processor mirrors the source, but waits for an inactivity gap before forwarding records.
+ * <br>
+ * The suppression is key based. Newer values will replace previous values, and reset the inactivity
+ * gap.
+ */
+public class SuppressProcessor<K, V> implements Processor<K, V, K, V> {
+
+  private final String storeName;
+  private final Duration debounceCheckInterval;
+  private final long suppressTimeoutMillis;
+
+  private TimestampedKeyValueStore<K, V> stateStore;
+  private ProcessorContext<K, V> context;
+
+  /**
+   * @param storeName             The name of the {@link TimestampedKeyValueStore} which will hold
+   *                              records while they are being debounced.
+   * @param suppressTimeout       The duration of inactivity before records will be forwarded.
+   * @param debounceCheckInterval How regularly all records will be checked to see if they are
+   *                              eligible to be forwarded. The interval should be shorter than
+   *                              {@code suppressTimeout}.
+   */
+  public SuppressProcessor(
+      String storeName,
+      Duration suppressTimeout,
+      Duration debounceCheckInterval
+  ) {
+    this.storeName = storeName;
+    this.suppressTimeoutMillis = suppressTimeout.toMillis();
+    this.debounceCheckInterval = debounceCheckInterval;
+  }
+
+  @Override
+  public void init(ProcessorContext<K, V> context) {
+    this.context = context;
+
+    stateStore = context.getStateStore(storeName);
+
+    context.schedule(debounceCheckInterval, PunctuationType.WALL_CLOCK_TIME, this::punctuate);
+  }
+
+  @Override
+  public void process(Record<K, V> record) {
+
+    final var key = record.key();
+    final var value = record.value();
+
+    final var storedRecord = stateStore.get(key);
+
+    final var isNewRecord = storedRecord == null;
+
+    final var timestamp = isNewRecord ? System.currentTimeMillis() : storedRecord.timestamp();
+
+    stateStore.put(key, ValueAndTimestamp.make(value, timestamp));
+  }
+
+  private void punctuate(long timestamp) {
+    try (var iterator = stateStore.all()) {
+      while (iterator.hasNext()) {
+        KeyValue<K, ValueAndTimestamp<V>> storedRecord = iterator.next();
+        if (timestamp - storedRecord.value.timestamp() > suppressTimeoutMillis) {
+
+          final var record = new Record<>(
+              storedRecord.key,
+              storedRecord.value.value(),
+              storedRecord.value.timestamp()
+          );
+
+          context.forward(record);
+          stateStore.delete(storedRecord.key);
+        }
+      }
+    }
+  }
+}
diff --git a/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessorTest.kt b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessorTest.kt
new file mode 100644
index 0000000..b7ff622
--- /dev/null
+++ b/modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/processor/misc/DebounceProcessorTest.kt
@@ -0,0 +1,49 @@
+package dev.adamko.kafkatorio.processor.misc
+
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.core.spec.style.Test
+import io.mockk.every
+import io.mockk.mockk
+import kotlin.time.Duration.Companion.minutes
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.Topology
+import org.apache.kafka.streams.state.KeyValueIterator
+import org.apache.kafka.streams.state.TimestampedKeyValueStore
+
+class DebounceProcessorTest : FunSpec({
+
+
+}) {
+  @Test
+  fun blah() {
+
+
+    val topology = Topology()
+
+    topology.addProcessor(
+      "blah",
+      debounceProcessor(
+        1.minutes,
+        "asdasda",
+        Serdes.String(),
+        Serdes.String()
+      )
+    )
+
+
+//    val streamsBuilder = StreamsBuilder()
+//
+//    streamsBuilder
+//      .stream<String, String>("input-topic")
+//      .process(
+//        debounceProcessor(
+//          1.minutes,
+//          "input-topic-debounce-store",
+//          Serdes.String(),
+//          Serdes.String(),
+//        ),
+//      )
+
+  }
+}