From b5990a55ac0a80cafa5cf96fe46bacfdd2b5ba9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=94=D0=BC=D0=B8=D1=82=D1=80=D0=B8=D0=B9=20=D0=A1=D0=BE?= =?UTF-8?q?=D0=BB=D0=BE=D0=B2=D1=8C=D0=B5=D0=B2?= Date: Mon, 25 May 2026 15:05:04 +0300 Subject: [PATCH] =?UTF-8?q?=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2=D0=B8=D1=82?= =?UTF-8?q?=D1=8C=20=D0=BC=D0=B0=D1=81=D1=88=D1=82=D0=B0=D0=B1=D0=B8=D1=80?= =?UTF-8?q?=D1=83=D0=B5=D0=BC=D1=8B=D0=B9=20outbox=20worker=20=D0=B4=D0=BB?= =?UTF-8?q?=D1=8F=20=D0=BF=D0=B0=D1=81=D0=BF=D0=BE=D1=80=D1=82=D0=BE=D0=B2?= =?UTF-8?q?=20=D0=BC=D0=B0=D1=80=D1=88=D1=80=D1=83=D1=82=D0=BE=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - добавить properties и config values для route passport outbox worker - включить Spring scheduling в route-processing service - добавить транзакционный сервис claim и status transitions для outbox - публиковать claimed outbox rows через существующий Kafka publisher - обрабатывать publish failures по каждой row с bounded exponential backoff - обеспечить безопасность при нескольких pod через FOR UPDATE SKIP LOCKED и lockedUntil - добавить unit tests для claim, publish transitions, retry/fail behavior и batch-обработки worker --- .../config/RoutePassportOutboxProperties.kt | 44 ++++ .../service/RoutePassportOutboxService.kt | 107 ++++++++ .../service/RoutePassportOutboxWorker.kt | 100 +++++++ .../service/RoutePassportOutboxServiceTest.kt | 249 ++++++++++++++++++ .../service/RoutePassportOutboxWorkerTest.kt | 194 ++++++++++++++ 5 files changed, 694 insertions(+) create mode 100644 services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/RoutePassportOutboxProperties.kt create mode 100644 services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxService.kt create mode 100644 services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorker.kt create mode 100644 services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxServiceTest.kt create mode 100644 services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorkerTest.kt diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/RoutePassportOutboxProperties.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/RoutePassportOutboxProperties.kt new file mode 100644 index 0000000..879185b --- /dev/null +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/config/RoutePassportOutboxProperties.kt @@ -0,0 +1,44 @@ +package space.nstart.pcp.routepassportconsumer.config + +import jakarta.validation.constraints.AssertTrue +import jakarta.validation.constraints.Min +import org.springframework.boot.context.properties.ConfigurationProperties +import org.springframework.validation.annotation.Validated + +/** + * Operational settings for the route passport outbox worker. + * + * Values control scheduling cadence, DB claim size, lock lease time and bounded retry backoff. + */ +@Validated +@ConfigurationProperties(prefix = "app.route-passport.outbox") +data class RoutePassportOutboxProperties( + + /** Enables scheduled publication of pending route passport outbox rows. */ + val enabled: Boolean = true, + + /** Maximum number of due rows claimed by one worker tick. */ + @field:Min(1) + val batchSize: Int = 50, + + /** Delay between scheduled worker runs. */ + @field:Min(100) + val fixedDelayMs: Long = 1_000, + + /** Lease duration for a claimed row; expired locks can be claimed by another instance. */ + @field:Min(1_000) + val lockTimeoutMs: Long = 60_000, + + /** Backoff after the first failed publication attempt. */ + @field:Min(100) + val initialBackoffMs: Long = 1_000, + + /** Maximum backoff between publication attempts. */ + @field:Min(100) + val maxBackoffMs: Long = 60_000, +) { + + @get:AssertTrue(message = "maxBackoffMs must be greater than or equal to initialBackoffMs") + val isMaxBackoffGreaterThanOrEqualToInitialBackoff: Boolean + get() = maxBackoffMs >= initialBackoffMs +} diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxService.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxService.kt new file mode 100644 index 0000000..2a7d20b --- /dev/null +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxService.kt @@ -0,0 +1,107 @@ +package space.nstart.pcp.routepassportconsumer.service + +import org.springframework.stereotype.Service +import org.springframework.transaction.annotation.Transactional +import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus +import space.nstart.pcp.routepassportconsumer.repository.RoutePassportOutboxRepository +import java.time.Duration +import java.time.LocalDateTime +import java.util.UUID + +/** + * Owns transactional claim and status transitions for route passport outbox rows. + */ +@Service +class RoutePassportOutboxService( + private val routePassportOutboxRepository: RoutePassportOutboxRepository, + private val outboxProperties: RoutePassportOutboxProperties, +) { + + /** + * Claims due pending rows in one DB transaction using SELECT FOR UPDATE SKIP LOCKED. + */ + @Transactional + fun claimDueBatch(): List { + if (!outboxProperties.enabled) { + return emptyList() + } + + val now = LocalDateTime.now() + val lockedUntil = now.plus(Duration.ofMillis(outboxProperties.lockTimeoutMs)) + val claimedRows = routePassportOutboxRepository.findDueForUpdateSkipLocked(outboxProperties.batchSize) + + claimedRows.forEach { outbox -> + outbox.lockedUntil = lockedUntil + outbox.updatedAt = now + } + + return routePassportOutboxRepository.saveAll(claimedRows).toList() + } + + /** + * Marks a successfully published row as terminally published. + */ + @Transactional + fun markPublished(outboxId: UUID) { + val now = LocalDateTime.now() + val outbox = findOutbox(outboxId) + + outbox.status = RoutePassportOutboxStatus.PUBLISHED + outbox.publishedAt = now + outbox.lockedUntil = null + outbox.lastError = null + outbox.updatedAt = now + + routePassportOutboxRepository.save(outbox) + } + + /** + * Records a publish failure and either schedules bounded retry or marks the row as failed. + */ + @Transactional + fun markRetryOrFailed(outboxId: UUID, exception: Exception) { + val now = LocalDateTime.now() + val outbox = findOutbox(outboxId) + + outbox.attempts += 1 + outbox.lastError = compactError(exception) + outbox.lockedUntil = null + outbox.updatedAt = now + + if (outbox.attempts >= outbox.maxAttempts) { + outbox.status = RoutePassportOutboxStatus.FAILED + } else { + outbox.status = RoutePassportOutboxStatus.PENDING + outbox.nextAttemptAt = now.plus(Duration.ofMillis(calculateBackoffMs(outbox.attempts))) + } + + routePassportOutboxRepository.save(outbox) + } + + private fun findOutbox(outboxId: UUID): RoutePassportOutboxEntity = + routePassportOutboxRepository.findById(outboxId) + .orElseThrow { IllegalStateException("Route passport outbox row not found: $outboxId") } + + private fun calculateBackoffMs(attempts: Int): Long { + var backoffMs = outboxProperties.initialBackoffMs.coerceAtMost(outboxProperties.maxBackoffMs) + + repeat((attempts - 1).coerceAtLeast(0)) { + backoffMs = if (backoffMs >= outboxProperties.maxBackoffMs / 2) { + outboxProperties.maxBackoffMs + } else { + (backoffMs * 2).coerceAtMost(outboxProperties.maxBackoffMs) + } + } + + return backoffMs + } + + private fun compactError(exception: Exception): String = + "${exception::class.java.name}: ${exception.message ?: ""}".take(MAX_ERROR_LENGTH) + + private companion object { + const val MAX_ERROR_LENGTH = 2_000 + } +} diff --git a/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorker.kt b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorker.kt new file mode 100644 index 0000000..58721b6 --- /dev/null +++ b/services/pcp-route-processing-service/src/main/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorker.kt @@ -0,0 +1,100 @@ +package space.nstart.pcp.routepassportconsumer.service + +import org.slf4j.LoggerFactory +import org.springframework.scheduling.annotation.Scheduled +import org.springframework.stereotype.Component +import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus + +/** + * Scheduled worker that publishes already persisted route passport outbox rows. + */ +@Component +class RoutePassportOutboxWorker( + private val routePassportOutboxService: RoutePassportOutboxService, + private val routePassportKafkaPublisherService: RoutePassportKafkaPublisherService, + private val outboxProperties: RoutePassportOutboxProperties, +) { + private val log = LoggerFactory.getLogger(this::class.java) + + @Scheduled(fixedDelayString = "\${app.route-passport.outbox.fixed-delay-ms:1000}") + fun publishDueOutboxRows() { + if (!outboxProperties.enabled) { + return + } + + val outboxRows = routePassportOutboxService.claimDueBatch() + outboxRows.forEach { outbox -> + logClaimed(outbox) + } + + outboxRows.forEach { outbox -> + try { + publishOne(outbox) + } catch (exception: Exception) { + log.error( + "Kafka outbox row processing failed: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}, exceptionClass={}, exceptionMessage={}", + outbox.id, + outbox.routeId, + outbox.eventType, + outbox.topic, + outbox.attempts, + outbox.status, + exception::class.java.name, + exception.message, + ) + } + } + } + + private fun publishOne(outbox: RoutePassportOutboxEntity) { + try { + routePassportKafkaPublisherService.publish(outbox) + routePassportOutboxService.markPublished(outbox.id) + + log.info( + "Kafka outbox row marked published: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}", + outbox.id, + outbox.routeId, + outbox.eventType, + outbox.topic, + outbox.attempts, + RoutePassportOutboxStatus.PUBLISHED, + ) + } catch (exception: Exception) { + routePassportOutboxService.markRetryOrFailed(outbox.id, exception) + + val attemptsAfterFailure = outbox.attempts + 1 + val statusAfterFailure = if (attemptsAfterFailure >= outbox.maxAttempts) { + RoutePassportOutboxStatus.FAILED + } else { + RoutePassportOutboxStatus.PENDING + } + + log.warn( + "Kafka outbox row publish failed: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}, exceptionClass={}, exceptionMessage={}", + outbox.id, + outbox.routeId, + outbox.eventType, + outbox.topic, + attemptsAfterFailure, + statusAfterFailure, + exception::class.java.name, + exception.message, + ) + } + } + + private fun logClaimed(outbox: RoutePassportOutboxEntity) { + log.info( + "Kafka outbox row claimed: outboxId={}, routeId={}, eventType={}, topic={}, attempts={}, status={}", + outbox.id, + outbox.routeId, + outbox.eventType, + outbox.topic, + outbox.attempts, + outbox.status, + ) + } +} diff --git a/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxServiceTest.kt b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxServiceTest.kt new file mode 100644 index 0000000..20bb5cc --- /dev/null +++ b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxServiceTest.kt @@ -0,0 +1,249 @@ +package space.nstart.pcp.routepassportconsumer.service + +import org.junit.jupiter.api.Test +import org.mockito.ArgumentMatchers.any +import org.mockito.ArgumentMatchers.anyInt +import org.mockito.Mockito.doAnswer +import org.mockito.Mockito.doReturn +import org.mockito.Mockito.mock +import org.mockito.Mockito.never +import org.mockito.Mockito.verify +import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent +import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus +import space.nstart.pcp.routepassportconsumer.repository.RoutePassportOutboxRepository +import tools.jackson.databind.ObjectMapper +import tools.jackson.databind.json.JsonMapper +import tools.jackson.module.kotlin.KotlinModule +import java.time.Duration +import java.time.LocalDateTime +import java.util.Optional +import java.util.UUID +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +class RoutePassportOutboxServiceTest { + + private val objectMapper: ObjectMapper = JsonMapper.builder() + .addModule(KotlinModule.Builder().build()) + .build() + + @Test + fun `claimDueBatch returns empty when outbox is disabled`() { + val repository = mock(RoutePassportOutboxRepository::class.java) + val service = service( + repository = repository, + properties = properties(enabled = false), + ) + + val claimedRows = service.claimDueBatch() + + assertTrue(claimedRows.isEmpty()) + verify(repository, never()).findDueForUpdateSkipLocked(anyInt()) + } + + @Test + fun `claimDueBatch locks returned rows`() { + val repository = mock(RoutePassportOutboxRepository::class.java) + val service = service(repository) + val first = outbox(id = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62")) + val second = outbox(id = UUID.fromString("5d75d482-e402-46d4-8e2b-4c19bff6f795")) + val originalUpdatedAt = first.updatedAt + + doReturn(listOf(first, second)) + .`when`(repository) + .findDueForUpdateSkipLocked(50) + stubSaveAll(repository) + + val claimedRows = service.claimDueBatch() + + assertEquals(listOf(first, second), claimedRows) + claimedRows.forEach { outbox -> + assertNotNull(outbox.lockedUntil) + assertEquals(60_000, Duration.between(outbox.updatedAt, outbox.lockedUntil).toMillis()) + assertTrue(outbox.updatedAt != originalUpdatedAt) + } + verify(repository).saveAll(listOf(first, second)) + } + + @Test + fun `markPublished moves row to published state`() { + val repository = mock(RoutePassportOutboxRepository::class.java) + val service = service(repository) + val outbox = outbox( + lockedUntil = LocalDateTime.of(2026, 4, 1, 10, 20), + lastError = "previous failure", + ) + + stubFindById(repository, outbox) + stubSave(repository) + + service.markPublished(outbox.id) + + assertEquals(RoutePassportOutboxStatus.PUBLISHED, outbox.status) + assertNotNull(outbox.publishedAt) + assertNull(outbox.lockedUntil) + assertNull(outbox.lastError) + verify(repository).save(outbox) + } + + @Test + fun `markRetryOrFailed below max attempts schedules pending retry`() { + val repository = mock(RoutePassportOutboxRepository::class.java) + val service = service(repository) + val outbox = outbox( + attempts = 0, + maxAttempts = 3, + lockedUntil = LocalDateTime.of(2026, 4, 1, 10, 20), + ) + val exception = IllegalStateException("broker unavailable") + + stubFindById(repository, outbox) + stubSave(repository) + + service.markRetryOrFailed(outbox.id, exception) + + assertEquals(1, outbox.attempts) + assertEquals(RoutePassportOutboxStatus.PENDING, outbox.status) + assertNull(outbox.lockedUntil) + assertTrue(outbox.lastError?.contains(IllegalStateException::class.java.name) == true) + assertTrue(outbox.lastError?.contains("broker unavailable") == true) + assertEquals(1_000, Duration.between(outbox.updatedAt, outbox.nextAttemptAt).toMillis()) + verify(repository).save(outbox) + } + + @Test + fun `markRetryOrFailed at max attempts moves row to failed state`() { + val repository = mock(RoutePassportOutboxRepository::class.java) + val service = service(repository) + val outbox = outbox( + attempts = 2, + maxAttempts = 3, + lockedUntil = LocalDateTime.of(2026, 4, 1, 10, 20), + ) + + stubFindById(repository, outbox) + stubSave(repository) + + service.markRetryOrFailed(outbox.id, IllegalStateException("broker unavailable")) + + assertEquals(3, outbox.attempts) + assertEquals(RoutePassportOutboxStatus.FAILED, outbox.status) + assertNull(outbox.lockedUntil) + assertTrue(outbox.lastError?.contains("broker unavailable") == true) + verify(repository).save(outbox) + } + + @Test + fun `markRetryOrFailed caps exponential backoff by maxBackoffMs`() { + val repository = mock(RoutePassportOutboxRepository::class.java) + val service = service( + repository = repository, + properties = properties( + initialBackoffMs = 1_000, + maxBackoffMs = 1_500, + ), + ) + val outbox = outbox( + attempts = 2, + maxAttempts = 10, + lockedUntil = LocalDateTime.of(2026, 4, 1, 10, 20), + ) + + stubFindById(repository, outbox) + stubSave(repository) + + service.markRetryOrFailed(outbox.id, IllegalStateException("broker unavailable")) + + assertEquals(3, outbox.attempts) + assertEquals(RoutePassportOutboxStatus.PENDING, outbox.status) + assertEquals(1_500, Duration.between(outbox.updatedAt, outbox.nextAttemptAt).toMillis()) + } + + private fun service( + repository: RoutePassportOutboxRepository, + properties: RoutePassportOutboxProperties = properties(), + ): RoutePassportOutboxService = + RoutePassportOutboxService( + routePassportOutboxRepository = repository, + outboxProperties = properties, + ) + + private fun properties( + enabled: Boolean = true, + initialBackoffMs: Long = 1_000, + maxBackoffMs: Long = 60_000, + ): RoutePassportOutboxProperties = + RoutePassportOutboxProperties( + enabled = enabled, + batchSize = 50, + fixedDelayMs = 1_000, + lockTimeoutMs = 60_000, + initialBackoffMs = initialBackoffMs, + maxBackoffMs = maxBackoffMs, + ) + + private fun stubFindById( + repository: RoutePassportOutboxRepository, + outbox: RoutePassportOutboxEntity, + ) { + doReturn(Optional.of(outbox)) + .`when`(repository) + .findById(outbox.id) + } + + private fun stubSave(repository: RoutePassportOutboxRepository) { + doAnswer { invocation -> + @Suppress("UNCHECKED_CAST") + invocation.arguments[0] as RoutePassportOutboxEntity + } + .`when`(repository) + .save(anyOutboxEntity()) + } + + private fun stubSaveAll(repository: RoutePassportOutboxRepository) { + doAnswer { invocation -> + @Suppress("UNCHECKED_CAST") + invocation.arguments[0] as Iterable + }.`when`(repository).saveAll(anyOutboxIterable()) + } + + private fun outbox( + id: UUID = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62"), + attempts: Int = 0, + maxAttempts: Int = 10, + lockedUntil: LocalDateTime? = null, + lastError: String? = null, + ): RoutePassportOutboxEntity { + val routeId = UUID.fromString("51b07d12-1823-4bd5-b55d-de02a5304c03") + val now = LocalDateTime.of(2026, 4, 1, 10, 15, 0) + + return RoutePassportOutboxEntity( + id = id, + routeId = routeId, + eventType = PcpKafkaEvent.ModeStatusChangedEvent.name, + topic = "pcp.route.in.v1", + messageKey = routeId.toString(), + payload = objectMapper.readTree("""{"type":"ModeStatusChangedEvent","data":{"routeNameFull":"ROUTE-001"}}"""), + headers = null, + status = RoutePassportOutboxStatus.PENDING, + attempts = attempts, + maxAttempts = maxAttempts, + nextAttemptAt = now, + lockedUntil = lockedUntil, + lastError = lastError, + createdAt = now, + updatedAt = now, + ) + } + + private fun anyOutboxEntity(): RoutePassportOutboxEntity = + any(RoutePassportOutboxEntity::class.java) ?: outbox() + + @Suppress("UNCHECKED_CAST") + private fun anyOutboxIterable(): Iterable = + (any(Iterable::class.java) as Iterable?) ?: emptyList() +} diff --git a/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorkerTest.kt b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorkerTest.kt new file mode 100644 index 0000000..5546788 --- /dev/null +++ b/services/pcp-route-processing-service/src/test/kotlin/space/nstart/pcp/routepassportconsumer/service/RoutePassportOutboxWorkerTest.kt @@ -0,0 +1,194 @@ +package space.nstart.pcp.routepassportconsumer.service + +import org.junit.jupiter.api.Test +import org.mockito.Mockito.doReturn +import org.mockito.Mockito.doThrow +import org.mockito.Mockito.mock +import org.mockito.Mockito.never +import org.mockito.Mockito.verify +import space.nstart.pcp.pcp_types_lib.message.PcpKafkaEvent +import space.nstart.pcp.routepassportconsumer.config.RoutePassportOutboxProperties +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxEntity +import space.nstart.pcp.routepassportconsumer.entity.RoutePassportOutboxStatus +import tools.jackson.databind.ObjectMapper +import tools.jackson.databind.json.JsonMapper +import tools.jackson.module.kotlin.KotlinModule +import java.time.LocalDateTime +import java.util.UUID + +class RoutePassportOutboxWorkerTest { + + private val objectMapper: ObjectMapper = JsonMapper.builder() + .addModule(KotlinModule.Builder().build()) + .build() + + @Test + fun `when disabled no claim or publish happens`() { + val outboxService = mock(RoutePassportOutboxService::class.java) + val publisherService = mock(RoutePassportKafkaPublisherService::class.java) + val worker = worker( + outboxService = outboxService, + publisherService = publisherService, + properties = properties(enabled = false), + ) + + worker.publishDueOutboxRows() + + verify(outboxService, never()).claimDueBatch() + verify(publisherService, never()).publish(anyOutboxEntity()) + } + + @Test + fun `when claim returns rows publisher is called for each row`() { + val outboxService = mock(RoutePassportOutboxService::class.java) + val publisherService = mock(RoutePassportKafkaPublisherService::class.java) + val first = outbox(id = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62")) + val second = outbox(id = UUID.fromString("5d75d482-e402-46d4-8e2b-4c19bff6f795")) + val worker = worker(outboxService, publisherService) + + doReturn(listOf(first, second)) + .`when`(outboxService) + .claimDueBatch() + + worker.publishDueOutboxRows() + + verify(publisherService).publish(first) + verify(publisherService).publish(second) + } + + @Test + fun `successful publish marks row as published`() { + val outboxService = mock(RoutePassportOutboxService::class.java) + val publisherService = mock(RoutePassportKafkaPublisherService::class.java) + val outbox = outbox() + val worker = worker(outboxService, publisherService) + + doReturn(listOf(outbox)) + .`when`(outboxService) + .claimDueBatch() + + worker.publishDueOutboxRows() + + verify(publisherService).publish(outbox) + verify(outboxService).markPublished(outbox.id) + } + + @Test + fun `publish failure marks row retry or failed`() { + val outboxService = mock(RoutePassportOutboxService::class.java) + val publisherService = mock(RoutePassportKafkaPublisherService::class.java) + val outbox = outbox() + val failure = IllegalStateException("broker unavailable") + val worker = worker(outboxService, publisherService) + + doReturn(listOf(outbox)) + .`when`(outboxService) + .claimDueBatch() + doThrow(failure) + .`when`(publisherService) + .publish(outbox) + + worker.publishDueOutboxRows() + + verify(outboxService).markRetryOrFailed(outbox.id, failure) + verify(outboxService, never()).markPublished(outbox.id) + } + + @Test + fun `failure of one row does not prevent processing next row`() { + val outboxService = mock(RoutePassportOutboxService::class.java) + val publisherService = mock(RoutePassportKafkaPublisherService::class.java) + val first = outbox(id = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62")) + val second = outbox(id = UUID.fromString("5d75d482-e402-46d4-8e2b-4c19bff6f795")) + val failure = IllegalStateException("broker unavailable") + val worker = worker(outboxService, publisherService) + + doReturn(listOf(first, second)) + .`when`(outboxService) + .claimDueBatch() + doThrow(failure) + .`when`(publisherService) + .publish(first) + + worker.publishDueOutboxRows() + + verify(outboxService).markRetryOrFailed(first.id, failure) + verify(publisherService).publish(second) + verify(outboxService).markPublished(second.id) + } + + @Test + fun `row status transition failure does not prevent processing next row`() { + val outboxService = mock(RoutePassportOutboxService::class.java) + val publisherService = mock(RoutePassportKafkaPublisherService::class.java) + val first = outbox(id = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62")) + val second = outbox(id = UUID.fromString("5d75d482-e402-46d4-8e2b-4c19bff6f795")) + val publishFailure = IllegalStateException("broker unavailable") + val transitionFailure = IllegalStateException("database unavailable") + val worker = worker(outboxService, publisherService) + + doReturn(listOf(first, second)) + .`when`(outboxService) + .claimDueBatch() + doThrow(publishFailure) + .`when`(publisherService) + .publish(first) + doThrow(transitionFailure) + .`when`(outboxService) + .markRetryOrFailed(first.id, publishFailure) + + worker.publishDueOutboxRows() + + verify(publisherService).publish(second) + verify(outboxService).markPublished(second.id) + } + + private fun worker( + outboxService: RoutePassportOutboxService, + publisherService: RoutePassportKafkaPublisherService, + properties: RoutePassportOutboxProperties = properties(), + ): RoutePassportOutboxWorker = + RoutePassportOutboxWorker( + routePassportOutboxService = outboxService, + routePassportKafkaPublisherService = publisherService, + outboxProperties = properties, + ) + + private fun properties(enabled: Boolean = true): RoutePassportOutboxProperties = + RoutePassportOutboxProperties( + enabled = enabled, + batchSize = 50, + fixedDelayMs = 1_000, + lockTimeoutMs = 60_000, + initialBackoffMs = 1_000, + maxBackoffMs = 60_000, + ) + + private fun outbox( + id: UUID = UUID.fromString("3ed5e75c-a5c0-4d63-88be-a701b5222b62"), + ): RoutePassportOutboxEntity { + val routeId = UUID.fromString("51b07d12-1823-4bd5-b55d-de02a5304c03") + val now = LocalDateTime.of(2026, 4, 1, 10, 15, 0) + + return RoutePassportOutboxEntity( + id = id, + routeId = routeId, + eventType = PcpKafkaEvent.ModeStatusChangedEvent.name, + topic = "pcp.route.in.v1", + messageKey = routeId.toString(), + payload = objectMapper.readTree("""{"type":"ModeStatusChangedEvent","data":{"routeNameFull":"ROUTE-001"}}"""), + headers = null, + status = RoutePassportOutboxStatus.PENDING, + attempts = 0, + maxAttempts = 10, + nextAttemptAt = now, + lockedUntil = now.plusMinutes(1), + lastError = null, + createdAt = now, + updatedAt = now, + ) + } + + private fun anyOutboxEntity(): RoutePassportOutboxEntity = + org.mockito.ArgumentMatchers.any(RoutePassportOutboxEntity::class.java) ?: outbox() +}