#60 Can follow citizen

build.gradle.kts

@@ -426,7 +426,7 @@ dependencies {
     testImplementation("io.insert-koin:koin-test:+")
     testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:+")
     testImplementation("io.mockk:mockk:+")
-    testImplementation("org.junit.jupiter:junit-jupiter:+")
+    testImplementation("org.junit.jupiter:junit-jupiter:5.7.+")
     testImplementation("org.amshove.kluent:kluent:+")
     testImplementation("io.mockk:mockk-agent-api:+")
     testImplementation("io.mockk:mockk-agent-jvm:+")

src/main/kotlin/fr/dcproject/application/Application.kt

@@ -9,6 +9,7 @@ import com.fasterxml.jackson.datatype.joda.JodaModule
 import fr.dcproject.application.Env.PROD
 import fr.dcproject.application.Env.TEST
 import fr.dcproject.application.http.statusPagesInstallation
+import fr.dcproject.common.utils.onApplicationStopped
 import fr.dcproject.component.article.articleKoinModule
 import fr.dcproject.component.article.routes.installArticleRoutes
 import fr.dcproject.component.auth.authKoinModule
@@ -25,8 +26,10 @@ import fr.dcproject.component.constitution.routes.installConstitutionRoutes
 import fr.dcproject.component.doc.routes.installDocRoutes
 import fr.dcproject.component.follow.followKoinModule
 import fr.dcproject.component.follow.routes.article.installFollowArticleRoutes
+import fr.dcproject.component.follow.routes.citizen.installFollowCitizenRoutes
 import fr.dcproject.component.follow.routes.constitution.installFollowConstitutionRoutes
-import fr.dcproject.component.notification.NotificationConsumer
+import fr.dcproject.component.notification.email.NotificationEmailConsumer
+import fr.dcproject.component.notification.push.NotificationPushConsumer
 import fr.dcproject.component.notification.routes.installNotificationsRoutes
 import fr.dcproject.component.opinion.opinionKoinModule
 import fr.dcproject.component.opinion.routes.installOpinionRoutes
@@ -37,7 +40,6 @@ import fr.dcproject.component.workgroup.routes.installWorkgroupRoutes
 import fr.dcproject.component.workgroup.workgroupKoinModule
 import fr.postgresjson.migration.Migrations
 import io.ktor.application.Application
-import io.ktor.application.ApplicationStopped
 import io.ktor.application.install
 import io.ktor.auth.Authentication
 import io.ktor.client.HttpClient
@@ -117,11 +119,14 @@ fun Application.module(env: Env = PROD) {
         masking = false
     }
 
-    get<NotificationConsumer>().run {
+    get<NotificationEmailConsumer>().run {
         start()
-        environment.monitor.subscribe(ApplicationStopped) {
-            close()
-        }
+        onApplicationStopped { close() }
+    }
+
+    get<NotificationPushConsumer>().run {
+        start()
+        onApplicationStopped { close() }
     }
 
     install(Authentication, jwtInstallation(get(), get()))
@@ -154,6 +159,7 @@ fun Application.module(env: Env = PROD) {
         installCommentRoutes()
         installFollowArticleRoutes()
         installFollowConstitutionRoutes()
+        installFollowCitizenRoutes()
         installWorkgroupRoutes()
         installOpinionRoutes()
         installVoteRoutes()

src/main/kotlin/fr/dcproject/application/KoinModule.kt

@@ -10,10 +10,11 @@ import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import com.rabbitmq.client.ConnectionFactory
 import fr.dcproject.common.email.Mailer
 import fr.dcproject.component.auth.jwt.JwtConfig
-import fr.dcproject.component.notification.NotificationConsumer
-import fr.dcproject.component.notification.NotificationEmailSender
-import fr.dcproject.component.notification.NotificationsPush
-import fr.dcproject.component.notification.Publisher
+import fr.dcproject.component.notification.NotificationPublisherAsync
+import fr.dcproject.component.notification.email.NotificationEmailConsumer
+import fr.dcproject.component.notification.email.NotificationEmailSender
+import fr.dcproject.component.notification.push.NotificationPushConsumer
+import fr.dcproject.component.notification.push.NotificationPushListener
 import fr.postgresjson.connexion.Connection
 import fr.postgresjson.connexion.Requester
 import fr.postgresjson.migration.Migrations
@@ -65,11 +66,15 @@ val KoinModule = module {
         }
     }
 
-    single { NotificationsPush.Builder(get()) }
+    single { NotificationPushListener.Builder(get()) }
 
     single {
         val config: Configuration = get()
-        NotificationConsumer(get(), get(), get(), get(), get(), config.exchangeNotificationName)
+        NotificationEmailConsumer(get(), get(), get(), get(), get(), config.exchangeNotificationName)
+    }
+    single {
+        val config: Configuration = get()
+        NotificationPushConsumer(get(), get(), get(), get(), get(), config.exchangeNotificationName)
     }
 
     // RabbitMQ
@@ -114,7 +119,7 @@ val KoinModule = module {
 
     single {
         val config: Configuration = get()
-        Publisher(factory = get(), exchangeName = config.exchangeNotificationName)
+        NotificationPublisherAsync(factory = get(), exchangeName = config.exchangeNotificationName)
     }
 
     single {

src/main/kotlin/fr/dcproject/common/email/Mailer.kt

@@ -9,6 +9,9 @@ import java.io.IOException
 class Mailer(
     private val key: String
 ) {
+    /**
+     * Send email via Sendgrid
+     */
     fun sendEmail(action: () -> Mail): Boolean {
         val mail = action()
 

src/main/kotlin/fr/dcproject/common/entity/Extra.kt

@@ -2,6 +2,7 @@ package fr.dcproject.common.entity
 
 import fr.dcproject.component.article.database.ArticleRef
 import fr.dcproject.component.citizen.database.CitizenI
+import fr.dcproject.component.citizen.database.CitizenRef
 import fr.dcproject.component.comment.generic.database.CommentRef
 import fr.dcproject.component.constitution.database.ConstitutionRef
 import fr.dcproject.component.opinion.database.OpinionRef
@@ -34,7 +35,8 @@ interface TargetI : EntityI {
         Article("article"),
         Constitution("constitution"),
         Comment("comment"),
-        Opinion("opinion")
+        Opinion("opinion"),
+        Citizen("citizen"),
     }
 
     companion object {
@@ -44,6 +46,7 @@ interface TargetI : EntityI {
                 t.isSubclassOf(ConstitutionRef::class) -> TargetName.Constitution.targetReference
                 t.isSubclassOf(CommentRef::class) -> TargetName.Comment.targetReference
                 t.isSubclassOf(OpinionRef::class) -> TargetName.Opinion.targetReference
+                t.isSubclassOf(CitizenRef::class) -> TargetName.Citizen.targetReference
                 else -> throw error("target not implemented: ${t.qualifiedName} \nImplement it or return 'reference' from SQL")
             }
         }

src/main/kotlin/fr/dcproject/common/utils/Ktor.kt

@@ -0,0 +1,10 @@
+package fr.dcproject.common.utils
+
+import io.ktor.application.Application
+import io.ktor.application.ApplicationStopped
+
+fun Application.onApplicationStopped(callback: Application.() -> Unit) {
+    environment.monitor.subscribe(ApplicationStopped) {
+        callback()
+    }
+}

src/main/kotlin/fr/dcproject/common/utils/RabbitConsume.kt

@@ -0,0 +1,30 @@
+package fr.dcproject.common.utils
+
+import com.rabbitmq.client.AMQP
+import com.rabbitmq.client.Channel
+import com.rabbitmq.client.Consumer
+import com.rabbitmq.client.DefaultConsumer
+import com.rabbitmq.client.Envelope
+import kotlinx.coroutines.runBlocking
+import java.io.IOException
+
+fun Channel.consumeQueue(queueName: String, callback: DefaultConsumer.(ByteArray) -> Unit) {
+    val consumer: Consumer = object : DefaultConsumer(this) {
+        @Throws(IOException::class)
+        override fun handleDelivery(
+            consumerTag: String,
+            envelope: Envelope,
+            properties: AMQP.BasicProperties,
+            body: ByteArray
+        ) = runBlocking {
+            try {
+                callback(body)
+                basicAck(envelope.deliveryTag, false)
+            } catch (e: Throwable) {
+                basicNack(envelope.deliveryTag, false, true)
+            }
+        }
+    }
+    /* Launch Consumer */
+    basicConsume(queueName, false, consumer)
+}

src/main/kotlin/fr/dcproject/component/article/routes/UpsertArticle.kt

@@ -10,8 +10,8 @@ import fr.dcproject.component.article.routes.UpsertArticle.UpsertArticleRequest.
 import fr.dcproject.component.auth.citizen
 import fr.dcproject.component.auth.citizenOrNull
 import fr.dcproject.component.auth.mustBeAuth
-import fr.dcproject.component.notification.ArticleUpdateNotification
-import fr.dcproject.component.notification.Publisher
+import fr.dcproject.component.notification.ArticleUpdateNotificationMessage
+import fr.dcproject.component.notification.NotificationPublisherAsync
 import fr.dcproject.component.workgroup.database.WorkgroupRef
 import io.konform.validation.Validation
 import io.konform.validation.jsonschema.maxItems
@@ -63,7 +63,7 @@ object UpsertArticle {
         }
     }
 
-    fun Route.upsertArticle(repo: ArticleRepository, publisher: Publisher, ac: ArticleAccessControl) {
+    fun Route.upsertArticle(repo: ArticleRepository, notificationPublisher: NotificationPublisherAsync, ac: ArticleAccessControl) {
         suspend fun ApplicationCall.convertRequestToEntity(): ArticleForUpdate = receiveOrBadRequest<Input>().run {
             validate().badRequestIfNotValid()
             ArticleForUpdate(
@@ -92,7 +92,7 @@ object UpsertArticle {
                         val versionNumber = a.versionNumber
                     }
                 )
-                publisher.publish(ArticleUpdateNotification(a))
+                notificationPublisher.publishAsync(ArticleUpdateNotificationMessage(a))
             } ?: error("Article not updated")
         }
     }

src/main/kotlin/fr/dcproject/component/citizen/database/Citizen.kt

@@ -2,8 +2,9 @@ package fr.dcproject.component.citizen.database
 
 import fr.dcproject.common.entity.CreatedAt
 import fr.dcproject.common.entity.DeletedAt
-import fr.dcproject.common.entity.Entity
 import fr.dcproject.common.entity.EntityI
+import fr.dcproject.common.entity.TargetI
+import fr.dcproject.common.entity.TargetRef
 import fr.dcproject.component.auth.database.User
 import fr.dcproject.component.auth.database.UserCreator
 import fr.dcproject.component.auth.database.UserForCreate
@@ -95,10 +96,10 @@ open class CitizenRefWithUser(
 
 open class CitizenRef(
     id: UUID = UUID.randomUUID()
-) : Entity(id),
+) : TargetRef(id),
     CitizenI
 
-interface CitizenI : EntityI {
+interface CitizenI : EntityI, TargetI {
     data class Name(
         override val firstName: String,
         override val lastName: String,

src/main/kotlin/fr/dcproject/component/follow/KoinModule.kt

@@ -1,11 +1,13 @@
 package fr.dcproject.component.follow
 
 import fr.dcproject.component.follow.database.FollowArticleRepository
+import fr.dcproject.component.follow.database.FollowCitizenRepository
 import fr.dcproject.component.follow.database.FollowConstitutionRepository
 import org.koin.dsl.module
 
 val followKoinModule = module {
     single { FollowArticleRepository(get()) }
     single { FollowConstitutionRepository(get()) }
+    single { FollowCitizenRepository(get()) }
     single { FollowAccessControl() }
 }

src/main/kotlin/fr/dcproject/component/follow/database/FollowRepository.kt

@@ -4,7 +4,9 @@ import fr.dcproject.common.entity.Entity
 import fr.dcproject.common.entity.TargetRef
 import fr.dcproject.component.article.database.ArticleForView
 import fr.dcproject.component.article.database.ArticleRef
+import fr.dcproject.component.citizen.database.Citizen
 import fr.dcproject.component.citizen.database.CitizenI
+import fr.dcproject.component.citizen.database.CitizenRef
 import fr.dcproject.component.constitution.database.ConstitutionForView
 import fr.dcproject.component.constitution.database.ConstitutionRef
 import fr.postgresjson.connexion.Paginated
@@ -72,21 +74,24 @@ sealed class FollowRepository<IN : TargetRef, OUT : TargetRef>(override var requ
         target: Entity,
         bulkSize: Int = 300
     ): Flow<FollowForView<IN>> = flow {
-        var nextPage = 1
-        do {
-            val paginate = findFollowsByTarget(target, nextPage, bulkSize)
-            paginate.result.forEach {
+        var lastId: UUID? = null
+        while (true) {
+            val result = findFollowsByTarget(target, lastId, bulkSize)
+            if (result.count() == 0) {
+                break
+            }
+            result.forEach {
                 emit(it)
             }
-            nextPage = paginate.currentPage + 1
-        } while (!paginate.isLastPage())
+            lastId = result.last().id
+        }
     }
 
     abstract fun findFollowsByTarget(
         target: Entity,
-        page: Int = 1,
+        lastId: UUID?,
         limit: Int = 300
-    ): Paginated<FollowForView<IN>>
+    ): List<FollowForView<IN>>
 }
 
 class FollowArticleRepository(requester: Requester) : FollowRepository<ArticleRef, ArticleForView>(requester) {
@@ -107,14 +112,14 @@ class FollowArticleRepository(requester: Requester) : FollowRepository<ArticleRe
 
     override fun findFollowsByTarget(
         target: Entity,
-        page: Int,
+        lastId: UUID?,
         limit: Int
-    ): Paginated<FollowForView<ArticleRef>> {
+    ): List<FollowForView<ArticleRef>> {
         return requester
             .getFunction("find_follows_article_by_target")
             .select(
-                page,
-                limit,
+                "start_id" to lastId,
+                "limit" to limit,
                 "target_id" to target.id
             )
     }
@@ -138,9 +143,34 @@ class FollowConstitutionRepository(requester: Requester) : FollowRepository<Cons
 
     override fun findFollowsByTarget(
         target: Entity,
+        lastId: UUID?,
+        limit: Int
+    ): List<FollowForView<ConstitutionRef>> {
+        TODO("Not yet implemented")
+    }
+}
+
+class FollowCitizenRepository(requester: Requester) : FollowRepository<CitizenRef, Citizen>(requester) {
+    override fun findByCitizen(
+        citizenId: UUID,
         page: Int,
         limit: Int
-    ): Paginated<FollowForView<ConstitutionRef>> {
+    ): Paginated<FollowForView<Citizen>> {
+        return requester.run {
+            getFunction("find_follows_citizen_by_citizen")
+                .select(
+                    page,
+                    limit,
+                    "created_by_id" to citizenId
+                )
+        }
+    }
+
+    override fun findFollowsByTarget(
+        target: Entity,
+        lastId: UUID?,
+        limit: Int
+    ): List<FollowForView<CitizenRef>> {
         TODO("Not yet implemented")
     }
 }

src/main/kotlin/fr/dcproject/component/follow/routes/article/GetFollowArticle.kt

@@ -7,14 +7,14 @@ import fr.dcproject.component.auth.citizen
 import fr.dcproject.component.auth.citizenOrNull
 import fr.dcproject.component.follow.FollowAccessControl
 import fr.dcproject.component.follow.database.FollowArticleRepository
+import fr.dcproject.component.follow.routes.citizen.toOutput
 import io.ktor.application.call
 import io.ktor.http.HttpStatusCode
 import io.ktor.locations.KtorExperimentalLocationsAPI
 import io.ktor.locations.Location
 import io.ktor.locations.get
 import io.ktor.response.respond
 import io.ktor.routing.Route
-import org.joda.time.DateTime
 import java.util.UUID
 
 @KtorExperimentalLocationsAPI
@@ -30,19 +30,7 @@ object GetFollowArticle {
                 ac.assert { canView(follow, citizenOrNull) }
                 call.respond(
                     HttpStatusCode.OK,
-                    follow.let { f ->
-                        object {
-                            val id: UUID = f.id
-                            val createdBy: Any = f.createdBy.toOutput()
-                            val target: Any = f.target.let { t ->
-                                object {
-                                    val id: UUID = t.id
-                                    val reference: String = f.target.reference
-                                }
-                            }
-                            val createdAt: DateTime = f.createdAt
-                        }
-                    }
+                    follow.toOutput()
                 )
             } ?: call.respond(HttpStatusCode.NoContent)
         }