diff --git a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortMemberRemovalPolicy.kt b/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortMemberRemovalPolicy.kt new file mode 100644 index 000000000..085e0ebde --- /dev/null +++ b/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 +} diff --git a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortMembershipSyncService.kt b/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortMembershipSyncService.kt index 8e14d2404..950ad86e3 100644 --- a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortMembershipSyncService.kt +++ b/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) } diff --git a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRemediationService.kt b/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRemediationService.kt index 70ef97fa1..628094957 100644 --- a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRemediationService.kt +++ b/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) } diff --git a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRuleEvaluator.kt b/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRuleEvaluator.kt index 9329bf582..be99a2681 100644 --- a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRuleEvaluator.kt +++ b/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( diff --git a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/listener/CohortRuleListener.kt b/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/listener/CohortRuleListener.kt index 5d9a57856..b30e949ba 100644 --- a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/application/listener/CohortRuleListener.kt +++ b/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) } diff --git a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/in/CohortMembershipSync.kt b/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/in/CohortMembershipSync.kt index 399d9917e..5613dace2 100644 --- a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/in/CohortMembershipSync.kt +++ b/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 } diff --git a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/in/CohortReconciliation.kt b/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/in/CohortReconciliation.kt index fee693b96..1ae5e49f2 100644 --- a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/in/CohortReconciliation.kt +++ b/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) diff --git a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/out/CohortPort.kt b/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/out/CohortPort.kt index 154c96697..34d433e1b 100644 --- a/services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/port/out/CohortPort.kt +++ b/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//`. * * 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 diff --git a/services/api/src/main/kotlin/net/blueshell/api/shared/job/CohortJobs.kt b/services/api/src/main/kotlin/net/blueshell/api/shared/job/CohortJobs.kt index 6fcb11771..04e3135d7 100644 --- a/services/api/src/main/kotlin/net/blueshell/api/shared/job/CohortJobs.kt +++ b/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 { override val type: String = "cohort.evaluate-user" diff --git a/services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortMembershipSyncServiceTest.kt b/services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortMembershipSyncServiceTest.kt index 96b94c472..129777bef 100644 --- a/services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortMembershipSyncServiceTest.kt +++ b/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 diff --git a/services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRemediationServiceTest.kt b/services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRemediationServiceTest.kt index c89474d87..9a96e597e 100644 --- a/services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRemediationServiceTest.kt +++ b/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(), ) diff --git a/services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRuleEvaluatorTest.kt b/services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRuleEvaluatorTest.kt index a91d28616..323c37558 100644 --- a/services/api/src/test/kotlin/net/blueshell/api/platform/integration/cohort/application/CohortRuleEvaluatorTest.kt +++ b/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()) } answers { firstArg() } 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 }, ) } }