Add Quartz Scheduler (#7)

sevenreup · web-flow · commit fdc1a1f2622a · 2025-01-24T10:00:42.000+02:00
* init scheduler

* clean repo before pull

* Fix collision
diff --git a/.idea/misc.xml b/.idea/misc.xml
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
@@ -24,11 +24,15 @@ kotlinLogging = "5.1.0"
 junit = "5.9.2"
 jsonToolsJsonPatch = "1.13"
 mustache = "0.9.10"
+quartz = "2.5.0"
+exposed-version = "0.58.0"
+c3p0 = "0.10.1"
 
 [libraries]
 ktor-server-core = { module = "io.ktor:ktor-server-core-jvm", version.ref = "ktor" }
 ktor-server-netty = { module = "io.ktor:ktor-server-netty-jvm", version.ref = "ktor" }
 ktor-server-webjars = { module = "io.ktor:ktor-server-webjars-jvm", version.ref = "ktor" }
+ktor-server-cors = { module = "io.ktor:ktor-server-cors", version.ref = "ktor" }
 jquery = { module = "org.webjars:jquery", version.ref = "jquery" }
 ktor-swagger-ui = { module = "io.github.smiley4:ktor-swagger-ui", version.ref = "swagger-ui" }
 ktor-server-auth = { module = "io.ktor:ktor-server-auth-jvm", version.ref = "ktor" }
@@ -79,6 +83,14 @@ gson = { group = "com.google.code.gson", name = "gson", version.ref = "gson" }
 kaml = { group = "com.charleskorn.kaml", name = "kaml", version.ref = "kaml" }
 hamcrest-core = { group = "org.hamcrest", name = "hamcrest-core", version.ref = "hamcrest" }
 
+quartz = { group = "org.quartz-scheduler", name = "quartz", version.ref = "quartz" }
+c3p0 = { group = "com.mchange", name = "c3p0", version.ref = "c3p0" }
+
+exposed-core = { module = "org.jetbrains.exposed:exposed-core", version.ref = "exposed-version" }
+exposed-jdbc = { module = "org.jetbrains.exposed:exposed-jdbc", version.ref = "exposed-version" }
+exposed-dao = { module = "org.jetbrains.exposed:exposed-dao", version.ref = "exposed-version" }
+exposed-java-time = { module = "org.jetbrains.exposed:exposed-java-time", version.ref = "exposed-version" }
+
 # Testing dependencies
 junit-jupiter-api = { group = "org.junit.jupiter", name = "junit-jupiter-api", version.ref = "junit" }
 junit-jupiter-engine = { group = "org.junit.jupiter", name = "junit-jupiter-engine", version.ref = "junit" }
diff --git a/server/build.gradle.kts b/server/build.gradle.kts
@@ -37,8 +37,9 @@ dependencies {
 
     implementation(libs.ktor.server.core)
     implementation(libs.ktor.server.netty)
-
+    implementation(libs.ktor.server.cors)
     implementation(libs.ktor.server.webjars)
+
     implementation(libs.jquery)
     implementation(libs.ktor.swagger.ui)
     implementation(libs.ktor.server.auth)
@@ -52,7 +53,6 @@ dependencies {
     implementation(libs.koin.logger.slf4j)
     implementation(libs.ktor.serialization.json)
     implementation(libs.ktor.server.content.negotiation)
-    implementation(libs.postgresql)
     implementation(libs.h2)
 
     implementation(libs.logback.classic)
@@ -66,6 +66,15 @@ dependencies {
 
     implementation(libs.spullara.mustache)
 
+    implementation(libs.quartz)
+    implementation(libs.c3p0)
+
+    implementation(libs.exposed.core)
+    implementation(libs.exposed.jdbc)
+    implementation(libs.exposed.dao)
+    implementation(libs.exposed.java.time)
+    implementation(libs.postgresql)
+
     testImplementation(libs.ktor.server.test.host)
     testImplementation(libs.kotlin.test.junit)
 
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/Application.kt b/server/src/main/kotlin/org/dtree/fhir/server/Application.kt
@@ -3,6 +3,7 @@ package org.dtree.fhir.server
 import io.ktor.server.application.*
 import org.dtree.fhir.server.plugins.configureFrameworks
 import org.dtree.fhir.server.plugins.configureRouting
+import org.dtree.fhir.server.plugins.configureScheduler
 import org.dtree.fhir.server.plugins.configureSecurity
 import org.dtree.fhir.server.util.loadEnv
 
@@ -16,4 +17,6 @@ fun Application.module() {
     configureFrameworks(dotEnv)
     configureSecurity()
     configureRouting()
+
+    configureScheduler()
 }
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/data/JobHistory.kt b/server/src/main/kotlin/org/dtree/fhir/server/data/JobHistory.kt
@@ -0,0 +1,17 @@
+package org.dtree.fhir.server.data
+
+import org.jetbrains.exposed.sql.Table
+import org.jetbrains.exposed.sql.javatime.datetime
+
+object JobHistoryTable : Table("job_history") {
+    val id = integer("id").autoIncrement()
+    val jobName = varchar("job_name", 100)
+    val jobGroup = varchar("job_group", 100)
+    val startTime = datetime("start_time")
+    val endTime = datetime("end_time").nullable()
+    val status = varchar("status", 20)
+    val message = text("message").nullable()
+    val retryCount = integer("retry_count").default(0)
+
+    override val primaryKey = PrimaryKey(id)
+}
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/data/JobHistoryRepository.kt b/server/src/main/kotlin/org/dtree/fhir/server/data/JobHistoryRepository.kt
@@ -0,0 +1,54 @@
+package org.dtree.fhir.server.data
+
+import org.jetbrains.exposed.sql.*
+import org.jetbrains.exposed.sql.transactions.transaction
+import java.time.LocalDateTime
+
+class JobHistoryRepository(database: Database) {
+
+    init {
+        transaction(database) {
+            SchemaUtils.create(JobHistoryTable)
+        }
+    }
+
+    fun logJobStart(jobName: String, jobGroup: String): Int = transaction {
+        JobHistoryTable.insert {
+            it[this.jobName] = jobName
+            it[this.jobGroup] = jobGroup
+            it[startTime] = LocalDateTime.now()
+            it[status] = "RUNNING"
+        } get JobHistoryTable.id
+    }
+
+    fun logJobEnd(historyId: Int, status: String, message: String? = null) = transaction {
+        JobHistoryTable.update({ JobHistoryTable.id eq historyId }) {
+            it[endTime] = LocalDateTime.now()
+            it[this.status] = status
+            it[this.message] = message
+        }
+    }
+
+    fun updateRetryCount(historyId: Int, retryCount: Int) = transaction {
+        JobHistoryTable.update({ JobHistoryTable.id eq historyId }) {
+            it[this.retryCount] = retryCount
+        }
+    }
+
+    fun getJobHistory(jobName: String, jobGroup: String, limit: Int = 100) = transaction {
+        JobHistoryTable.selectAll()
+            .where { (JobHistoryTable.jobName eq jobName) and (JobHistoryTable.jobGroup eq jobGroup) }
+            .orderBy(JobHistoryTable.startTime to SortOrder.DESC)
+            .limit(limit)
+            .map {
+                mapOf(
+                    "id" to it[JobHistoryTable.id],
+                    "startTime" to it[JobHistoryTable.startTime],
+                    "endTime" to it[JobHistoryTable.endTime],
+                    "status" to it[JobHistoryTable.status],
+                    "message" to it[JobHistoryTable.message],
+                    "retryCount" to it[JobHistoryTable.retryCount]
+                )
+            }
+    }
+}
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/plugins/Frameworks.kt b/server/src/main/kotlin/org/dtree/fhir/server/plugins/Frameworks.kt
@@ -5,10 +5,15 @@ import io.ktor.serialization.gson.*
 import io.ktor.server.application.*
 import io.ktor.server.plugins.contentnegotiation.*
 import org.dtree.fhir.core.di.FhirProvider
+import org.dtree.fhir.server.data.JobHistoryRepository
 import org.dtree.fhir.server.plugins.injection.ModulesInjection
+import org.dtree.fhir.server.plugins.scheduler.JobFactory
+import org.dtree.fhir.server.plugins.scheduler.JobSchedulerManager
 import org.dtree.fhir.server.services.injection.ServicesInjection
+import org.dtree.fhir.server.services.tracing.TracingService
 import org.dtree.fhir.server.util.LocalDateAdapter
 import org.dtree.fhir.server.util.LocalDateTimeTypeAdapter
+import org.jetbrains.exposed.sql.Database
 import org.koin.core.logger.Level
 import org.koin.dsl.module
 import org.koin.ktor.plugin.Koin
@@ -19,7 +24,7 @@ import java.time.LocalDateTime
 fun Application.configureFrameworks(dotEnv: Dotenv) {
     install(ContentNegotiation) {
         gson {
-            registerTypeAdapter(LocalDate::class.java,LocalDateAdapter())
+            registerTypeAdapter(LocalDate::class.java, LocalDateAdapter())
             registerTypeAdapter(LocalDateTime::class.java, LocalDateTimeTypeAdapter())
         }
     }
@@ -28,7 +33,20 @@ fun Application.configureFrameworks(dotEnv: Dotenv) {
         modules(module {
             single { dotEnv }
             single { FhirProvider() }
-        }, ModulesInjection.koinBeans, ServicesInjection.koinBeans)
+            single<Database>(createdAtStart = true) {
+                Database.connect(
+                    url = dotEnv["DB_URL"] ?: "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1",
+                    user = dotEnv["DB_USER"] ?: "sa",
+                    driver = dotEnv["DB_DRIVER"] ?: "org.h2.Driver",
+                    password = dotEnv["DB_PASSWORD"] ?: "",
+                )
+            }
+        }, ModulesInjection.koinBeans, ServicesInjection.koinBeans, module {
+            single<JobHistoryRepository> {
+                JobHistoryRepository(this.get<Database>())
+            }
+            single { JobSchedulerManager(dotEnv, JobFactory(this.get<TracingService>(), this.get())) }
+        })
 
     }
 }
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/plugins/Routing.kt b/server/src/main/kotlin/org/dtree/fhir/server/plugins/Routing.kt
@@ -14,6 +14,7 @@ import io.ktor.server.routing.get
 import org.dtree.fhir.server.plugins.appointment.appointmentModule
 import org.dtree.fhir.server.plugins.careplan.carePlanModule
 import org.dtree.fhir.server.plugins.patient.patientModule
+import org.dtree.fhir.server.plugins.scheduler.schedulerRoutes
 import org.dtree.fhir.server.plugins.stats.statsModule
 import org.dtree.fhir.server.plugins.tasks.tasksModule
 import org.dtree.fhir.server.plugins.tracing.tracingModule
@@ -51,6 +52,7 @@ fun Application.configureRouting() {
         carePlanModule()
         tasksModule()
         patientModule()
+        schedulerRoutes()
 
         get("/") {
             call.respondText("Hello, World!")
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/plugins/configureScheduler.kt b/server/src/main/kotlin/org/dtree/fhir/server/plugins/configureScheduler.kt
@@ -0,0 +1,11 @@
+package org.dtree.fhir.server.plugins
+
+import io.ktor.server.application.*
+import org.dtree.fhir.server.plugins.scheduler.JobSchedulerManager
+import org.koin.ktor.ext.inject
+
+fun Application.configureScheduler() {
+    val jobSchedulerManager by inject<JobSchedulerManager>()
+
+    jobSchedulerManager.startScheduler()
+}
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/plugins/injection/ModulesInjection.kt b/server/src/main/kotlin/org/dtree/fhir/server/plugins/injection/ModulesInjection.kt
@@ -12,12 +12,8 @@ import org.dtree.fhir.server.controller.*
 import org.dtree.fhir.server.services.form.LocalResourceFetcher
 import org.dtree.fhir.server.services.form.ResourceFetcher
 import org.dtree.fhir.server.services.form.ResponseGenerator
-import org.koin.core.module.dsl.singleOf
-import org.dtree.fhir.server.controller.StatsController
-import org.dtree.fhir.server.controller.StatsControllerImpl
-import org.dtree.fhir.server.controller.TracingController
-import org.dtree.fhir.server.controller.TracingControllerImpl
 import org.koin.core.module.dsl.bind
+import org.koin.core.module.dsl.singleOf
 import org.koin.dsl.module
 
 object ModulesInjection {
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/plugins/scheduler/JobFactory.kt b/server/src/main/kotlin/org/dtree/fhir/server/plugins/scheduler/JobFactory.kt
@@ -0,0 +1,22 @@
+package org.dtree.fhir.server.plugins.scheduler
+
+import org.dtree.fhir.server.data.JobHistoryRepository
+import org.dtree.fhir.server.plugins.scheduler.jobs.ClearTracingJob
+import org.dtree.fhir.server.services.tracing.TracingService
+import org.quartz.Job
+import org.quartz.Scheduler
+import org.quartz.spi.JobFactory
+import org.quartz.spi.TriggerFiredBundle
+import kotlin.reflect.jvm.jvmName
+
+class JobFactory(private val tracingService: TracingService, private val historyService: JobHistoryRepository) : JobFactory {
+    override fun newJob(bundle: TriggerFiredBundle?, scheduler: Scheduler?): Job {
+        if (bundle != null) {
+            val jobClass = bundle.jobDetail.jobClass
+            if (jobClass.name == ClearTracingJob::class.jvmName) {
+                return ClearTracingJob(tracingService, historyService)
+            }
+        }
+        throw NotImplementedError("Job Factory error")
+    }
+}
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/plugins/scheduler/JobSchedulerManager.kt b/server/src/main/kotlin/org/dtree/fhir/server/plugins/scheduler/JobSchedulerManager.kt
@@ -0,0 +1,48 @@
+package org.dtree.fhir.server.plugins.scheduler
+
+import io.github.cdimascio.dotenv.Dotenv
+import org.dtree.fhir.server.plugins.scheduler.jobs.ClearTracingJob
+import org.quartz.*
+import org.quartz.impl.StdSchedulerFactory
+import java.util.*
+
+class JobSchedulerManager(dotenv: Dotenv, jobFactory: JobFactory) {
+    var scheduler: Scheduler
+
+    init {
+        val props = Properties()
+
+        for (entry in dotenv.entries()) {
+            if (entry.key.startsWith("org.quartz")) {
+                props[entry.key] = entry.value
+            }
+        }
+
+        val schedulerFactory: SchedulerFactory = StdSchedulerFactory(props)
+        scheduler = schedulerFactory.scheduler
+
+        scheduler.setJobFactory(jobFactory)
+
+        scheduleJobs()
+    }
+
+    fun startScheduler() {
+        scheduler.start()
+    }
+
+    private fun scheduleJobs() {
+        val jobKey = JobKey("clear-tracing-list", "jobs")
+        if (!scheduler.checkExists(jobKey)) {
+            val job = JobBuilder.newJob(ClearTracingJob::class.java)
+                .withIdentity(jobKey.name, jobKey.group)
+                .build()
+
+            val trigger = TriggerBuilder.newTrigger()
+                .withIdentity("clear-tracing-list-trigger", "jobs")
+                .withSchedule(CronScheduleBuilder.cronSchedule("0 0 7,12,17 * * ?"))
+                .build()
+
+            scheduler.scheduleJob(job, trigger)
+        }
+    }
+}
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/plugins/scheduler/endpoints.kt b/server/src/main/kotlin/org/dtree/fhir/server/plugins/scheduler/endpoints.kt
@@ -0,0 +1,58 @@
+package org.dtree.fhir.server.plugins.scheduler
+
+import io.ktor.server.response.*
+import io.ktor.server.routing.*
+import io.ktor.server.util.*
+import io.ktor.utils.io.*
+import org.koin.ktor.ext.inject
+import org.quartz.impl.matchers.GroupMatcher
+import java.time.format.DateTimeFormatter
+
+@OptIn(InternalAPI::class)
+fun Route.schedulerRoutes() {
+    val schedulerManager by inject<JobSchedulerManager>()
+
+    get("/jobs/active") {
+        val activeJobs = schedulerManager.scheduler.jobGroupNames.flatMap { groupName ->
+            schedulerManager.scheduler.getJobKeys(GroupMatcher.groupEquals(groupName)).map { jobKey ->
+                val jobDetail = schedulerManager.scheduler.getJobDetail(jobKey)
+                val triggers = schedulerManager.scheduler.getTriggersOfJob(jobKey)
+                mapOf(
+                    "jobName" to jobKey.name,
+                    "jobGroup" to jobKey.group,
+                    "description" to jobDetail.description,
+                    "triggerCount" to triggers.size
+                )
+            }
+        }
+        call.respond(activeJobs)
+    }
+
+    get("/jobs/upcoming") {
+        val upcomingJobs = schedulerManager.scheduler.jobGroupNames.flatMap { groupName ->
+            schedulerManager.scheduler.getJobKeys(GroupMatcher.groupEquals(groupName)).flatMap { jobKey ->
+                schedulerManager.scheduler.getTriggersOfJob(jobKey).map { trigger ->
+                    mapOf(
+                        "jobName" to jobKey.name,
+                        "jobGroup" to jobKey.group,
+                        "nextFireTime" to trigger.nextFireTime?.toZonedDateTime()?.format(DateTimeFormatter.ISO_INSTANT),
+                        "previousFireTime" to trigger.previousFireTime?.toZonedDateTime()?.format(DateTimeFormatter.ISO_INSTANT)
+                    )
+                }
+            }
+        }
+        call.respond(upcomingJobs)
+    }
+
+    get("/jobs/running") {
+        val runningJobs = schedulerManager.scheduler.currentlyExecutingJobs.map { executingJob ->
+            mapOf(
+                "jobName" to executingJob.jobDetail.key.name,
+                "jobGroup" to executingJob.jobDetail.key.group,
+                "fireTime" to executingJob.fireTime.toString(),
+                "scheduledFireTime" to executingJob.scheduledFireTime.toString()
+            )
+        }
+        call.respond(runningJobs)
+    }
+}
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/plugins/scheduler/jobs/ClearTracingJob.kt b/server/src/main/kotlin/org/dtree/fhir/server/plugins/scheduler/jobs/ClearTracingJob.kt
diff --git a/server/src/main/kotlin/org/dtree/fhir/server/services/form/ResourceFetcher.kt b/server/src/main/kotlin/org/dtree/fhir/server/services/form/ResourceFetcher.kt
