fix(cohort): stop target retry storm and ledger duplicate-key failures

services/api/src/integrationTest/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortLedgerAutoflushIT.kt

@@ -0,0 +1,143 @@
+package net.blueshell.api.platform.integration.cohort.application
+
+import net.blueshell.api.platform.integration.cohort.persistence.Cohort
+import net.blueshell.api.platform.integration.cohort.persistence.CohortKind
+import net.blueshell.api.platform.integration.cohort.persistence.CohortMember
+import net.blueshell.api.platform.integration.cohort.persistence.CohortMemberState
+import net.blueshell.api.platform.integration.cohort.persistence.CohortSubject
+import net.blueshell.api.platform.integration.cohort.persistence.CohortSubjectType
+import net.blueshell.api.platform.integration.cohort.persistence.repository.CohortMemberRepository
+import net.blueshell.api.platform.integration.cohort.persistence.repository.CohortRepository
+import net.blueshell.api.platform.integration.cohort.persistence.repository.CohortSubjectRepository
+import net.blueshell.api.platform.integration.cohort.persistence.state
+import net.blueshell.api.platform.integration.mock.MockCohortPort
+import net.blueshell.api.platform.integration.sync.persistence.ExternalIdMapping
+import net.blueshell.api.platform.integration.sync.persistence.repository.ExternalIdMappingRepository
+import net.blueshell.api.shared.enums.Role
+import net.blueshell.api.shared.enums.TargetSystem
+import net.blueshell.api.testsupport.UserTestSupport
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatCode
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.springframework.beans.factory.annotation.Autowired
+import java.time.LocalDateTime
+
+class CohortLedgerAutoflushIT : UserTestSupport() {
+
+    @Autowired
+    private lateinit var cohorts: CohortRepository
+
+    @Autowired
+    private lateinit var subjects: CohortSubjectRepository
+
+    @Autowired
+    private lateinit var members: CohortMemberRepository
+
+    @Autowired
+    private lateinit var externalIds: ExternalIdMappingRepository
+
+    @Autowired
+    private lateinit var remediation: CohortRemediationService
+
+    @Autowired
+    private lateinit var mockCohortPort: MockCohortPort
+
+    @BeforeEach
+    fun resetCohortPort() {
+        mockCohortPort.clear()
+    }
+
+    @Test
+    fun `confirming desired row with matching stranger does not violate external unique key`() {
+        val user = createUserWithRole(Role.MEMBER)
+        val subject = newSubject()
+        val cohort = newCohort(subject, externalId = "list-99")
+        members.saveAndFlush(CohortMember(cohort = cohort, userId = user.id!!, subject = subject))
+        members.saveAndFlush(
+            CohortMember(
+                cohort = cohort,
+                userId = null,
+                subject = subject,
+                externalUserId = "ext-1",
+                verifiedAt = LocalDateTime.parse("2026-01-01T12:00:00"),
+                label = "old stranger",
+            ),
+        )
+        externalIds.saveAndFlush(ExternalIdMapping("USER", user.id!!, TargetSystem.BREVO.name, "ext-1"))
+        mockCohortPort.seedMember("ext-1", "list-99", "Ada Remote")
+
+        assertThatCode { remediation.verifyCohort(cohort.id!!) }.doesNotThrowAnyException()
+
+        val desired = members.findByCohortIdAndUserId(cohort.id!!, user.id!!)!!
+        assertThat(desired.externalUserId).isEqualTo("ext-1")
+        assertThat(desired.label).isEqualTo("Ada Remote")
+        assertThat(desired.state).isEqualTo(CohortMemberState.VERIFIED)
+        assertThat(members.findByCohortIdAndExternalUserIdAndUserIdIsNull(cohort.id!!, "ext-1")).isNull()
+    }
+
+    @Test
+    fun `rapid same-key delete re-add delete uses distinct soft-delete timestamps`() {
+        val subject = newSubject()
+        val cohort = newCohort(subject, externalId = "list-fast")
+        val first = members.saveAndFlush(
+            CohortMember(
+                cohort = cohort,
+                userId = null,
+                subject = subject,
+                externalUserId = "ext-fast",
+                verifiedAt = LocalDateTime.parse("2026-01-01T12:00:00"),
+            ),
+        )
+
+        assertThatCode {
+            members.delete(first)
+            members.flush()
+            val second = members.saveAndFlush(
+                CohortMember(
+                    cohort = cohort,
+                    userId = null,
+                    subject = subject,
+                    externalUserId = "ext-fast",
+                    verifiedAt = LocalDateTime.parse("2026-01-01T12:00:01"),
+                ),
+            )
+            members.delete(second)
+            members.flush()
+        }.doesNotThrowAnyException()
+
+        val deletedAtValues = entityManager.createNativeQuery(
+            """
+            SELECT DATE_FORMAT(deleted_at, '%Y-%m-%d %H:%i:%s.%f')
+            FROM cohort_member
+            WHERE cohort_id = :cohortId
+              AND external_user_id = :externalUserId
+              AND deleted_at <> '9999-12-31 23:59:59'
+            ORDER BY id
+            """.trimIndent(),
+        )
+            .setParameter("cohortId", cohort.id!!)
+            .setParameter("externalUserId", "ext-fast")
+            .resultList
+            .map { it.toString() }
+
+        assertThat(deletedAtValues)
+            .hasSize(2)
+            .allMatch { it.matches(Regex(""".*\.\d{6}$""")) }
+        assertThat(deletedAtValues.toSet()).hasSize(2)
+    }
+
+    private fun newSubject(): CohortSubject =
+        subjects.saveAndFlush(CohortSubject(type = CohortSubjectType.CUSTOM, label = "Members"))
+
+    private fun newCohort(subject: CohortSubject, externalId: String): Cohort =
+        cohorts.saveAndFlush(
+            Cohort(
+                system = TargetSystem.BREVO.name,
+                kind = CohortKind.LIST,
+                label = "Members",
+                subjectId = subject.id,
+                externalId = externalId,
+            ),
+        )
+}

services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/adapter/job/MaterializeCohortTargetJobHandler.kt

@@ -7,9 +7,9 @@ import org.springframework.stereotype.Component
 import tools.jackson.databind.ObjectMapper
 
 /**
- * Driving adapter: invokes [CohortTargeting.materialize]. Creates one
- * cohort's external target when it has none yet, so a per-member ADD that
- * found no target id can retry once it exists. Deduplicated per cohort.
+ * Driving adapter for stale `cohort.materialize-target` rows. The use case
+ * now returns an existing target id or fails terminally; it never creates a
+ * provider target.
  */
 @Component
 class MaterializeCohortTargetJobHandler(

services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/adapter/web/CohortController.kt

@@ -8,28 +8,31 @@ import net.blueshell.api.platform.integration.cohort.application.CohortQueryServ
 import net.blueshell.api.platform.integration.cohort.application.CohortSummary
 import net.blueshell.api.platform.integration.cohort.persistence.CohortFactKind
 import net.blueshell.api.platform.integration.cohort.persistence.CohortKind
+import net.blueshell.api.platform.integration.cohort.port.`in`.CohortRemediation
+import net.blueshell.api.platform.integration.cohort.port.`in`.CohortRepairResult
 import org.springframework.security.access.prepost.PreAuthorize
 import org.springframework.web.bind.annotation.GetMapping
 import org.springframework.web.bind.annotation.PathVariable
+import org.springframework.web.bind.annotation.PostMapping
 import org.springframework.web.bind.annotation.RequestMapping
 import org.springframework.web.bind.annotation.RestController
 import java.time.Instant
 
 /**
- * Read-only admin endpoints for cohort listings + the per-cohort
- * detail view (members + rules). Used today by the admin cohort
- * dashboard and the `CohortPicker.vue` component in the JobManager
- * trigger modal.
+ * Admin endpoints for cohort listings, per-cohort detail, and targeted
+ * repair actions. Used today by the admin cohort dashboard and the
+ * `CohortPicker.vue` component in the JobManager trigger modal.
  *
  * Full CRUD lands in a later PR; for now this is the picker- and
- * dashboard-side read.
+ * dashboard-side read plus operational repair.
  */
 @RestController
 @RequestMapping("/management/cohorts")
 @Tag(name = "Cohorts", description = "Admin cohort listings + detail")
 @PreAuthorize("hasAuthority('ADMIN')")
 class CohortController(
     private val cohortQueries: CohortQueryService,
+    private val remediation: CohortRemediation,
 ) {
     @GetMapping
     fun findCohorts(): List<CohortSummaryResponse> =
@@ -38,6 +41,10 @@ class CohortController(
     @GetMapping("/{id}")
     fun findCohortById(@PathVariable id: Long): CohortDetailResponse =
         cohortQueries.detail(id).toResponse()
+
+    @PostMapping("/{id}/repair-missing-adds")
+    fun repairMissingAdds(@PathVariable id: Long): CohortRepairResponse =
+        remediation.repairMissingAdds(id).toResponse()
 }
 
 @Schema(name = "CohortSummary")
@@ -90,6 +97,9 @@ data class CohortRuleResponse(
     val enabled: Boolean,
 )
 
+@Schema(name = "CohortRepair")
+data class CohortRepairResponse(val cohortId: Long, val enqueuedAdds: Int)
+
 private fun CohortSummary.toResponse(): CohortSummaryResponse =
     CohortSummaryResponse(
         id = cohort.id!!,
@@ -130,3 +140,6 @@ private fun CohortMemberRow.toResponse(): CohortMemberRowResponse =
         isUserDeleted = isUserDeleted,
         joinedAt = member.createdAt,
     )
+
+private fun CohortRepairResult.toResponse(): CohortRepairResponse =
+    CohortRepairResponse(cohortId = cohortId, enqueuedAdds = enqueuedAdds)

services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortMembershipSyncService.kt

@@ -10,7 +10,6 @@ import net.blueshell.api.platform.integration.cohort.port.out.CohortPortRegistry
 import net.blueshell.api.platform.integration.sync.application.ExternalIdMappingService
 import net.blueshell.api.platform.integration.sync.application.ExternalIdMappingService.Companion.USER_AGGREGATE
 import net.blueshell.api.shared.enums.TargetSystem
-import net.blueshell.api.shared.job.CohortJobs
 import net.blueshell.api.shared.job.ContactJobs
 import net.blueshell.api.shared.job.NonRetryableJobException
 import net.blueshell.api.shared.job.TrackedJobDispatcher
@@ -33,9 +32,8 @@ import java.time.LocalDateTime
  * - `ADD` with no user external id enqueues `SyncContact` and throws
  *   a retryable exception so the retry picks up after the contact
  *   has materialised externally.
- * - `ADD` with no cohort target id enqueues `cohort.materialize-target`
- *   and throws a retryable exception — it never creates the target
- *   itself, so two racing ADDs cannot create two remote targets.
+ * - `ADD` with no cohort target id fails terminally. An operator must
+ *   explicitly create or link a target before retrying the membership push.
  * - `REMOVE` with no external state on either side is a no-op —
  *   there is nothing to converge to.
  */
@@ -87,10 +85,7 @@ class CohortMembershipSyncService(
         }
         val externalCohortId = targetIds.find(cohort)
         if (externalCohortId == null) {
-            jobs.enqueue(CohortJobs.MaterializeCohortTarget, CohortJobs.MaterializeCohortTargetPayload(cohortId))
-            throw CohortMembershipNotReadyException(
-                "cohort $cohortId has no $system target — enqueued materialize-target, will retry",
-            )
+            throw CohortTargetNotLinkedException(cohortId, system)
         }
         outsideTransaction.executeWithoutResult { port.addMember(externalUserId, externalCohortId) }
 
@@ -131,3 +126,7 @@ class CohortMembershipSyncService(
  * the prerequisite job has had a chance to complete.
  */
 class CohortMembershipNotReadyException(message: String) : RuntimeException(message)
+
+class CohortTargetNotLinkedException(cohortId: Long, system: String) : NonRetryableJobException(
+    "cohort $cohortId has no $system target — create or link an external target, then retry the membership job",
+)