use async_trait::async_trait;
use chrono::{DateTime, Utc};
use chronon_core::models::{
Job, JobRevision, PartitionAssignment, Run, RunStatus, SchedulerLeader, Script, Worker,
};
use chronon_core::store::SchedulerStore;
use chronon_core::Result;
use crate::{claims, coordinator, jobs, runs, SqlSchedulerStore};
#[async_trait]
impl SchedulerStore for SqlSchedulerStore {
async fn upsert_job(&self, job: &Job) -> Result<()> {
jobs::upsert_job(self, job).await
}
async fn get_job(&self, job_id: &str) -> Result<Option<Job>> {
jobs::get_job(self, job_id).await
}
async fn get_job_by_name(&self, job_name: &str) -> Result<Option<Job>> {
jobs::get_job_by_name(self, job_name).await
}
async fn list_jobs(&self) -> Result<Vec<Job>> {
jobs::list_jobs(self).await
}
async fn list_due_jobs(&self, before: DateTime<Utc>) -> Result<Vec<Job>> {
jobs::list_due_jobs(self, before).await
}
async fn pause_job(&self, job_id: &str) -> Result<()> {
jobs::pause_job(self, job_id).await
}
async fn resume_job(&self, job_id: &str) -> Result<()> {
jobs::resume_job(self, job_id).await
}
async fn create_run(&self, run: &Run) -> Result<()> {
runs::create_run(self, run).await
}
async fn update_run(&self, run: &Run) -> Result<()> {
runs::update_run(self, run).await
}
async fn get_run(&self, run_id: &str) -> Result<Option<Run>> {
runs::get_run(self, run_id).await
}
async fn list_runs_for_job(&self, job_id: &str, limit: usize) -> Result<Vec<Run>> {
runs::list_runs_for_job(self, job_id, limit).await
}
async fn list_runs_filtered(
&self,
job_id: Option<&str>,
status: Option<RunStatus>,
offset: usize,
limit: usize,
) -> Result<Vec<Run>> {
runs::list_runs_filtered(self, job_id, status, offset, limit).await
}
async fn claim_next_queued(
&self,
pool_id: &str,
worker_id: &str,
now: DateTime<Utc>,
lease_ttl_secs: i64,
) -> Result<Option<Run>> {
runs::claim_next_queued(self, pool_id, worker_id, now, lease_ttl_secs).await
}
async fn claim_run_by_id(
&self,
run_id: &str,
pool_id: &str,
worker_id: &str,
now: DateTime<Utc>,
lease_ttl_secs: i64,
) -> Result<Option<Run>> {
runs::claim_run_by_id(self, run_id, pool_id, worker_id, now, lease_ttl_secs).await
}
async fn claim_runs_by_ids(
&self,
run_ids: &[&str],
pool_id: &str,
worker_id: &str,
now: DateTime<Utc>,
lease_ttl_secs: i64,
) -> Result<Vec<Run>> {
runs::claim_runs_by_ids(self, run_ids, pool_id, worker_id, now, lease_ttl_secs).await
}
async fn renew_run_lease(
&self,
run_id: &str,
worker_id: &str,
now: DateTime<Utc>,
lease_ttl_secs: i64,
) -> Result<bool> {
runs::renew_run_lease(self, run_id, worker_id, now, lease_ttl_secs).await
}
async fn reclaim_expired_run_leases(&self, now: DateTime<Utc>) -> Result<Vec<String>> {
runs::reclaim_expired_run_leases(self, now).await
}
async fn append_revision(&self, revision: &JobRevision) -> Result<()> {
coordinator::append_revision(self, revision).await
}
async fn list_revisions(&self, job_id: &str) -> Result<Vec<JobRevision>> {
coordinator::list_revisions(self, job_id).await
}
async fn upsert_script(&self, script: &Script) -> Result<()> {
coordinator::upsert_script(self, script).await
}
async fn get_script(&self, script_name: &str) -> Result<Option<Script>> {
coordinator::get_script(self, script_name).await
}
async fn try_claim_run_once(
&self,
job_id: &str,
claimed_by: &str,
now: DateTime<Utc>,
claim_ttl_secs: i64,
) -> Result<bool> {
claims::try_claim_run_once(self, job_id, claimed_by, now, claim_ttl_secs).await
}
async fn mark_run_once_completed(
&self,
job_id: &str,
completed_at: DateTime<Utc>,
) -> Result<()> {
claims::mark_run_once_completed(self, job_id, completed_at).await
}
async fn release_run_once_claim(
&self,
job_id: &str,
claimed_by: &str,
now: DateTime<Utc>,
) -> Result<()> {
claims::release_run_once_claim(self, job_id, claimed_by, now).await
}
async fn find_due_job_ids_in_partitions(
&self,
owned_partitions: &[u32],
due_until: DateTime<Utc>,
limit: u32,
) -> Result<Vec<String>> {
claims::find_due_job_ids_in_partitions(self, owned_partitions, due_until, limit).await
}
async fn min_next_run_at_in_partitions(
&self,
owned_partitions: &[u32],
) -> Result<Option<DateTime<Utc>>> {
claims::min_next_run_at_in_partitions(self, owned_partitions).await
}
async fn claim_job_for_tick(
&self,
job_id: &str,
claim_id: &str,
now: DateTime<Utc>,
lease_ttl_secs: i64,
) -> Result<bool> {
claims::claim_job_for_tick(self, job_id, claim_id, now, lease_ttl_secs).await
}
async fn release_job_tick_claim(&self, job_id: &str) -> Result<()> {
claims::release_job_tick_claim(self, job_id).await
}
async fn persist_post_tick_job_state(
&self,
job_id: &str,
next_run_at: Option<DateTime<Utc>>,
) -> Result<()> {
claims::persist_post_tick_job_state(self, job_id, next_run_at).await
}
async fn try_acquire_leader(&self, instance_id: &str, ttl_secs: i64) -> Result<bool> {
coordinator::try_acquire_leader(self, instance_id, ttl_secs).await
}
async fn renew_leader_lease(&self, instance_id: &str, ttl_secs: i64) -> Result<()> {
coordinator::renew_leader_lease(self, instance_id, ttl_secs).await
}
async fn get_leader(&self) -> Result<Option<SchedulerLeader>> {
coordinator::get_leader(self).await
}
async fn upsert_partition_assignment(&self, assignment: &PartitionAssignment) -> Result<()> {
coordinator::upsert_partition_assignment(self, assignment).await
}
async fn list_partition_assignments(&self) -> Result<Vec<PartitionAssignment>> {
coordinator::list_partition_assignments(self).await
}
async fn register_worker(&self, worker: &Worker) -> Result<()> {
coordinator::register_worker(self, worker).await
}
async fn heartbeat_worker(&self, worker_id: &str, at: DateTime<Utc>) -> Result<()> {
coordinator::heartbeat_worker(self, worker_id, at).await
}
}