feat(cohort): operator-confirmed reconcile of external members into Blueshell

services/api/src/main/kotlin/net/blueshell/api/domain/contribution/application/ContributionService.kt

@@ -4,16 +4,19 @@ import net.blueshell.api.domain.contribution.application.event.ContributionChang
 import net.blueshell.api.domain.contribution.application.event.ContributionChanged
 import net.blueshell.api.domain.contribution.persistence.Contribution
 import net.blueshell.api.domain.contribution.persistence.repository.ContributionRepository
+import net.blueshell.api.domain.user.application.UserService
 import net.blueshell.api.shared.event.TrackedEventPublisher
 import net.blueshell.api.shared.service.BaseModelService
 import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.dao.DataIntegrityViolationException
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
 
 @Service
 class ContributionService @Autowired constructor(
     repository: ContributionRepository,
     private val periodService: ContributionPeriodService,
+    private val users: UserService,
     private val trackedEvents: TrackedEventPublisher
 ) : BaseModelService<Contribution, Contribution.Id, ContributionRepository>(repository) {
     @Transactional
@@ -63,6 +66,23 @@ class ContributionService @Autowired constructor(
         return repository.existsById(Contribution.Id(userId, periodId))
     }
 
+    @Transactional
+    fun ensurePaid(userId: Long, periodId: Long): Boolean {
+        if (existsByUserIdAndPeriodId(userId, periodId)) return false
+        try {
+            create(
+                Contribution(
+                    user = users.findById(userId),
+                    contributionPeriod = periodService.findById(periodId),
+                ),
+            )
+        } catch (ex: DataIntegrityViolationException) {
+            if (existsByUserIdAndPeriodId(userId, periodId)) return false
+            throw ex
+        }
+        return true
+    }
+
     private fun publishChange(contribution: Contribution, changeType: ContributionChange) {
         trackedEvents.publish { actor ->
             ContributionChanged(

services/api/src/main/kotlin/net/blueshell/api/domain/user/application/UserService.kt

@@ -114,6 +114,9 @@ class UserService @Autowired constructor(
         return repository.existsByDiscordAndIdNot(discord, id)
     }
 
+    fun existsActiveMembershipByUserId(userId: Long): Boolean =
+        repository.existsActiveMembershipByUserId(userId)
+
     /**
      * Batch lookup. Returns the users whose ids are in [ids] in unspecified
      * order; ids without a corresponding active user are silently skipped.

services/api/src/main/kotlin/net/blueshell/api/domain/user/application/listener/MembershipEventListener.kt

@@ -20,7 +20,7 @@ class MembershipEventListener(
         when (evt.changeType) {
             MembershipChange.CREATED,
             MembershipChange.UPDATED -> {
-                if (evt.active) {
+                if (evt.active || users.existsActiveMembershipByUserId(evt.userId)) {
                     log.info("Updating membership for user {} adding role {}", evt.userId, Role.MEMBER)
                     users.addRole(evt.userId, Role.MEMBER)
                 } else {
@@ -29,8 +29,13 @@ class MembershipEventListener(
                 }
             }
             MembershipChange.DELETED -> {
-                log.info("Deleting membership for user {} removing role {}", evt.userId, Role.MEMBER)
-                users.removeRole(evt.userId, Role.MEMBER)
+                if (users.existsActiveMembershipByUserId(evt.userId)) {
+                    log.info("Deleting inactive membership for user {} while another active membership remains", evt.userId)
+                    users.addRole(evt.userId, Role.MEMBER)
+                } else {
+                    log.info("Deleting membership for user {} removing role {}", evt.userId, Role.MEMBER)
+                    users.removeRole(evt.userId, Role.MEMBER)
+                }
             }
         }
     }

services/api/src/main/kotlin/net/blueshell/api/domain/user/application/listener/UserEventListener.kt

@@ -19,7 +19,7 @@ class UserEventListener(
     @Transactional(propagation = Propagation.REQUIRES_NEW)
     fun onUpdate(evt: UserUpdated) {
         val u = users.findById(evt.userId)
-        if (!u.hasRole(Role.MEMBER)) {
+        if (!u.hasAuthority(Role.MEMBER)) {
             u.committeeMembers.forEach { committeeMembers.delete(it) }
         }
     }

services/api/src/main/kotlin/net/blueshell/api/domain/user/persistence/repository/UserRepository.kt

@@ -28,6 +28,16 @@ interface UserRepository : BaseRepository<User, Long> {
 
     fun existsByPhoneNumberAndIdNot(phoneNumber: String, id: Long): Boolean
 
+    @Query(
+        """
+        SELECT CASE WHEN COUNT(m) > 0 THEN true ELSE false END
+        FROM Membership m
+        WHERE m.user.id = :userId
+          AND m.endDate IS NULL
+        """,
+    )
+    fun existsActiveMembershipByUserId(@Param("userId") userId: Long): Boolean
+
     /**
      * Native query that bypasses the @SQLRestriction on User so callers can
      * tell "this user existed at some point" apart from "this id was never

services/api/src/main/kotlin/net/blueshell/api/platform/integration/cohort/adapter/brevo/BrevoTargetStrategy.kt

@@ -0,0 +1,140 @@
+package net.blueshell.api.platform.integration.cohort.adapter.brevo
+
+import net.blueshell.api.platform.integration.cohort.persistence.CohortKind
+import net.blueshell.api.platform.integration.cohort.port.out.ExternalMember
+import net.blueshell.api.platform.integration.cohort.port.out.ExternalTarget
+import net.blueshell.api.platform.integration.cohort.port.out.TargetCapability
+import net.blueshell.api.platform.integration.cohort.port.out.TargetDescriptor
+import net.blueshell.api.platform.integration.cohort.port.out.TargetStrategy
+import net.blueshell.api.platform.integration.contact.adapter.ContactListAdapter
+import net.blueshell.api.platform.integration.contact.adapter.ContactServiceException
+import net.blueshell.api.shared.enums.ContactSystem
+import net.blueshell.api.shared.enums.TargetSystem
+import net.blueshell.clients.brevo.api.ContactsApi
+import net.blueshell.clients.brevo.model.GetLists200ResponseListsInner
+import net.blueshell.clients.brevo.model.GetProcessesSortParameter
+import org.slf4j.LoggerFactory
+import org.springframework.context.annotation.Profile
+import org.springframework.stereotype.Service
+import org.springframework.web.client.RestClientResponseException
+
+@Service
+@Profile("!test & !dev")
+class BrevoTargetStrategy(
+    contactListAdapters: List<ContactListAdapter>,
+    private val contactsApi: ContactsApi,
+) : TargetStrategy {
+    private val lists = contactListAdapters.single { it.system == ContactSystem.BREVO }
+
+    override val descriptor = TargetDescriptor(
+        system = TargetSystem.BREVO,
+        kind = CohortKind.LIST,
+        systemLabel = "Brevo",
+        targetLabel = "Brevo list",
+        idLabel = "List id",
+        folderLabel = "Folder",
+        capabilities = setOf(
+            TargetCapability.CATALOG,
+            TargetCapability.CREATE,
+            TargetCapability.READ_MEMBERS,
+            TargetCapability.WRITE_MEMBERS,
+            TargetCapability.DELETE,
+        ),
+    )
+
+    override fun catalog(query: String?): List<ExternalTarget> {
+        val q = query?.trim()?.lowercase().orEmpty()
+        return listTargets(folders())
+            .filter { it.matches(q) }
+            .sortedWith(compareBy({ it.folderLabel.orEmpty() }, { it.label }))
+    }
+
+    override fun create(label: String, folder: String?): ExternalTarget = ExternalTarget(
+        system = system,
+        externalId = lists.createList(label, folder).toString(),
+        kind = descriptor.kind,
+        label = label,
+        folderLabel = folder,
+    )
+
+    override fun members(target: ExternalTarget): List<ExternalMember> =
+        lists.listMembers(target.externalId.toBrevoId("externalId", "members"))
+            .map { ExternalMember(it.externalUserId.toString(), it.email) }
+            .filter { it.externalUserId.isNotBlank() }
+
+    override fun add(target: ExternalTarget, externalUserId: String) {
+        lists.addToList(
+            externalUserId.toBrevoId("externalUserId", "add"),
+            target.externalId.toBrevoId("externalId", "add"),
+        )
+    }
+
+    override fun remove(target: ExternalTarget, externalUserId: String) {
+        lists.removeFromList(
+            externalUserId.toBrevoId("externalUserId", "remove"),
+            target.externalId.toBrevoId("externalId", "remove"),
+        )
+    }
+
+    override fun delete(target: ExternalTarget) {
+        lists.deleteList(target.externalId.toBrevoId("externalId", "delete"))
+    }
+
+    private fun folders(): Map<String, String> =
+        page("folders") { limit, offset ->
+            contactsApi.getFolders(limit, offset, GetProcessesSortParameter.ASC).let { page ->
+                Page(page.count, page.folders.orEmpty().associate { it.id.toString() to it.name }.entries.toList())
+            }
+        }.associate { it.key to it.value }
+
+    private fun listTargets(folderNames: Map<String, String>): List<ExternalTarget> =
+        page("lists") { limit, offset ->
+            contactsApi.getLists(limit, offset, GetProcessesSortParameter.ASC).let { page ->
+                Page(page.count, page.lists.orEmpty().map { it.toTarget(folderNames) })
+            }
+        }
+
+    private fun <T> page(kind: String, fetch: (Long, Long) -> Page<T>): List<T> {
+        val results = mutableListOf<T>()
+        var offset = 0L
+        while (true) {
+            val page = try {
+                fetch(PAGE_SIZE, offset)
+            } catch (e: RestClientResponseException) {
+                if (e.statusCode.value() == 429) log.warn("Brevo target catalog {} fetch was rate limited", kind)
+                throw ContactServiceException("Failed to fetch Brevo $kind catalog", e)
+            }
+            results += page.items
+            if (page.items.size < PAGE_SIZE || page.count != null && results.size >= page.count) break
+            offset += PAGE_SIZE
+        }
+        return results
+    }
+
+    private fun ExternalTarget.matches(query: String): Boolean =
+        query.isBlank() ||
+            externalId == query ||
+            label.lowercase().contains(query) ||
+            folderLabel.orEmpty().lowercase().contains(query)
+
+    private fun String.toBrevoId(field: String, operation: String): Long =
+        toLongOrNull() ?: throw InvalidExternalIdException("Brevo $operation: $field \"$this\" is not numeric")
+
+    private data class Page<T>(val count: Long?, val items: List<T>)
+
+    companion object {
+        const val PAGE_SIZE: Long = 50
+        private val log = LoggerFactory.getLogger(BrevoTargetStrategy::class.java)
+    }
+}
+
+private fun GetLists200ResponseListsInner.toTarget(folderNames: Map<String, String>): ExternalTarget {
+    return ExternalTarget(
+        system = TargetSystem.BREVO,
+        externalId = id.toString(),
+        kind = CohortKind.LIST,
+        label = name,
+        folderLabel = folderNames[folderId.toString()],
+        memberCount = uniqueSubscribers,
+    )
+}

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

@@ -0,0 +1,22 @@
+package net.blueshell.api.platform.integration.cohort.adapter.job
+
+import net.blueshell.api.platform.integration.cohort.application.InboundReconcile
+import net.blueshell.api.platform.integration.queue.AbstractJsonJobHandler
+import net.blueshell.api.shared.job.CohortJobs
+import org.springframework.stereotype.Component
+import tools.jackson.databind.ObjectMapper
+
+@Component
+class ApplyInboundReconcileJobHandler(
+    objectMapper: ObjectMapper,
+    private val inbound: InboundReconcile,
+) : AbstractJsonJobHandler<CohortJobs.ApplyInboundReconcilePayload>(
+    objectMapper,
+    CohortJobs.ApplyInboundReconcile.payloadType,
+) {
+    override val jobType: String = CohortJobs.ApplyInboundReconcile.type
+
+    override fun handlePayload(payload: CohortJobs.ApplyInboundReconcilePayload) {
+        inbound.applyJob(payload)
+    }
+}

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

@@ -11,6 +11,10 @@ import net.blueshell.api.platform.integration.cohort.application.CohortSubjectDe
 import net.blueshell.api.platform.integration.cohort.application.CohortSubjectQueryService
 import net.blueshell.api.platform.integration.cohort.application.CohortSubjectSummary
 import net.blueshell.api.platform.integration.cohort.application.DriftReport
+import net.blueshell.api.platform.integration.cohort.application.InboundReconcile
+import net.blueshell.api.platform.integration.cohort.application.InboundReconcileApplyRequest
+import net.blueshell.api.platform.integration.cohort.application.InboundReconcileApplyResponse
+import net.blueshell.api.platform.integration.cohort.application.InboundReconcilePreview
 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.persistence.CohortSubjectCategory
@@ -49,6 +53,7 @@ class CohortSubjectController(
     private val drift: CohortDrift,
     private val remediation: CohortRemediation,
     private val targeting: CohortTargeting,
+    private val inboundReconcile: InboundReconcile,
 ) {
     @GetMapping
     fun findCohortSubjects(): List<CohortSubjectSummaryResponse> =
@@ -104,6 +109,21 @@ class CohortSubjectController(
         @RequestBody @Valid body: SwitchTargetRequest,
     ): CohortMappingResponse =
         targeting.switchTarget(id, cohortId, body.externalId, body.deletePrevious, body.reconcileNow).toResponse()
+
+    @PostMapping("/{id}/targets/{cohortId}/inbound-reconcile/preview")
+    fun previewInboundReconcile(
+        @PathVariable id: Long,
+        @PathVariable cohortId: Long,
+    ): InboundReconcilePreview =
+        inboundReconcile.preview(id, cohortId)
+
+    @PostMapping("/{id}/targets/{cohortId}/inbound-reconcile/apply")
+    fun applyInboundReconcile(
+        @PathVariable id: Long,
+        @PathVariable cohortId: Long,
+        @RequestBody @Valid body: InboundReconcileApplyRequest,
+    ): InboundReconcileApplyResponse =
+        inboundReconcile.apply(id, cohortId, body)
 }
 
 // ── Request / response DTOs ──────────────────────────────────────────────────

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

@@ -0,0 +1,33 @@
+package net.blueshell.api.platform.integration.cohort.adapter.web
+
+import io.swagger.v3.oas.annotations.Operation
+import io.swagger.v3.oas.annotations.tags.Tag
+import net.blueshell.api.platform.integration.cohort.application.TargetCatalog
+import net.blueshell.api.platform.integration.cohort.port.out.ExternalTarget
+import net.blueshell.api.platform.integration.cohort.port.out.TargetDescriptor
+import net.blueshell.api.shared.enums.TargetSystem
+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.RequestMapping
+import org.springframework.web.bind.annotation.RequestParam
+import org.springframework.web.bind.annotation.RestController
+
+@RestController
+@RequestMapping("/management/cohort-targets")
+@Tag(name = "Cohort Targets", description = "Admin: external cohort target catalog")
+@PreAuthorize("hasAuthority('ADMIN')")
+class CohortTargetController(
+    private val catalog: TargetCatalog,
+) {
+    @GetMapping("/systems")
+    @Operation(operationId = "listCohortTargetSystems")
+    fun systems(): List<TargetDescriptor> = catalog.descriptors()
+
+    @GetMapping("/{system}")
+    @Operation(operationId = "searchCohortTargets")
+    fun targets(
+        @PathVariable system: TargetSystem,
+        @RequestParam(required = false) query: String?,
+    ): List<ExternalTarget> = catalog.search(system, query)
+}

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

@@ -2,7 +2,6 @@ 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.CohortFactKind
-import net.blueshell.api.platform.integration.cohort.persistence.CohortKind
 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.CohortRepository
@@ -23,6 +22,7 @@ import org.springframework.transaction.annotation.Transactional
 class CohortProvisioningService(
     private val subjects: CohortSubjectRepository,
     private val cohorts: CohortRepository,
+    private val strategies: TargetStrategies,
 ) {
     @Transactional
     fun provision(spec: CohortProvisioningSpec): CohortProvisioningResult {
@@ -43,19 +43,14 @@ class CohortProvisioningService(
             ?: cohorts.save(
                 Cohort(
                     system = spec.system.name,
-                    kind = kindFor(spec.system),
+                    kind = strategies.descriptor(spec.system).kind,
                     label = spec.label,
                     folder = spec.folder,
                     subjectId = subject.id,
                 ),
-            )
+        )
         return CohortProvisioningResult.Ready(cohort)
     }
-
-    private fun kindFor(system: TargetSystem): CohortKind = when (system) {
-        TargetSystem.BREVO -> CohortKind.LIST
-        TargetSystem.GOOGLE_CALENDAR -> error("$system has no cohort target kind")
-    }
 }
 
 /**