Merge pull request #149 from usbharu/feature/jdbc-timeline

feat: タイムラインのJDBC実装

src/main/kotlin/dev/usbharu/hideout/application/infrastructure/exposed/ExposedTransaction.kt

@@ -4,11 +4,12 @@ import dev.usbharu.hideout.application.external.Transaction
 import kotlinx.coroutines.slf4j.MDCContext
 import org.jetbrains.exposed.sql.transactions.experimental.newSuspendedTransaction
 import org.springframework.stereotype.Service
+import java.sql.Connection
 
 @Service
 class ExposedTransaction : Transaction {
     override suspend fun <T> transaction(block: suspend () -> T): T {
-        return newSuspendedTransaction(MDCContext()) {
+        return newSuspendedTransaction(MDCContext(), transactionIsolation = Connection.TRANSACTION_SERIALIZABLE) {
             block()
         }
     }

src/main/kotlin/dev/usbharu/hideout/core/infrastructure/exposedrepository/ExposedTimelineRepository.kt

@@ -0,0 +1,119 @@
+package dev.usbharu.hideout.core.infrastructure.exposedrepository
+
+import dev.usbharu.hideout.application.service.id.IdGenerateService
+import dev.usbharu.hideout.core.domain.model.post.Visibility
+import dev.usbharu.hideout.core.domain.model.timeline.Timeline
+import dev.usbharu.hideout.core.domain.model.timeline.TimelineRepository
+import org.jetbrains.exposed.sql.*
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.stereotype.Repository
+
+@Repository
+@Qualifier("jdbc")
+@ConditionalOnProperty("hideout.use-mongodb", havingValue = "false", matchIfMissing = true)
+class ExposedTimelineRepository(private val idGenerateService: IdGenerateService) : TimelineRepository {
+    override suspend fun generateId(): Long = idGenerateService.generateId()
+
+    override suspend fun save(timeline: Timeline): Timeline {
+        if (Timelines.select { Timelines.id eq timeline.id }.singleOrNull() == null) {
+            Timelines.insert {
+                it[id] = timeline.id
+                it[userId] = timeline.userId
+                it[timelineId] = timeline.timelineId
+                it[postId] = timeline.postId
+                it[postUserId] = timeline.postUserId
+                it[createdAt] = timeline.createdAt
+                it[replyId] = timeline.replyId
+                it[repostId] = timeline.repostId
+                it[visibility] = timeline.visibility.ordinal
+                it[sensitive] = timeline.sensitive
+                it[isLocal] = timeline.isLocal
+                it[isPureRepost] = timeline.isPureRepost
+                it[mediaIds] = timeline.mediaIds.joinToString(",")
+            }
+        } else {
+            Timelines.update({ Timelines.id eq timeline.id }) {
+                it[userId] = timeline.userId
+                it[timelineId] = timeline.timelineId
+                it[postId] = timeline.postId
+                it[postUserId] = timeline.postUserId
+                it[createdAt] = timeline.createdAt
+                it[replyId] = timeline.replyId
+                it[repostId] = timeline.repostId
+                it[visibility] = timeline.visibility.ordinal
+                it[sensitive] = timeline.sensitive
+                it[isLocal] = timeline.isLocal
+                it[isPureRepost] = timeline.isPureRepost
+                it[mediaIds] = timeline.mediaIds.joinToString(",")
+            }
+        }
+        return timeline
+    }
+
+    override suspend fun saveAll(timelines: List<Timeline>): List<Timeline> {
+        Timelines.batchInsert(timelines, true, false) {
+            this[Timelines.id] = it.id
+            this[Timelines.userId] = it.userId
+            this[Timelines.timelineId] = it.timelineId
+            this[Timelines.postId] = it.postId
+            this[Timelines.postUserId] = it.postUserId
+            this[Timelines.createdAt] = it.createdAt
+            this[Timelines.replyId] = it.replyId
+            this[Timelines.repostId] = it.repostId
+            this[Timelines.visibility] = it.visibility.ordinal
+            this[Timelines.sensitive] = it.sensitive
+            this[Timelines.isLocal] = it.isLocal
+            this[Timelines.isPureRepost] = it.isPureRepost
+            this[Timelines.mediaIds] = it.mediaIds.joinToString(",")
+        }
+        return timelines
+    }
+
+    override suspend fun findByUserId(id: Long): List<Timeline> =
+        Timelines.select { Timelines.userId eq id }.map { it.toTimeline() }
+
+    override suspend fun findByUserIdAndTimelineId(userId: Long, timelineId: Long): List<Timeline> =
+        Timelines.select { Timelines.userId eq userId and (Timelines.timelineId eq timelineId) }
+            .map { it.toTimeline() }
+}
+
+fun ResultRow.toTimeline(): Timeline {
+    return Timeline(
+        id = this[Timelines.id],
+        userId = this[Timelines.userId],
+        timelineId = this[Timelines.timelineId],
+        postId = this[Timelines.postId],
+        postUserId = this[Timelines.postUserId],
+        createdAt = this[Timelines.createdAt],
+        replyId = this[Timelines.replyId],
+        repostId = this[Timelines.repostId],
+        visibility = Visibility.values().first { it.ordinal == this[Timelines.visibility] },
+        sensitive = this[Timelines.sensitive],
+        isLocal = this[Timelines.isLocal],
+        isPureRepost = this[Timelines.isPureRepost],
+        mediaIds = this[Timelines.mediaIds].split(",").mapNotNull { it.toLongOrNull() }
+    )
+}
+
+object Timelines : Table("timelines") {
+    val id = long("id")
+    val userId = long("user_id")
+    val timelineId = long("timeline_id")
+    val postId = long("post_id")
+    val postUserId = long("post_user_id")
+    val createdAt = long("created_at")
+    val replyId = long("reply_id").nullable()
+    val repostId = long("repost_id").nullable()
+    val visibility = integer("visibility")
+    val sensitive = bool("sensitive")
+    val isLocal = bool("is_local")
+    val isPureRepost = bool("is_pure_repost")
+    val mediaIds = varchar("media_ids", 255)
+
+    override val primaryKey: PrimaryKey = PrimaryKey(id)
+
+    init {
+        uniqueIndex(userId, timelineId, postId)
+    }
+}

src/main/kotlin/dev/usbharu/hideout/core/infrastructure/exposedrepository/PostRepositoryImpl.kt

@@ -59,11 +59,6 @@ class PostRepositoryImpl(
                 it[apId] = post.apId
             }
         }
-
-        assert(Posts.select { Posts.id eq post.id }.singleOrNull() != null) {
-            "Faild to insert"
-        }
-
         return singleOrNull == null
     }
 

src/main/kotlin/dev/usbharu/hideout/core/infrastructure/kjobexposed/KJobJobQueueParentService.kt

@@ -5,21 +5,23 @@ import kjob.core.Job
 import kjob.core.KJob
 import kjob.core.dsl.ScheduleContext
 import kjob.core.kjob
-import org.jetbrains.exposed.sql.Database
+import org.jetbrains.exposed.sql.transactions.TransactionManager
 import org.slf4j.LoggerFactory
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.stereotype.Service
 
 @Service
 @ConditionalOnProperty(name = ["hideout.use-mongodb"], havingValue = "false", matchIfMissing = true)
-class KJobJobQueueParentService(private val database: Database) : JobQueueParentService {
+class KJobJobQueueParentService() : JobQueueParentService {
 
     private val logger = LoggerFactory.getLogger(this::class.java)
 
-    val kjob: KJob = kjob(ExposedKJob) {
+    val kjob: KJob by lazy {
-        connectionDatabase = database
+        kjob(ExposedKJob) {
-        isWorker = false
+            connectionDatabase = TransactionManager.defaultDatabase
-    }.start()
+            isWorker = false
+        }.start()
+    }
 
     override fun init(jobDefines: List<Job>) = Unit
 

src/main/kotlin/dev/usbharu/hideout/core/infrastructure/kjobexposed/KJobJobQueueWorkerService.kt

@@ -4,7 +4,7 @@ import dev.usbharu.hideout.core.service.job.JobQueueWorkerService
 import kjob.core.dsl.JobRegisterContext
 import kjob.core.dsl.KJobFunctions
 import kjob.core.kjob
-import org.jetbrains.exposed.sql.Database
+import org.jetbrains.exposed.sql.transactions.TransactionManager
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.stereotype.Service
 import dev.usbharu.hideout.core.external.job.HideoutJob as HJ
@@ -12,11 +12,11 @@ import kjob.core.dsl.JobContextWithProps as JCWP
 
 @Service
 @ConditionalOnProperty(name = ["hideout.use-mongodb"], havingValue = "false", matchIfMissing = true)
-class KJobJobQueueWorkerService(private val database: Database) : JobQueueWorkerService {
+class KJobJobQueueWorkerService() : JobQueueWorkerService {
 
     val kjob by lazy {
         kjob(ExposedKJob) {
-            connectionDatabase = database
+            connectionDatabase = TransactionManager.defaultDatabase
             nonBlockingMaxJobs = 10
             blockingMaxJobs = 10
             jobExecutionPeriodInSeconds = 1

src/main/kotlin/dev/usbharu/hideout/core/infrastructure/mongorepository/MongoTimelineRepositoryWrapper.kt

@@ -10,7 +10,7 @@ import org.springframework.stereotype.Repository
 
 @Repository
 @Suppress("InjectDispatcher")
-@ConditionalOnProperty("hideout.use-mongodb", havingValue = "", matchIfMissing = false)
+@ConditionalOnProperty("hideout.use-mongodb", havingValue = "true", matchIfMissing = false)
 class MongoTimelineRepositoryWrapper(
     private val mongoTimelineRepository: MongoTimelineRepository,
     private val idGenerateService: IdGenerateService

src/main/kotlin/dev/usbharu/hideout/core/service/post/PostServiceImpl.kt

@@ -43,11 +43,13 @@ class PostServiceImpl(
             if (postRepository.save(post)) {
                 try {
                     timelineService.publishTimeline(post, isLocal)
-                } catch (_: DuplicateKeyException) {
+                } catch (e: DuplicateKeyException) {
+                    logger.trace("Timeline already exists.", e)
                 }
             }
             post
-        } catch (_: ExposedSQLException) {
+        } catch (e: ExposedSQLException) {
+            logger.warn("FAILED Save to post. url: ${post.apId}", e)
             postQueryService.findByApId(post.apId)
         }
     }

src/main/kotlin/dev/usbharu/hideout/core/service/timeline/ExposedGenerateTimelineService.kt

@@ -0,0 +1,54 @@
+package dev.usbharu.hideout.core.service.timeline
+
+import dev.usbharu.hideout.core.infrastructure.exposedrepository.Timelines
+import dev.usbharu.hideout.domain.mastodon.model.generated.Status
+import dev.usbharu.hideout.mastodon.interfaces.api.status.StatusQuery
+import dev.usbharu.hideout.mastodon.query.StatusQueryService
+import org.jetbrains.exposed.sql.SortOrder
+import org.jetbrains.exposed.sql.andWhere
+import org.jetbrains.exposed.sql.selectAll
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.stereotype.Service
+
+@Service
+@ConditionalOnProperty("hideout.use-mongodb", havingValue = "false", matchIfMissing = true)
+class ExposedGenerateTimelineService(private val statusQueryService: StatusQueryService) : GenerateTimelineService {
+    override suspend fun getTimeline(
+        forUserId: Long?,
+        localOnly: Boolean,
+        mediaOnly: Boolean,
+        maxId: Long?,
+        minId: Long?,
+        sinceId: Long?,
+        limit: Int
+    ): List<Status> {
+        val query = Timelines.selectAll()
+
+        if (forUserId != null) {
+            query.andWhere { Timelines.userId eq forUserId }
+        }
+        if (localOnly) {
+            query.andWhere { Timelines.isLocal eq true }
+        }
+        if (maxId != null) {
+            query.andWhere { Timelines.id lessEq maxId }
+        }
+        if (minId != null) {
+            query.andWhere { Timelines.id greaterEq minId }
+        }
+        val result = query
+            .limit(limit)
+            .orderBy(Timelines.createdAt, SortOrder.DESC)
+
+        val statusQueries = result.map {
+            StatusQuery(
+                it[Timelines.postId],
+                it[Timelines.replyId],
+                it[Timelines.repostId],
+                it[Timelines.mediaIds].split(",").mapNotNull { it.toLongOrNull() }
+            )
+        }
+
+        return statusQueryService.findByPostIdsWithMediaIds(statusQueries)
+    }
+}

src/main/kotlin/dev/usbharu/hideout/core/service/timeline/TimelineService.kt

@@ -6,6 +6,7 @@ import dev.usbharu.hideout.core.domain.model.timeline.Timeline
 import dev.usbharu.hideout.core.domain.model.timeline.TimelineRepository
 import dev.usbharu.hideout.core.query.FollowerQueryService
 import dev.usbharu.hideout.core.query.UserQueryService
+import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Service
 
 @Service
@@ -58,5 +59,10 @@ class TimelineService(
             )
         }
         timelineRepository.saveAll(timelines)
+        logger.debug("SUCCESS Timeline published. {}", timelines.size)
+    }
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(TimelineService::class.java)
     }
 }

src/main/resources/application.yml

@@ -1,6 +1,6 @@
 hideout:
   url: "https://test-hideout.usbharu.dev"
-  use-mongodb: true
+  use-mongodb: false
   security:
     jwt:
       generate: true
@@ -19,15 +19,15 @@ spring:
     default-property-inclusion: always
   datasource:
     driver-class-name: org.h2.Driver
-    url: "jdbc:h2:./test-dev3;MODE=POSTGRESQL"
+    url: "jdbc:h2:./test-dev3;MODE=POSTGRESQL;TRACE_LEVEL_FILE=4"
     username: ""
     password: ""
-  data:
+  #  data:
-    mongodb:
+  #    mongodb:
-      auto-index-creation: true
+  #      auto-index-creation: true
-      host: localhost
+  #      host: localhost
-      port: 27017
+  #      port: 27017
-      database: hideout
+  #      database: hideout
   #      username: hideoutuser
   #      password: hideoutpass
   servlet: