feat: Data Submission Worker Integration.

CHANGELOG.md

@@ -5,6 +5,7 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Added
+- creation of data submission jobs.
 
 ## Changed
 

crates/orchestrator/src/main.rs

@@ -2,6 +2,7 @@ use dotenvy::dotenv;
 use orchestrator::config::config;
 use orchestrator::queue::init_consumers;
 use orchestrator::routes::app_router;
+use orchestrator::workers::data_submission::DataSubmissionWorker;
 use orchestrator::workers::proof_registration::ProofRegistrationWorker;
 use orchestrator::workers::proving::ProvingWorker;
 use orchestrator::workers::snos::SnosWorker;
@@ -33,6 +34,7 @@ async fn main() {
     tokio::spawn(start_cron(Box::new(ProvingWorker), 60));
     tokio::spawn(start_cron(Box::new(ProofRegistrationWorker), 60));
     tokio::spawn(start_cron(Box::new(UpdateStateWorker), 60));
+    tokio::spawn(start_cron(Box::new(DataSubmissionWorker), 60));
 
     tracing::info!("Listening on http://{}", address);
     axum::serve(listener, app).await.expect("Failed to start axum server");

crates/orchestrator/src/workers/data_submission.rs

@@ -0,0 +1,55 @@
+use crate::config::config;
+use crate::jobs::create_job;
+use crate::jobs::types::JobType;
+use crate::workers::Worker;
+use async_trait::async_trait;
+use std::collections::HashMap;
+use std::error::Error;
+
+pub struct DataSubmissionWorker;
+
+#[async_trait]
+impl Worker for DataSubmissionWorker {
+    // 0. All ids are assumed to be block numbers.
+    // 1. Fetch the latest completed Proving job.
+    // 2. Fetch the latest DA job creation.
+    // 3. Create jobs from after the lastest DA job already created till latest completed proving job.
+    async fn run_worker(&self) -> Result<(), Box<dyn Error>> {
+        let config = config().await;
+
+        // provides latest completed proof creation job id
+        let latest_proven_job_id = config
+            .database()
+            .get_last_successful_job_by_type(JobType::ProofCreation)
+            .await
+            .unwrap()
+            .map(|item| item.internal_id)
+            .unwrap_or("0".to_string());
+
+        // provides latest triggered data submission job id
+        let latest_data_submission_job_id = config
+            .database()
+            .get_latest_job_by_type_and_internal_id(JobType::DataSubmission)
+            .await
+            .unwrap()
+            .map(|item| item.internal_id)
+            .unwrap_or("0".to_string());
+
+        let latest_data_submission_id: u64 = latest_data_submission_job_id.parse()?;
+        let latest_proven_id: u64 = latest_proven_job_id.parse()?;
+
+        let block_diff = latest_proven_id - latest_data_submission_id;
+
+        // if all blocks are processed
+        if block_diff == 0 {
+            return Ok(());
+        }
+
+        // creating data submission jobs for latest blocks without pre-running data submission jobs jobs don't yet exist.
+        for x in latest_data_submission_id + 1..latest_proven_id + 1 {
+            create_job(JobType::DataSubmission, x.to_string(), HashMap::new()).await?;
+        }
+
+        Ok(())
+    }
+}

crates/orchestrator/src/workers/mod.rs

@@ -2,6 +2,7 @@ use std::error::Error;
 
 use async_trait::async_trait;
 
+pub mod data_submission;
 pub mod proof_registration;
 pub mod proving;
 pub mod snos;