refactor: move state_json from job_execution -> jobs

cala-ledger-outbox-client/src/config.rs

@@ -4,3 +4,9 @@ use serde::{Deserialize, Serialize};
 pub struct CalaLedgerOutboxClientConfig {
     pub url: String,
 }
+
+impl CalaLedgerOutboxClientConfig {
+    pub fn new(url: String) -> Self {
+        Self { url }
+    }
+}

cala-server/migrations/20231211151809_cala_server_setup.sql

@@ -3,7 +3,7 @@ CREATE TABLE jobs (
   name VARCHAR NOT NULL,
   type VARCHAR NOT NULL,
   description VARCHAR,
-  data JSONB,
+  state_json JSONB,
   last_error VARCHAR,
   completed_at TIMESTAMPTZ,
   modified_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
@@ -18,9 +18,8 @@ CREATE TABLE job_executions (
   next_attempt INT NOT NULL DEFAULT 1,
   name VARCHAR NOT NULL,
   state JobExecutionState NOT NULL DEFAULT 'pending',
-  state_json JSONB,
-  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-  reschedule_after TIMESTAMPTZ NOT NULL DEFAULT NOW()
+  reschedule_after TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
 );
 
 CREATE TABLE integrations (

cala-server/src/extension/cala_outbox_import/config.rs

@@ -1,15 +0,0 @@
-use cala_ledger_outbox_client::CalaLedgerOutboxClientConfig;
-use serde::{Deserialize, Serialize};
-
-#[derive(Clone, Debug, Serialize, Deserialize)]
-pub struct CalaOutboxImportConfig {
-    pub endpoint: String,
-}
-
-impl From<&CalaOutboxImportConfig> for CalaLedgerOutboxClientConfig {
-    fn from(config: &CalaOutboxImportConfig) -> Self {
-        Self {
-            url: config.endpoint.clone(),
-        }
-    }
-}

cala-server/src/extension/cala_outbox_import/job.rs

@@ -12,8 +12,6 @@ use tracing::instrument;
 
 use crate::job::*;
 
-pub use super::config::*;
-
 pub const CALA_OUTBOX_IMPORT_JOB_TYPE: JobType = JobType::new("cala-outbox-import-job");
 
 #[derive(Default)]
@@ -26,24 +24,23 @@ impl JobInitializer for CalaOutboxImportJobInitializer {
 
     fn init(
         &self,
-        job: Job,
+        _job: Job,
         ledger: &CalaLedger,
     ) -> Result<Box<dyn JobRunner>, Box<dyn std::error::Error>> {
         Ok(Box::new(CalaOutboxImportJob {
-            config: job.data()?,
             ledger: ledger.clone(),
         }))
     }
 }
 
 pub struct CalaOutboxImportJob {
-    config: CalaOutboxImportConfig,
     ledger: CalaLedger,
 }
 
 #[derive(Default, Serialize, Deserialize)]
-struct CalaOutboxImportJobState {
-    last_synced: EventSequence,
+pub struct CalaOutboxImportJobState {
+    pub endpoint: String,
+    pub last_synced: EventSequence,
 }
 
 #[async_trait]
@@ -53,21 +50,21 @@ impl JobRunner for CalaOutboxImportJob {
         &mut self,
         mut current_job: CurrentJob,
     ) -> Result<JobCompletion, Box<dyn std::error::Error>> {
+        let mut state = current_job
+            .state::<CalaOutboxImportJobState>()?
+            .expect("Job state");
         println!(
             "Executing CalaOutboxImportJob importing from endpoint: {}",
-            self.config.endpoint
+            state.endpoint
         );
-        let mut client = Client::connect(ClientConfig::from(&self.config)).await?;
-        let mut state = current_job
-            .state::<CalaOutboxImportJobState>()?
-            .unwrap_or_default();
+        let mut client = Client::connect(ClientConfig::new(state.endpoint.clone())).await?;
         let mut stream = client.subscribe(Some(state.last_synced)).await?;
         loop {
             match stream.next().await {
                 Some(Ok(message)) => {
                     let mut tx = current_job.pool().begin().await?;
                     state.last_synced = message.sequence;
-                    current_job.update_execution_state(&mut tx, &state).await?;
+                    current_job.update_state(&mut tx, &state).await?;
                     self.ledger
                         .sync_outbox_event(
                             tx,

cala-server/src/extension/cala_outbox_import/mod.rs

@@ -1,4 +1,3 @@
-mod config;
 mod job;
 mod mutation;
 

cala-server/src/extension/cala_outbox_import/mutation.rs

@@ -41,7 +41,7 @@ impl Mutation {
                 input.job_id,
                 input.name.clone(),
                 input.description.clone(),
-                CalaOutboxImportConfig::from(input),
+                CalaOutboxImportJobState::from(input),
             )
             .await?;
         Ok(CalaOutboxImportJobCreatePayload {
@@ -50,10 +50,11 @@ impl Mutation {
     }
 }
 
-impl From<CalaOutboxImportJobCreateInput> for CalaOutboxImportConfig {
+impl From<CalaOutboxImportJobCreateInput> for CalaOutboxImportJobState {
     fn from(input: CalaOutboxImportJobCreateInput) -> Self {
-        CalaOutboxImportConfig {
+        Self {
             endpoint: input.endpoint,
+            ..Default::default()
         }
     }
 }

cala-server/src/job/current.rs

@@ -38,15 +38,15 @@ impl CurrentJob {
         }
     }
 
-    pub async fn update_execution_state<T: Serialize>(
+    pub async fn update_state<T: Serialize>(
         &mut self,
         db: &mut Transaction<'_, Postgres>,
         state: T,
     ) -> Result<(), JobError> {
         let state_json = serde_json::to_value(state).map_err(JobError::CouldNotSerializeState)?;
         sqlx::query!(
             r#"
-          UPDATE job_executions
+          UPDATE jobs
           SET state_json = $1
           WHERE id = $2
         "#,

cala-server/src/job/entity.rs

@@ -32,7 +32,7 @@ pub struct Job {
     pub job_type: JobType,
     pub description: Option<String>,
     pub last_error: Option<String>,
-    data: serde_json::Value,
+    state: serde_json::Value,
     pub completed_at: Option<DateTime<Utc>>,
 }
 
@@ -41,21 +41,21 @@ impl Job {
         name: String,
         job_type: JobType,
         description: Option<String>,
-        data: T,
+        initial_state: T,
     ) -> Self {
         Self {
             id: JobId::new(),
             name,
             job_type,
             description,
-            data: serde_json::to_value(data).expect("could not serialize job data"),
+            state: serde_json::to_value(initial_state).expect("could not serialize job state"),
             last_error: None,
             completed_at: None,
         }
     }
 
-    pub fn data<T: serde::de::DeserializeOwned>(&self) -> Result<T, serde_json::Error> {
-        serde_json::from_value(self.data.clone())
+    pub fn state<T: serde::de::DeserializeOwned>(&self) -> Result<T, serde_json::Error> {
+        serde_json::from_value(self.state.clone())
     }
 
     pub(super) fn success(&mut self) {

cala-server/src/job/executor.rs

@@ -145,18 +145,19 @@ impl JobExecutor {
         let rows = sqlx::query!(
             r#"
               WITH selected_jobs AS (
-                  SELECT id
-                  FROM job_executions
+                  SELECT je.id, state_json
+                  FROM job_executions je
+                  JOIN jobs ON je.id = jobs.id
                   WHERE reschedule_after < NOW()
-                  AND state = 'pending'
+                  AND je.state = 'pending'
                   LIMIT $1
                   FOR UPDATE
               )
               UPDATE job_executions AS je
               SET state = 'running', reschedule_after = NOW() + $2::interval
               FROM selected_jobs
               WHERE je.id = selected_jobs.id
-              RETURNING je.id AS "id!: JobId", je.state_json, je.next_attempt
+              RETURNING je.id AS "id!: JobId", selected_jobs.state_json, je.next_attempt
               "#,
             poll_limit as i32,
             pg_interval

cala-server/src/job/repo.rs

@@ -20,14 +20,14 @@ impl JobRepo {
         job: Job,
     ) -> Result<Job, JobError> {
         sqlx::query!(
-            r#"INSERT INTO jobs (id, type, name, description, data)
+            r#"INSERT INTO jobs (id, type, name, description, state_json)
             VALUES ($1, $2, $3, $4, $5)"#,
             job.id as JobId,
             &job.job_type as &JobType,
             &job.name,
             job.description.as_ref(),
-            job.data::<serde_json::Value>()
-                .expect("Could not serialize data")
+            job.state::<serde_json::Value>()
+                .expect("Could not serialize state")
         )
         .execute(&mut **db)
         .await?;
@@ -56,7 +56,7 @@ impl JobRepo {
 
     pub async fn find_by_id(&self, id: JobId) -> Result<Job, JobError> {
         let row = sqlx::query!(
-            r#"SELECT id as "id: JobId", type AS job_type, name, description, data, completed_at, last_error
+            r#"SELECT id as "id: JobId", type AS job_type, name, description, state_json, completed_at, last_error
             FROM jobs
             WHERE id = $1"#,
             id as JobId
@@ -67,7 +67,7 @@ impl JobRepo {
             row.name,
             JobType::from_string(row.job_type),
             row.description,
-            row.data,
+            row.state_json,
         );
         job.id = row.id;
         job.completed_at = row.completed_at;