fix(cohort): make Brevo cohort membership sync additive-only

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

@@ -0,0 +1,15 @@
+package net.blueshell.api.platform.integration.cohort.application
+
+import net.blueshell.api.shared.enums.TargetSystem
+import org.springframework.stereotype.Component
+
+enum class CohortMemberRemovalOrigin {
+    AUTOMATIC_SYNC,
+    EXPLICIT_OPERATOR,
+}
+
+@Component
+class CohortMemberRemovalPolicy {
+    fun allows(system: TargetSystem, origin: CohortMemberRemovalOrigin): Boolean =
+        origin == CohortMemberRemovalOrigin.EXPLICIT_OPERATOR || system != TargetSystem.BREVO
+}

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

@@ -34,8 +34,9 @@ import java.time.LocalDateTime
  *   has materialised externally.
  * - `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.
+ * - `REMOVE` is a legacy automatic intent. For Brevo it is blocked before
+ *   calling the provider; explicit operator drift removal uses
+ *   [CohortRemediationService.removeExternalMember] instead.
  */
 @Service
 class CohortMembershipSyncService(
@@ -45,6 +46,7 @@ class CohortMembershipSyncService(
     private val externalIds: ExternalIdMappingService,
     private val targetIds: CohortTargetIds,
     private val jobs: TrackedJobDispatcher,
+    private val removalPolicy: CohortMemberRemovalPolicy,
     transactionManager: PlatformTransactionManager,
 ) : CohortMembershipSync {
 
@@ -109,6 +111,13 @@ class CohortMembershipSyncService(
             )
             return
         }
+        if (!removalPolicy.allows(TargetSystem.valueOf(system), CohortMemberRemovalOrigin.AUTOMATIC_SYNC)) {
+            log.info(
+                "Blocked automatic {} external removal for user {} / cohort {}; reconcile will surface any remote extra",
+                system, userId, cohortId,
+            )
+            return
+        }
         outsideTransaction.executeWithoutResult { port.removeMember(externalUserId, externalCohortId) }
         log.debug("Removed user {} from {} cohort {} (ext={})", userId, system, cohortId, externalCohortId)
     }

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

@@ -47,6 +47,7 @@ class CohortRemediationService(
     private val targetIds: CohortTargetIds,
     private val registry: CohortPortRegistry,
     private val jobs: TrackedJobDispatcher,
+    private val removalPolicy: CohortMemberRemovalPolicy,
     transactionManager: PlatformTransactionManager,
 ) : CohortRemediation {
 
@@ -79,7 +80,12 @@ class CohortRemediationService(
         val system = TargetSystem.valueOf(cohort.system)
         val externalCohortId = targetIds.require(cohort)
 
-        outsideTransaction.executeWithoutResult { registry.require(system).removeMember(externalUserId, externalCohortId) }
+        if (!removalPolicy.allows(system, CohortMemberRemovalOrigin.EXPLICIT_OPERATOR)) {
+            throw NonRetryableJobException("External member removal is not allowed for $system cohort $cohortId")
+        }
+        outsideTransaction.executeWithoutResult {
+            registry.require(system).removeMember(externalUserId, externalCohortId)
+        }
         ledger.removeStranger(cohortId, externalUserId)
     }
 

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

@@ -28,12 +28,13 @@ import org.springframework.transaction.annotation.Transactional
  * 3. Diffs against the user's current `cohort_member` rows.
  * 4. For each cohort to add: inserts a desired `CohortMember` row and
  *    enqueues a per-member `SyncCohortMembership(ADD)`.
- * 5. For each cohort to remove: soft-deletes the desired row and
- *    enqueues a per-member `SyncCohortMembership(REMOVE)`.
+ * 5. For each cohort to remove: soft-deletes the desired row locally.
  *
- * The per-member sync is the primary path to a healthy ledger: a
- * successful ADD stamps `synced_at` on the desired row. List reconcile
- * (`ReconcileList`) is a separate periodic/manual verifier, not enqueued
+ * The per-member ADD sync is the primary path to a healthy ledger: a
+ * successful ADD stamps `synced_at` on the desired row. Automatic
+ * removals are local-only; if the external list still contains the
+ * member, list reconcile records it as a stranger for operator review.
+ * `ReconcileList` is a separate periodic/manual verifier, not enqueued
  * here.
  *
  * All of the above runs in a single transaction. The job dispatch
@@ -86,10 +87,6 @@ class CohortRuleEvaluator(
         }
         evaluation.toRemove.forEach { cohortId ->
             removeExistingMembership(currentMemberships, cohortId)
-            jobs.enqueue(
-                CohortJobs.SyncCohortMembership,
-                CohortJobs.SyncCohortMembershipPayload(userId, cohortId, SyncCohortMembershipIntent.REMOVE),
-            )
         }
 
         log.info(

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

@@ -18,7 +18,9 @@ import org.springframework.transaction.annotation.Transactional
  * Funnels every change event that can alter a user's [UserFact]s into a
  * single re-evaluation of cohort membership. The evaluator writes the
  * desired `cohort_member` rows and enqueues per-member
- * `cohort.membership-sync` ADD/REMOVE jobs — it is not shadow mode.
+ * `cohort.membership-sync` ADD jobs for newly desired memberships. Rows
+ * that leave the desired set are soft-deleted locally and reviewed later
+ * through drift reconciliation.
  *
  * The engine still diffs cohort rows; the subject-level engine sketched
  * by V72 is a deferred follow-up (see [CohortSubject]).
@@ -49,8 +51,8 @@ class CohortRuleListener(
     @Transactional(propagation = Propagation.REQUIRES_NEW)
     fun onUserDeleted(evt: UserDeleted) {
         // After deletion the fact collector returns an empty set, so the
-        // diff is "remove from every cohort the user was in". Exactly the
-        // semantics we want.
+        // diff soft-deletes every desired cohort row the user was in without
+        // pushing external member removals.
         evaluator.evaluate(evt.userId)
     }
 

services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/in/CohortMembershipSync.kt

@@ -9,17 +9,17 @@ package net.blueshell.api.platform.integration.cohort.port.`in`
  * Use case shape: a single deterministic `sync(...)` call. The
  * implementation decides whether to enqueue a prerequisite user sync
  * and retry, whether a missing cohort target is terminal, and how
- * `REMOVE` interacts with missing external state. Callers do not branch
- * on these — they hand off the (userId, cohortId, intent) triple and
- * let the application layer route it.
+ * legacy `REMOVE` intents are guarded before provider access. Callers do
+ * not branch on these — they hand off the (userId, cohortId, intent)
+ * triple and let the application layer route it.
  */
 interface CohortMembershipSync {
     fun sync(userId: Long, cohortId: Long, intent: SyncCohortMembershipIntent)
 }
 
 /**
- * Direction of a single cohort-membership sync call. Lives on the
- * inbound port so callers and the implementation share one source
- * of truth for the verb.
+ * Direction of a single cohort-membership sync call. `REMOVE` remains for
+ * compatibility with old queued payloads; automatic Brevo removal is
+ * blocked by the application removal policy.
  */
 enum class SyncCohortMembershipIntent { ADD, REMOVE }

services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/in/CohortReconciliation.kt

@@ -14,8 +14,10 @@ interface CohortReconciliation {
     /**
      * Re-evaluates one user's cohort membership against the current
      * rules. Desired-row writes happen synchronously; external state
-     * converges through per-member `cohort.membership-sync` ADD/REMOVE
-     * jobs enqueued for each cohort the user joins or leaves.
+     * converges through per-member `cohort.membership-sync` ADD jobs
+     * enqueued for each cohort the user joins. Cohorts the user leaves are
+     * soft-deleted locally and surface as external extras during drift
+     * reconciliation.
      */
     fun evaluateUserCohorts(userId: Long)
 

services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/out/CohortPort.kt

@@ -8,7 +8,8 @@ import net.blueshell.api.shared.enums.TargetSystem
  * `cohort/adapter/<vendor>/`.
  *
  * The contract is operation-based rather than state-replace: callers
- * add or remove a single `(user, cohort)` pair by external id. External
+ * add a single `(user, cohort)` pair by external id, list remote state,
+ * and perform explicit operator-driven removals/deletions. External
  * target creation is explicit operator-driven provisioning; membership
  * sync requires the cohort target id to already be linked.
  * All ids are passed as `String` so Discord snowflakes and Google

services/api/src/main/kotlin/net/blueshell/api/shared/job/CohortJobs.kt

@@ -5,10 +5,10 @@ import net.blueshell.api.platform.integration.cohort.port.`in`.SyncCohortMembers
 /**
  * Per-target cohort membership sync jobs. One job execution pushes one
  * `(user, cohort)` pair to one external system, idempotently. The
- * payload's [SyncCohortMembershipIntent] decides whether the inbound
- * port is called with `ADD` or `REMOVE` semantics — the enum lives on
- * the application port so callers and the driving job handler share
- * one source of truth for the verb.
+ * payload's [SyncCohortMembershipIntent] decides which legacy inbound
+ * intent is sent. `ADD` pushes membership to the external system; `REMOVE`
+ * is retained for payload compatibility but provider removal is blocked
+ * for automatic Brevo sync.
  *
  * Per-pair fan-out (rather than per-user batch) means a single sync
  * failure is isolated to its own JobExecution row in the Job Manager
@@ -50,9 +50,9 @@ object CohortJobs {
     /**
      * Re-evaluates one user's cohort membership against the current
      * rules. Local desired-row writes happen here, and the evaluator
-     * enqueues one per-member [SyncCohortMembership] ADD/REMOVE job for
-     * each cohort the user joins or leaves. [ReconcileList] is the
-     * separate periodic verifier, not this job's convergence path.
+     * enqueues one per-member [SyncCohortMembership] ADD job for each
+     * newly joined cohort. Leaving a cohort is local-only; [ReconcileList]
+     * later surfaces any external extra for operator review.
      */
     object EvaluateUserCohorts : JobDefinition<EvaluateUserCohortsPayload> {
         override val type: String = "cohort.evaluate-user"

services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortMembershipSyncServiceTest.kt

@@ -38,6 +38,7 @@ class CohortMembershipSyncServiceTest {
         externalIds = externalIds,
         targetIds = targetIds,
         jobs = jobs,
+        removalPolicy = CohortMemberRemovalPolicy(),
         // A relaxed manager still runs the TransactionTemplate callbacks; the
         // real no-active-transaction guarantee is asserted in
         // CohortProviderTransactionBoundaryIT against a real transaction manager.
@@ -105,14 +106,14 @@ class CohortMembershipSyncServiceTest {
     }
 
     @Test
-    fun `REMOVE calls port when both external ids exist`() {
+    fun `automatic REMOVE for Brevo is blocked even when both external ids exist`() {
         givenCohort(id = 10L, system = "BREVO", label = "Members")
         every { externalIds.find("USER", 1L, "BREVO") } returns mapping("USER", 1L, "BREVO", "777")
         every { targetIds.find(any()) } returns "42"
 
         service.sync(userId = 1L, cohortId = 10L, intent = SyncCohortMembershipIntent.REMOVE)
 
-        verify { brevoPort.removeMember("777", "42") }
+        verify(exactly = 0) { brevoPort.removeMember(any(), any()) }
     }
 
     @Test

services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRemediationServiceTest.kt

@@ -49,6 +49,7 @@ class CohortRemediationServiceTest {
         targetIds = targetIds,
         registry = CohortPortRegistry(listOf(port)),
         jobs = jobs,
+        removalPolicy = CohortMemberRemovalPolicy(),
         transactionManager = ImmediateTransactionManager(),
     )
 

services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRuleEvaluatorTest.kt

@@ -73,21 +73,31 @@ class CohortRuleEvaluatorTest {
     }
 
     @Test
-    fun `cohorts currently joined but no longer desired are soft-deleted and a per-member REMOVE is enqueued`() {
-        every { factCollector.collect(1L) } returns emptySet()
+    fun `desired set shrink soft-deletes locally without enqueueing external REMOVE and still enqueues ADDs`() {
+        every { factCollector.collect(1L) } returns setOf(UserFact(CohortFactKind.NEWSLETTER, "true"))
         val stale = cohort(id = 99L)
+        val new = cohort(id = 20L)
         val staleMembership = membership(stale)
+        every { cohorts.findAllForEnabledSubjectFact(CohortFactKind.NEWSLETTER, "true") } returns listOf(new)
         every { memberships.findAllByUserIdAndUserIdIsNotNull(1L) } returns listOf(staleMembership)
+        every { cohorts.findById(20L) } returns Optional.of(new)
+        every { memberships.save(any<CohortMember>()) } answers { firstArg<CohortMember>() }
 
         val result = evaluator.evaluate(1L)
 
         assertThat(result.toRemove).containsExactly(99L)
-        assertThat(result.toAdd).isEmpty()
+        assertThat(result.toAdd).containsExactly(20L)
         verify { memberships.delete(staleMembership) }
         verify {
             jobs.enqueue(
                 CohortJobs.SyncCohortMembership,
-                CohortJobs.SyncCohortMembershipPayload(1L, 99L, SyncCohortMembershipIntent.REMOVE),
+                CohortJobs.SyncCohortMembershipPayload(1L, 20L, SyncCohortMembershipIntent.ADD),
+            )
+        }
+        verify(exactly = 0) {
+            jobs.enqueue(
+                CohortJobs.SyncCohortMembership,
+                match { it.intent == SyncCohortMembershipIntent.REMOVE },
             )
         }
     }