core/core/exposed/src/jvmMain/kotlin/dev/inmo/postssystem/core/exposed/ExposedPostsRepo.kt

216 lines
7.0 KiB
Kotlin

package dev.inmo.postssystem.core.exposed
import dev.inmo.postssystem.core.content.ContentId
import dev.inmo.postssystem.core.generatePostId
import dev.inmo.postssystem.core.post.*
import dev.inmo.postssystem.core.post.repo.PostsRepo
import com.soywiz.klock.*
import dev.inmo.micro_utils.pagination.*
import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.asFlow
import org.jetbrains.exposed.sql.*
import org.jetbrains.exposed.sql.transactions.transaction
private class PostsRepoContentRelations(
private val database: Database
) : Table() {
private val postIdColumn = text("postId")
private val contentIdColumn = text("contentId")
init {
transaction (
db = database
) {
SchemaUtils.createMissingTablesAndColumns(this@PostsRepoContentRelations)
}
}
fun getPostContents(postId: PostId): List<ContentId> {
return transaction(
db = database
) {
select { postIdColumn.eq(postId) }.map { it[contentIdColumn] }
}
}
fun getContentPosts(contentId: ContentId): List<PostId> {
return transaction(
db = database
) {
select { contentIdColumn.eq(contentId) }.map { it[postIdColumn] }
}
}
fun linkPostAndContents(postId: PostId, vararg contentIds: ContentId) {
transaction(
db = database
) {
val leftToPut = contentIds.toSet() - getPostContents(postId)
leftToPut.forEach { contentId ->
insert {
it[postIdColumn] = postId
it[contentIdColumn] = contentId
}
}
}
}
fun unlinkPostAndContents(postId: PostId, vararg contentIds: ContentId): Boolean {
return transaction(
db = database
) {
deleteWhere {
postIdColumn.eq(postId).and(contentIdColumn.inList(contentIds.toList()))
} > 0
}
}
}
private val dateTimeFormat = DateFormat("EEE, dd MMM yyyy HH:mm:ss z")
private class PostsRepoDatabaseTable(
private val database: Database,
tableName: String = ""
) : PostsRepo, Table(tableName) {
private val contentsTable = PostsRepoContentRelations(database)
private val idColumn = text("postId")
private val creationDateColumn = text("creationDate").default(
DateTime.now().toString(dateTimeFormat)
)
private val postCreatedBroadcastChannel = BroadcastChannel<RegisteredPost>(Channel.BUFFERED)
override val postCreatedFlow: Flow<RegisteredPost> = postCreatedBroadcastChannel.asFlow()
private val postDeletedBroadcastChannel = BroadcastChannel<RegisteredPost>(Channel.BUFFERED)
override val postDeletedFlow: Flow<RegisteredPost> = postDeletedBroadcastChannel.asFlow()
private val postUpdatedBroadcastChannel = BroadcastChannel<RegisteredPost>(Channel.BUFFERED)
override val postUpdatedFlow: Flow<RegisteredPost> = postUpdatedBroadcastChannel.asFlow()
init {
transaction (db = database) {
SchemaUtils.createMissingTablesAndColumns(this@PostsRepoDatabaseTable)
}
}
private fun ResultRow.toRegisteredPost(): RegisteredPost = get(idColumn).let { id ->
SimpleRegisteredPost(
id,
contentsTable.getPostContents(id),
dateTimeFormat.parse(get(creationDateColumn)).local.unixMillis
)
}
override suspend fun createPost(post: Post): RegisteredPost? {
val id = (post as? RegisteredPost) ?.let { it.id } ?: generatePostId()
return transaction(
db = database
) {
insert {
it[idColumn] = id
}
contentsTable.linkPostAndContents(id, *post.content.toTypedArray())
select { idColumn.eq(id) }.firstOrNull() ?.toRegisteredPost()
} ?.also {
postCreatedBroadcastChannel.send(it)
}
}
override suspend fun deletePost(id: PostId): Boolean {
val post = getPostById(id) ?: return false
return (transaction(
db = database
) {
deleteWhere { idColumn.eq(id) }
} > 0).also {
if (it) {
postDeletedBroadcastChannel.send(post)
contentsTable.unlinkPostAndContents(id, *post.content.toTypedArray())
}
}
}
override suspend fun updatePostContent(postId: PostId, post: Post): Boolean {
return transaction(
db = database
) {
val alreadyLinked = contentsTable.getPostContents(postId)
val toRemove = alreadyLinked - post.content
val toInsert = post.content - alreadyLinked
val updated = (toRemove.isNotEmpty() && contentsTable.unlinkPostAndContents(postId, *toRemove.toTypedArray())) || toInsert.isNotEmpty()
if (toInsert.isNotEmpty()) {
contentsTable.linkPostAndContents(postId, *toInsert.toTypedArray())
}
updated
}.also {
if (it) {
getPostById(postId) ?.also { updatedPost -> postUpdatedBroadcastChannel.send(updatedPost) }
}
}
}
override suspend fun getPostsIds(): Set<PostId> {
return transaction(
db = database
) {
selectAll().map { it[idColumn] }.toSet()
}
}
override suspend fun getPostById(id: PostId): RegisteredPost? {
return transaction(
db = database
) {
select { idColumn.eq(id) }.firstOrNull() ?.toRegisteredPost()
}
}
override suspend fun getPostsByContent(id: ContentId): List<RegisteredPost> {
return transaction(
db = database
) {
val postsIds = contentsTable.getContentPosts(id)
select { idColumn.inList(postsIds) }.map { it.toRegisteredPost() }
}
}
override suspend fun getPostsByCreatingDates(
from: DateTime,
to: DateTime,
pagination: Pagination
): PaginationResult<RegisteredPost> {
return transaction(
db = database
) {
select { creationDateColumn.between(from, to) }.paginate(
pagination
).map {
it.toRegisteredPost()
}.createPaginationResult(
pagination,
selectAll().count()
)
}
}
override suspend fun getPostsByPagination(pagination: Pagination): PaginationResult<RegisteredPost> {
return transaction(
db = database
) {
val posts = selectAll().paginate(pagination).orderBy(creationDateColumn).map {
it.toRegisteredPost()
}
val postsNumber = selectAll().count()
posts.createPaginationResult(pagination, postsNumber)
}
}
}
class ExposedPostsRepo (
database: Database,
tableName: String = ""
) : PostsRepo by PostsRepoDatabaseTable(database, tableName)