src/main/kotlin/no/iktdev/eventi/events/EventPollerImplementation.kt

@@ -8,30 +8,23 @@ import no.iktdev.eventi.stores.EventStore
 import java.time.Duration
 import java.time.Instant
 import java.util.UUID
+import kotlin.collections.iterator
 
 abstract class EventPollerImplementation(
     private val eventStore: EventStore,
     private val dispatchQueue: SequenceDispatchQueue,
     private val dispatcher: EventDispatcher
 ) {
-    private val log = KotlinLogging.logger {}
+    // Erstatter ikke lastSeenTime, men supplerer den
+    protected val refWatermark = mutableMapOf<UUID, Instant>()
 
-    /**
+    // lastSeenTime brukes kun som scan hint
-     * Per-reference watermark:
-     *  - first = last seen persistedAt
-     *  - second = last seen persistedId
-     */
-    protected val refWatermark = mutableMapOf<UUID, Pair<Instant, Long>>()
-
-    /**
-     * Global scan hint (timestamp only).
-     * Used to avoid scanning entire table every time.
-     */
     var lastSeenTime: Instant = Instant.EPOCH
 
     open var backoff = Duration.ofSeconds(2)
         protected set
     private val maxBackoff = Duration.ofMinutes(1)
+    private val log = KotlinLogging.logger {}
 
     open suspend fun start() {
         log.info { "EventPoller starting with initial backoff=$backoff" }
@@ -39,7 +32,7 @@ abstract class EventPollerImplementation(
             try {
                 pollOnce()
             } catch (e: Exception) {
-                log.error(e) { "Error in poller loop" }
+                e.printStackTrace()
                 delay(backoff.toMillis())
                 backoff = backoff.multipliedBy(2).coerceAtMost(maxBackoff)
             }
@@ -50,11 +43,11 @@ abstract class EventPollerImplementation(
         val pollStartedAt = MyTime.utcNow()
         log.debug { "🔍 Polling for new events" }
 
-        // Determine global scan start
+        // Global scan hint: kombiner refWatermark og lastSeenTime
-        val minRefTs = refWatermark.values.minOfOrNull { it.first }
+        val watermarkMin = refWatermark.values.minOrNull()
-        val scanFrom = when (minRefTs) {
+        val scanFrom = when (watermarkMin) {
             null -> lastSeenTime
-            else -> maxOf(lastSeenTime, minRefTs)
+            else -> maxOf(lastSeenTime, watermarkMin)
         }
 
         val newPersisted = eventStore.getPersistedEventsAfter(scanFrom)
@@ -66,73 +59,76 @@ abstract class EventPollerImplementation(
             return
         }
 
-        // Reset backoff
+        // Vi har sett nye events globalt – reset backoff
         backoff = Duration.ofSeconds(2)
         log.debug { "📬 Found ${newPersisted.size} new events after $scanFrom" }
 
         val grouped = newPersisted.groupBy { it.referenceId }
         var anyProcessed = false
 
-        // Track highest persistedAt seen globally this round
+        // Track høyeste persistedAt vi har sett i denne runden
         val maxPersistedThisRound = newPersisted.maxOf { it.persistedAt }
 
         for ((ref, eventsForRef) in grouped) {
-            val (refSeenAt, refSeenId) = refWatermark[ref] ?: (Instant.EPOCH to 0L)
+            val refSeen = refWatermark[ref] ?: Instant.EPOCH
-
-            // Filter new events using (timestamp, id) ordering
-            val newForRef = eventsForRef.filter { ev ->
-                ev.persistedAt > refSeenAt ||
-                        (ev.persistedAt == refSeenAt && ev.id > refSeenId)
-            }
 
+            // Finn kun nye events for denne ref’en
+            val newForRef = eventsForRef.filter { it.persistedAt > refSeen }
             if (newForRef.isEmpty()) {
-                log.debug { "🧊 No new events for $ref since ($refSeenAt, id=$refSeenId)" }
+                log.debug { "🧊 No new events for $ref since $refSeen" }
                 continue
             }
 
-            // If ref is busy, skip dispatch
+            // Hvis ref er busy → ikke oppdater watermark, ikke dispatch
             if (dispatchQueue.isProcessing(ref)) {
                 log.debug { "⏳ $ref is busy — deferring ${newForRef.size} events" }
                 continue
             }
 
-            // Fetch full sequence for dispatch
+            // Hent full sekvens for ref (Eventi-invariant)
             val fullLog = eventStore.getPersistedEventsFor(ref)
             val events = fullLog.mapNotNull { it.toEvent() }
 
             log.debug { "🚀 Dispatching ${events.size} events for $ref" }
             dispatchQueue.dispatch(ref, events, dispatcher)
 
-            // Update watermark for this reference
+            // Oppdater watermark for denne ref’en
-            val maxEvent = newForRef.maxWith(
+            val maxPersistedAtForRef = newForRef.maxOf { it.persistedAt }
-                compareBy({ it.persistedAt }, { it.id })
+            val newWatermark = minOf(pollStartedAt, maxPersistedAtForRef).plusNanos(1)
-            )
 
-            val newWatermarkAt = minOf(pollStartedAt, maxEvent.persistedAt)
+            refWatermark[ref] = newWatermark
-            val newWatermarkId = maxEvent.id
-
-            refWatermark[ref] = newWatermarkAt to newWatermarkId
             anyProcessed = true
 
-            log.debug { "⏩ Updated watermark for $ref → ($newWatermarkAt, id=$newWatermarkId)" }
+            log.debug { "⏩ Updated watermark for $ref → $newWatermark" }
         }
 
-        // Update global scan hint
+        // Oppdater global scan hint uansett – vi har sett nye events
+        // Dette hindrer livelock når alle events er <= watermark for sine refs
         val newLastSeen = maxOf(
             lastSeenTime,
             maxPersistedThisRound.plusNanos(1)
         )
 
         if (anyProcessed) {
-            val minRef = refWatermark.values.minOfOrNull { it.first }
+            // Behold intensjonen din: globalt hint basert på laveste watermark,
-            lastSeenTime = when (minRef) {
+            // men aldri gå bakover i tid ift lastSeenTime
+            val minRefWatermark = refWatermark.values.minOrNull()
+            lastSeenTime = when (minRefWatermark) {
                 null -> newLastSeen
-                else -> maxOf(newLastSeen, minRef)
+                else -> maxOf(newLastSeen, minRefWatermark)
             }
             log.debug { "📉 Global scanFrom updated → $lastSeenTime (anyProcessed=true)" }
         } else {
+            // Ingen refs prosessert, men vi vet at alle events vi så er <= watermark
+            // → trygt å flytte lastSeenTime forbi dem
             lastSeenTime = newLastSeen
             log.debug { "🔁 No refs processed — advancing global scanFrom to $lastSeenTime" }
         }
     }
+
+
+
+
+
+
 }

src/main/kotlin/no/iktdev/eventi/models/Event.kt

@@ -4,7 +4,7 @@ import java.util.UUID
 
 @Suppress("UNCHECKED_CAST")
 abstract class Event {
-    lateinit var referenceId: UUID
+    var referenceId: UUID = UUID.randomUUID()
         protected set
     var eventId: UUID = UUID.randomUUID()
         private set

src/main/kotlin/no/iktdev/eventi/tasks/TaskListener.kt

@@ -68,7 +68,9 @@ abstract class TaskListener(val taskType: TaskType = TaskType.CPU_INTENSIVE): Ta
         currentJob = getDispatcherForTask(task).launch {
             try {
                 val result = onTask(task)
+                reporter.markCompleted(task.taskId)
                 onComplete(task, result)
+
             } catch (e: CancellationException) {
                 // Dette er en ekte kansellering
                 onCancelled(task)

src/test/kotlin/no/iktdev/eventi/EventDispatcherTest.kt

@@ -64,7 +64,7 @@ class EventDispatcherTest : TestBase() {
     fun shouldProduceOneEventAndStop() {
         ProducingListener()
 
-        val trigger = TriggerEvent().newReferenceId()
+        val trigger = TriggerEvent()
         dispatcher.dispatch(trigger.referenceId, listOf(trigger))
 
         val produced = eventStore.all().firstOrNull()
@@ -87,7 +87,7 @@ class EventDispatcherTest : TestBase() {
     fun shouldSkipAlreadyDerivedEvents() {
         ProducingListener()
 
-        val trigger = TriggerEvent().newReferenceId()
+        val trigger = TriggerEvent()
         val derived = DerivedEvent().derivedOf(trigger).toPersisted(1L, MyTime.utcNow())
 
         eventStore.persist(derived!!.toEvent()!!) // simulate prior production
@@ -108,8 +108,8 @@ class EventDispatcherTest : TestBase() {
     fun shouldPassFullContextToListener() {
         val listener = ContextCapturingListener()
 
-        val e1 = TriggerEvent().newReferenceId()
+        val e1 = TriggerEvent()
-        val e2 = OtherEvent().newReferenceId()
+        val e2 = OtherEvent()
         dispatcher.dispatch(e1.referenceId, listOf(e1, e2))
 
         assertEquals(2, listener.context.size)
@@ -124,11 +124,9 @@ class EventDispatcherTest : TestBase() {
     """
     )
     fun shouldBehaveDeterministicallyAcrossReplays() {
-        val referenceId = UUID.randomUUID()
-
         ProducingListener()
 
-        val trigger = TriggerEvent().usingReferenceId(referenceId)
+        val trigger = TriggerEvent()
         dispatcher.dispatch(trigger.referenceId, listOf(trigger))
         val replayContext = listOf(trigger) + eventStore.all().mapNotNull { it.toEvent() }
 
@@ -146,8 +144,6 @@ class EventDispatcherTest : TestBase() {
     """
     )
     fun shouldNotDeliverDeletedEventsAsCandidates() {
-        val referenceId = UUID.randomUUID()
-
         val dispatcher = EventDispatcher(eventStore)
         val received = mutableListOf<Event>()
 
@@ -158,10 +154,11 @@ class EventDispatcherTest : TestBase() {
             }
         }
         // Original hendelse
-        val original = TriggerEvent().usingReferenceId(referenceId)
+        val original = TriggerEvent()
 
         // Slettehendelse som peker på original
-        val deleted = object : DeleteEvent(original.eventId) {}.apply { newReferenceId() }
+        val deleted = object : DeleteEvent(original.eventId) {
+        }
 
         // Dispatch med begge hendelser
         dispatcher.dispatch(original.referenceId, listOf(original, deleted))
@@ -187,8 +184,6 @@ class EventDispatcherTest : TestBase() {
     )
     fun shouldDeliverDeleteEventToListenersThatReactToIt() {
         val received = mutableListOf<Event>()
-        val referenceId = UUID.randomUUID()
-
         object : EventListener() {
             override fun onEvent(event: Event, history: List<Event>): Event? {
                 if (event is DeleteEvent) received += event
@@ -196,7 +191,7 @@ class EventDispatcherTest : TestBase() {
             }
         }
 
-        val deleted = object : DeleteEvent(UUID.randomUUID()) {}.apply { usingReferenceId(referenceId) }
+        val deleted = object : DeleteEvent(UUID.randomUUID()) {}
         dispatcher.dispatch(deleted.referenceId, listOf(deleted))
 
         assertTrue(received.contains(deleted))
@@ -213,7 +208,7 @@ class EventDispatcherTest : TestBase() {
     fun shouldNotRedeliverEventsThatHaveProducedDerivedEvents() {
         ProducingListener()
 
-        val trigger = TriggerEvent().newReferenceId()
+        val trigger = TriggerEvent()
         // Første dispatch: trigger produserer en DerivedEvent
         dispatcher.dispatch(trigger.referenceId, listOf(trigger))
 
@@ -243,8 +238,8 @@ class EventDispatcherTest : TestBase() {
     fun historyShouldExcludeDeletedEvents() {
         val dispatcher = EventDispatcher(eventStore)
 
-        val original = TriggerEvent().newReferenceId()
+        val original = TriggerEvent()
-        val deleted = object : DeleteEvent(original.eventId) {}.apply { usingReferenceId(original.referenceId) }
+        val deleted = object : DeleteEvent(original.eventId) {}
 
         var receivedHistory: List<Event> = emptyList()
 
@@ -271,9 +266,9 @@ class EventDispatcherTest : TestBase() {
     )
     fun historyShouldKeepNonDeletedEvents() {
         val dispatcher = EventDispatcher(eventStore)
-        val referenceId = UUID.randomUUID()
+
-        val e1 = TriggerEvent().usingReferenceId(referenceId)
+        val e1 = TriggerEvent()
-        val e2 = OtherEvent().usingReferenceId(referenceId)
+        val e2 = OtherEvent()
         val deleted = object : DeleteEvent(e1.eventId) {}
 
         var receivedHistory: List<Event> = emptyList()
@@ -303,8 +298,8 @@ class EventDispatcherTest : TestBase() {
     fun deleteEventShouldBeDeliveredButHistoryEmpty() {
         val dispatcher = EventDispatcher(eventStore)
 
-        val original = TriggerEvent().newReferenceId()
+        val original = TriggerEvent()
-        val deleted = object : DeleteEvent(original.eventId) {}.apply { newReferenceId() }
+        val deleted = object : DeleteEvent(original.eventId) {}
 
         var receivedEvent: Event? = null
         var receivedHistory: List<Event> = emptyList()

src/test/kotlin/no/iktdev/eventi/ZDSTest.kt

@@ -39,7 +39,7 @@ class ZDSTest {
     fun scenario1() {
         EventTypeRegistry.register(EchoEvent::class.java)
 
-        val echo = EchoEvent("hello").newReferenceId()
+        val echo = EchoEvent("hello")
         val persisted = echo.toPersisted(id = 1L)
 
         val restored = persisted!!.toEvent()

src/test/kotlin/no/iktdev/eventi/events/EventPollerImplementationTest.kt

@@ -214,7 +214,7 @@ class EventPollerImplementationTest : TestBase() {
             }
         }
 
-        val original = EchoEvent("Hello").newReferenceId()
+        val original = EchoEvent("Hello")
         eventStore.persist(original)
 
         poller.pollOnce()

src/test/kotlin/no/iktdev/eventi/events/poller/PollerStartLoopTest.kt

@@ -60,42 +60,6 @@ class PollerStartLoopTest : TestBase() {
         store.persistAt(e, time)
     }
 
-
-    @Test
-    @DisplayName("""
-    Når to events har identisk persistedAt
-    Hvis polleren kjører
-    Så skal begge events prosesseres og ingen mistes
-    """)
-    fun `poller handles same-timestamp events without losing any`() = runTest {
-        val ref = UUID.randomUUID()
-        val ts = Instant.parse("2025-01-01T12:00:00Z")
-
-        // Two events with same timestamp but different IDs
-        val e1 = TestEvent().withReference(ref).setMetadata(Metadata())
-        val e2 = TestEvent().withReference(ref).setMetadata(Metadata())
-
-        store.persistAt(e1, ts) // id=1
-        store.persistAt(e2, ts) // id=2
-
-        poller.startFor(iterations = 1)
-
-        // Verify dispatch happened
-        assertThat(dispatcher.dispatched).hasSize(1)
-
-        val (_, events) = dispatcher.dispatched.single()
-
-        // Both events must be present
-        assertThat(events.map { it.eventId })
-            .hasSize(2)
-            .doesNotHaveDuplicates()
-
-        // Watermark must reflect highest ID
-        val wm = poller.watermarkFor(ref)
-        assertThat(wm!!.first).isEqualTo(ts)
-        assertThat(wm.second).isEqualTo(2)
-    }
-
     @Test
     @DisplayName("""
     Når polleren kjører flere iterasjoner uten events
@@ -307,15 +271,11 @@ class PollerStartLoopTest : TestBase() {
 
         poller.startFor(iterations = 1)
 
-
-
         // A skal IKKE ha flyttet watermark
         assertThat(poller.watermarkFor(refA)).isEqualTo(wmA1)
 
-        // B skal ha flyttet watermark (på timestamp-nivå)
+        // B skal ha flyttet watermark
-        val wmB2 = poller.watermarkFor(refB)
+        assertThat(poller.watermarkFor(refB)).isGreaterThan(wmB1)
-        assertThat(wmB2!!.first).isGreaterThan(wmB1!!.first)
-
     }
 
     @DisplayName("🍌 Bananastesten™ — stress-test av watermark, busy refs og dispatch-semantikk")
@@ -473,7 +433,7 @@ class PollerStartLoopTest : TestBase() {
 
         // Sett watermark høyt (polleren setter watermark selv i ekte drift,
         // men i denne testen må vi simulere det)
-        poller.setWatermarkFor(ref, t(100), id = 999)
+        poller.setWatermarkFor(ref, t(100))
 
         // Sett lastSeenTime bak eventen
         poller.lastSeenTime = t(0)

src/test/kotlin/no/iktdev/eventi/events/poller/TestablePoller.kt

@@ -17,6 +17,8 @@ class TestablePoller(
     val scope: TestScope
 ) : EventPollerImplementation(eventStore, dispatchQueue, dispatcher), WatermarkDebugView {
 
+
+
     suspend fun startFor(iterations: Int) {
         repeat(iterations) {
             try {
@@ -30,17 +32,19 @@ class TestablePoller(
         }
     }
 
-    override fun watermarkFor(ref: UUID): Pair<Instant, Long>? {
+    override fun watermarkFor(ref: UUID): Instant? {
-        return refWatermark[ref]
+        return refWatermark[ref]?.let {
+            return it
+        }
     }
 
-    override fun setWatermarkFor(ref: UUID, time: Instant, id: Long) {
+    override fun setWatermarkFor(ref: UUID, time: Instant) {
-        refWatermark[ref] = time to id
+        refWatermark[ref] = time
     }
+
+
 }
-
 interface WatermarkDebugView {
-    fun watermarkFor(ref: UUID): Pair<Instant, Long>?
+    fun watermarkFor(ref: UUID): Instant?
-    fun setWatermarkFor(ref: UUID, time: Instant, id: Long)
+    fun setWatermarkFor(ref: UUID, time: Instant)
 }
-

src/test/kotlin/no/iktdev/eventi/tasks/TaskPollerImplementationTest.kt

@@ -114,7 +114,7 @@ class TaskPollerImplementationTest : TestBase() {
         val listener = EchoListener()
         val poller = object : TaskPollerImplementation(taskStore, reporterFactory) {}
 
-        val task = EchoTask("Hello").newReferenceId().derivedOf(object : Event() {}.apply { newReferenceId() })
+        val task = EchoTask("Hello").newReferenceId().derivedOf(object : Event() {})
         taskStore.persist(task)
 
         poller.pollOnce()